fix: Final aggregation should not bind to the input of partial aggreg…

…ation

core/src/execution/datafusion/planner.rs

@@ -43,7 +43,7 @@ use datafusion_physical_expr::{
     execution_props::ExecutionProps,
     expressions::{
         CaseExpr, CastExpr, Count, FirstValue, InListExpr, IsNullExpr, LastValue, Max, Min,
-        NegativeExpr, NotExpr, Sum,
+        NegativeExpr, NotExpr, Sum, UnKnownColumn,
     },
     AggregateExpr, ScalarFunctionExpr,
 };
@@ -201,9 +201,16 @@ impl PhysicalPlanner {
             }
             ExprStruct::Bound(bound) => {
                 let idx = bound.index as usize;
-                let column_name = format!("col_{}", idx);
-                Ok(Arc::new(Column::new(&column_name, idx)))
+                if idx >= input_schema.fields().len() {
+                    return Err(ExecutionError::GeneralError(format!(
+                        "Column index {} is out of bound. Schema: {}",
+                        idx, input_schema
+                    )));
+                }
+                let field = input_schema.field(idx);
+                Ok(Arc::new(Column::new(field.name().as_str(), idx)))
             }
+            ExprStruct::Unbound(unbound) => Ok(Arc::new(UnKnownColumn::new(unbound.name.as_str()))),
             ExprStruct::IsNotNull(is_notnull) => {
                 let child = self.create_expr(is_notnull.child.as_ref().unwrap(), input_schema)?;
                 Ok(Arc::new(IsNotNullExpr::new(child)))

core/src/execution/proto/expr.proto

@@ -75,6 +75,7 @@ message Expr {
     BitwiseNot bitwiseNot = 48;
     Abs abs = 49;
     Subquery subquery = 50;
+    UnboundReference unbound = 51;
   }
 }
 
@@ -254,6 +255,11 @@ message BoundReference {
   DataType datatype = 2;
 }
 
+message UnboundReference {
+  string name = 1;
+  DataType datatype = 2;
+}
+
 message SortOrder {
   Expr child = 1;
   SortDirection direction = 2;