x

Author: forsaken628

src/query/sql/src/executor/physical_plans/physical_aggregate_partial.rs

@@ -14,11 +14,11 @@
 
 use databend_common_exception::Result;
 use databend_common_expression::types::DataType;
-#[allow(unused_imports)]
-use databend_common_expression::DataBlock;
+use databend_common_expression::AggrStateSerdeType;
 use databend_common_expression::DataField;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::DataSchemaRefExt;
+use databend_common_functions::aggregates::AggregateFunctionFactory;
 
 use super::SortDesc;
 use crate::executor::explain::PlanStatsInfo;
@@ -47,11 +47,31 @@ impl AggregatePartial {
         let input_schema = self.input.output_schema()?;
 
         let mut fields = Vec::with_capacity(self.agg_funcs.len() + self.group_by.len());
+        let factory = AggregateFunctionFactory::instance();
 
-        fields.extend(self.agg_funcs.iter().map(|func| {
-            let name = func.output_column.to_string();
-            DataField::new(&name, DataType::Binary)
-        }));
+        for desc in &self.agg_funcs {
+            let name = desc.output_column.to_string();
+
+            let func = factory
+                .get(
+                    &desc.sig.name,
+                    desc.sig.params.clone(),
+                    desc.sig.args.clone(),
+                    desc.sig.sort_descs.clone(),
+                )
+                .unwrap();
+
+            let tuple = func
+                .serialize_type()
+                .iter()
+                .map(|serde_type| match serde_type {
+                    AggrStateSerdeType::Bool => DataType::Boolean,
+                    AggrStateSerdeType::Binary(_) => DataType::Binary,
+                })
+                .collect();
+
+            fields.push(DataField::new(&name, DataType::Tuple(tuple)))
+        }
 
         for (idx, field) in self.group_by.iter().zip(
             self.group_by