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(())
     }

datafusion/sqllogictest/test_files/aggregate.slt

@@ -1280,6 +1280,10 @@ NULL NULL 781 7.81 125 -117 100
 # ----
 # [4, 2, 3, 5, 1]
 
+# additional count(1) forces array_agg_distinct instead of array_agg over aggregated by c2 data
+statement ok
+SELECT array_agg(distinct c2), count(1) FROM aggregate_test_100
+
 # aggregate_time_min_and_max
 query TT
 select min(t), max(t) from  (select '00:00:00' as t union select '00:00:01' union select '00:00:02')