fix: inconsistent types in array_agg_distinct merge_batch

datafusion/physical-expr/src/aggregate/array_agg_distinct.rs

@@ -28,8 +28,7 @@ use std::collections::HashSet;
 use crate::aggregate::utils::down_cast_any_ref;
 use crate::expressions::format_state_name;
 use crate::{AggregateExpr, PhysicalExpr};
-use datafusion_common::Result;
-use datafusion_common::ScalarValue;
+use datafusion_common::{internal_err, DataFusionError, Result, ScalarValue};
 use datafusion_expr::Accumulator;
 
 /// Expression for a ARRAY_AGG(DISTINCT) aggregation.
@@ -147,11 +146,20 @@ impl Accumulator for DistinctArrayAggAccumulator {
             return Ok(());
         }
 
-        for array in states {
-            for j in 0..array.len() {
-                self.values.insert(ScalarValue::try_from_array(array, j)?);
+        assert!(
+            states.len() == 1,
+            "array_agg_distinct states must contain single array"
+        );
+        let state = &states[0];
+        (0..state.len()).try_for_each(|i| {
+            let scalar = ScalarValue::try_from_array(state, i)?;
+            if let ScalarValue::List(Some(values), _) = scalar {
+                self.values.extend(values);
+                Ok(())
+            } else {
+                internal_err!("array_agg_distinct state must be list")
             }
-        }
+        })?;
 
         Ok(())
     }