field rename

Author: Jiayu Liu

ballista/rust/core/proto/ballista.proto

@@ -312,8 +312,8 @@ message AggregateNode {
 
 message WindowNode {
   LogicalPlanNode input = 1;
-  repeated LogicalExprNode group_expr = 2;
-  repeated LogicalExprNode aggr_expr = 3;
+  repeated LogicalExprNode partition_by_expr = 2;
+  repeated LogicalExprNode order_by_expr = 3;
 }
 
 enum JoinType {

ballista/rust/core/src/serde/logical_plan/from_proto.rs

@@ -78,18 +78,18 @@ impl TryInto<LogicalPlan> for &protobuf::LogicalPlanNode {
             }
             LogicalPlanType::Window(window) => {
                 let input: LogicalPlan = convert_box_required!(window.input)?;
-                let group_expr = window
-                    .group_expr
+                let partition_by_expr = window
+                    .partition_by_expr
                     .iter()
                     .map(|expr| expr.try_into())
                     .collect::<Result<Vec<_>, _>>()?;
-                let aggr_expr = window
-                    .aggr_expr
+                let order_by_expr = window
+                    .order_by_expr
                     .iter()
                     .map(|expr| expr.try_into())
                     .collect::<Result<Vec<_>, _>>()?;
                 LogicalPlanBuilder::from(&input)
-                    .window(group_expr, aggr_expr)?
+                    .window(partition_by_expr, order_by_expr)?
                     .build()
                     .map_err(|e| e.into())
             }

ballista/rust/core/src/serde/logical_plan/to_proto.rs

@@ -775,20 +775,20 @@ impl TryInto<protobuf::LogicalPlanNode> for &LogicalPlan {
             }
             LogicalPlan::Window {
                 input,
-                group_expr,
-                aggr_expr,
+                partition_by_expr,
+                order_by_expr,
                 ..
             } => {
                 let input: protobuf::LogicalPlanNode = input.as_ref().try_into()?;
                 Ok(protobuf::LogicalPlanNode {
                     logical_plan_type: Some(LogicalPlanType::Window(Box::new(
                         protobuf::WindowNode {
                             input: Some(Box::new(input)),
-                            group_expr: group_expr
+                            partition_by_expr: partition_by_expr
                                 .iter()
                                 .map(|expr| expr.try_into())
                                 .collect::<Result<Vec<_>, BallistaError>>()?,
-                            aggr_expr: aggr_expr
+                            order_by_expr: order_by_expr
                                 .iter()
                                 .map(|expr| expr.try_into())
                                 .collect::<Result<Vec<_>, BallistaError>>()?,

datafusion/src/logical_plan/builder.rs

@@ -289,12 +289,29 @@ impl LogicalPlanBuilder {
         }))
     }
 
+    /// Apply a window: partition by the `partition_by_expr` expressions
+    /// and calculating the window spec.
     pub fn window(
         &self,
-        group_expr: impl IntoIterator<Item = Expr>,
-        aggr_expr: impl IntoIterator<Item = Expr>,
+        partition_by_expr: impl IntoIterator<Item = Expr>,
+        order_by_expr: impl IntoIterator<Item = Expr>,
     ) -> Result<Self> {
-        unimplemented!()
+        let partition_by_expr = partition_by_expr.into_iter().collect::<Vec<Expr>>();
+        let order_by_expr = order_by_expr.into_iter().collect::<Vec<Expr>>();
+
+        let all_expr = partition_by_expr.iter().chain(order_by_expr.iter());
+
+        validate_unique_names("Windows", all_expr.clone(), self.plan.schema())?;
+
+        let window_schema =
+            DFSchema::new(exprlist_to_fields(all_expr, self.plan.schema())?)?;
+
+        Ok(Self::from(&LogicalPlan::Window {
+            input: Arc::new(self.plan.clone()),
+            partition_by_expr,
+            order_by_expr,
+            schema: DFSchemaRef::new(window_schema),
+        }))
     }
 
     /// Apply an aggregate: grouping on the `group_expr` expressions

datafusion/src/logical_plan/plan.rs

@@ -87,10 +87,10 @@ pub enum LogicalPlan {
     Window {
         /// The incoming logical plan
         input: Arc<LogicalPlan>,
-        /// Grouping expressions
-        group_expr: Vec<Expr>,
-        /// Aggregate expressions
-        aggr_expr: Vec<Expr>,
+        /// Partition by expressions
+        partition_by_expr: Vec<Expr>,
+        /// Order by expressions
+        order_by_expr: Vec<Expr>,
         /// The schema description of the aggregate output
         schema: DFSchemaRef,
     },
@@ -302,11 +302,15 @@ impl LogicalPlan {
                 _ => vec![],
             },
             LogicalPlan::Window {
-                group_expr,
-                aggr_expr,
+                partition_by_expr,
+                order_by_expr,
                 ..
+            } => {
+                let mut result = partition_by_expr.clone();
+                result.extend(order_by_expr.clone());
+                result
             }
-            | LogicalPlan::Aggregate {
+            LogicalPlan::Aggregate {
                 group_expr,
                 aggr_expr,
                 ..
@@ -688,13 +692,13 @@ impl LogicalPlan {
                         ..
                     } => write!(f, "Filter: {:?}", expr),
                     LogicalPlan::Window {
-                        ref group_expr,
-                        ref aggr_expr,
+                        ref partition_by_expr,
+                        ref order_by_expr,
                         ..
                     } => write!(
                         f,
-                        "Window: groupBy=[{:?}], aggr=[{:?}]",
-                        group_expr, aggr_expr
+                        "Window: partitionBy=[{:?}], orderBy=[{:?}]",
+                        partition_by_expr, order_by_expr
                     ),
                     LogicalPlan::Aggregate {
                         ref group_expr,

datafusion/src/optimizer/projection_push_down.rs

@@ -196,11 +196,22 @@ fn optimize_plan(
         LogicalPlan::Window {
             schema,
             input,
-            group_expr,
-            aggr_expr,
+            partition_by_expr,
+            order_by_expr,
             ..
-        }
-        | LogicalPlan::Aggregate {
+        } => Ok(LogicalPlan::Window {
+            partition_by_expr: partition_by_expr.clone(),
+            order_by_expr: order_by_expr.clone(),
+            input: Arc::new(optimize_plan(
+                optimizer,
+                &input,
+                &new_required_columns,
+                true,
+                execution_props,
+            )?),
+            schema: schema.clone(),
+        }),
+        LogicalPlan::Aggregate {
             schema,
             input,
             group_expr,

datafusion/src/optimizer/utils.rs

@@ -190,10 +190,13 @@ pub fn from_plan(
             }),
         },
         LogicalPlan::Window {
-            group_expr, schema, ..
+            partition_by_expr,
+            order_by_expr,
+            schema,
+            ..
         } => Ok(LogicalPlan::Window {
-            group_expr: expr[0..group_expr.len()].to_vec(),
-            aggr_expr: expr[group_expr.len()..].to_vec(),
+            partition_by_expr: expr[0..partition_by_expr.len()].to_vec(),
+            order_by_expr: expr[order_by_expr.len()..].to_vec(),
             input: Arc::new(inputs[0].clone()),
             schema: schema.clone(),
         }),

datafusion/src/physical_plan/planner.rs

@@ -141,11 +141,13 @@ impl DefaultPhysicalPlanner {
             } => source.scan(projection, batch_size, filters, *limit),
             LogicalPlan::Window {
                 input,
-                group_expr,
-                aggr_expr,
+                partition_by_expr,
+                order_by_expr,
                 ..
+            } => {
+                unimplemented!()
             }
-            | LogicalPlan::Aggregate {
+            LogicalPlan::Aggregate {
                 input,
                 group_expr,
                 aggr_expr,