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

…ation (apache#155)

This patch adds the check of the index of bound reference. The aggregate expressions of final aggregation are not bound to the input of partial aggregation anymore but sent to native side as unbound expressions.

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;