remove order by expr

apache · Jun 2, 2021 · faa28c6 · faa28c6
1 parent 4eccb22
commit faa28c6
Show file tree

Hide file tree

Showing 7 changed files with 13 additions and 70 deletions.
diff --git a/ballista/rust/core/src/serde/logical_plan/to_proto.rs b/ballista/rust/core/src/serde/logical_plan/to_proto.rs
@@ -767,7 +767,6 @@ impl TryInto<protobuf::LogicalPlanNode> for &LogicalPlan {
                 // filter_by_expr,
                 // FIXME implement next
                 // partition_by_expr,
-                order_by_expr,
                 // FIXME implement next
                 // window_frame,
                 ..
@@ -776,8 +775,6 @@ impl TryInto<protobuf::LogicalPlanNode> for &LogicalPlan {
                 // FIXME: implement
                 // let filter_by_expr = vec![];
                 // FIXME: implement
-                let partition_by_expr = vec![];
-                // FIXME: implement
                 let window_frame = None;
                 Ok(protobuf::LogicalPlanNode {
                     logical_plan_type: Some(LogicalPlanType::Window(Box::new(
@@ -787,11 +784,8 @@ impl TryInto<protobuf::LogicalPlanNode> for &LogicalPlan {
                                 .iter()
                                 .map(|expr| expr.try_into())
                                 .collect::<Result<Vec<_>, _>>()?,
-                            partition_by_expr,
-                            order_by_expr: order_by_expr
-                                .iter()
-                                .map(|expr| expr.try_into())
-                                .collect::<Result<Vec<_>, _>>()?,
+                            partition_by_expr: vec![],
+                            order_by_expr: vec![],
                             window_frame,
                         },
                     ))),

diff --git a/datafusion/src/logical_plan/builder.rs b/datafusion/src/logical_plan/builder.rs
@@ -31,7 +31,6 @@ use super::{
 use crate::datasource::TableProvider;
 use crate::error::{DataFusionError, Result};
 use crate::logical_plan::{DFField, DFSchema, DFSchemaRef, Partitioning};
-use crate::sql::utils::find_sort_exprs;
 use crate::{
     datasource::{empty::EmptyTable, parquet::ParquetTable, CsvFile, MemTable},
     prelude::CsvReadOptions,
@@ -310,22 +309,18 @@ impl LogicalPlanBuilder {
     ) -> Result<Self> {
         // FIXME: implement next
         // let partition_by_expr = partition_by_expr.into_iter().collect::<Vec<Expr>>();
-        // pull order by expressions from the window expressions
-        let order_by_expr = find_sort_exprs(&window_expr);
-        // let order_by_expr = vec![];
-
         let all_expr = window_expr.iter();
         validate_unique_names("Windows", all_expr.clone(), self.plan.schema())?;
 
         let mut window_fields: Vec<DFField> =
             exprlist_to_fields(all_expr, self.plan.schema())?;
         window_fields.extend_from_slice(self.plan.schema().fields());
 
+        dbg!("before");
         Ok(Self::from(&LogicalPlan::Window {
             input: Arc::new(self.plan.clone()),
             // FIXME implement next
             // partition_by_expr,
-            order_by_expr,
             // FIXME implement next
             // window_frame,
             window_expr,

diff --git a/datafusion/src/logical_plan/plan.rs b/datafusion/src/logical_plan/plan.rs
@@ -90,8 +90,6 @@ pub enum LogicalPlan {
         // filter_by_expr: Vec<Expr>,
         /// Partition by expressions
         // partition_by_expr: Vec<Expr>,
-        /// Order by expressions
-        order_by_expr: Vec<Expr>,
         /// Window Frame
         // window_frame: Option<WindowFrame>,
         /// The schema description of the window output
@@ -310,13 +308,8 @@ impl LogicalPlan {
                 // filter_by_expr,
                 // FIXME implement next
                 // partition_by_expr,
-                order_by_expr,
                 ..
-            } => window_expr
-                .iter()
-                .chain(order_by_expr.iter())
-                .cloned()
-                .collect(),
+            } => window_expr.clone(),
             LogicalPlan::Aggregate {
                 group_expr,
                 aggr_expr,
@@ -698,14 +691,12 @@ impl LogicalPlan {
                         ref window_expr,
                         // FIXME implement next
                         // ref partition_by_expr,
-                        ref order_by_expr,
                         ..
                     } => {
                         write!(
                             f,
-                            "WindowAggr: windowExpr=[{:?}] partitionBy=[], orderBy=[{:?}]",
-                            window_expr,
-                            order_by_expr
+                            "WindowAggr: windowExpr=[{:?}] partitionBy=[]",
+                            window_expr
                         )
                     }
                     LogicalPlan::Aggregate {

diff --git a/datafusion/src/optimizer/projection_push_down.rs b/datafusion/src/optimizer/projection_push_down.rs
@@ -201,7 +201,6 @@ fn optimize_plan(
             // filter_by_expr,
             // FIXME implement next
             // partition_by_expr,
-            order_by_expr,
             // FIXME implement next
             // window_frame,
             ..
@@ -222,21 +221,6 @@ fn optimize_plan(
                 })?;
             }
 
-            let mut new_order_by_expr = Vec::new();
-            {
-                order_by_expr.iter().try_for_each(|expr| {
-                    let name = &expr.name(&schema)?;
-                    if required_columns.contains(name) {
-                        new_order_by_expr.push(expr.clone());
-                        new_required_columns.insert(name.clone());
-                        // add to the new set of required columns
-                        utils::expr_to_column_names(expr, &mut new_required_columns)
-                    } else {
-                        Ok(())
-                    }
-                })?;
-            }
-
             let new_schema = DFSchema::new(
                 schema
                     .fields()
@@ -250,7 +234,6 @@ fn optimize_plan(
                 window_expr: new_window_expr,
                 // FIXME implement next
                 // partition_by_expr: partition_by_expr.clone(),
-                order_by_expr: new_order_by_expr,
                 // FIXME implement next
                 // window_frame: window_frame.clone(),
                 input: Arc::new(optimize_plan(

diff --git a/datafusion/src/optimizer/utils.rs b/datafusion/src/optimizer/utils.rs
@@ -198,24 +198,13 @@ pub fn from_plan(
             // FIXME implement next
             // window_frame,
             window_expr,
-            order_by_expr,
             schema,
             ..
-        } => match expr.len() {
-            expr_len if expr_len == window_expr.len() + order_by_expr.len() => {
-                Ok(LogicalPlan::Window {
-                    input: Arc::new(inputs[0].clone()),
-                    window_expr: expr[0..window_expr.len()].to_vec(),
-                    order_by_expr: expr[window_expr.len()..].to_vec(),
-                    schema: schema.clone(),
-                })
-            }
-            _ => Err(DataFusionError::Internal(format!(
-                "Unexpected expression length, need {}, got {}",
-                window_expr.len() + order_by_expr.len(),
-                expr.len()
-            ))),
-        },
+        } => Ok(LogicalPlan::Window {
+            input: Arc::new(inputs[0].clone()),
+            window_expr: expr[0..window_expr.len()].to_vec(),
+            schema: schema.clone(),
+        }),
         LogicalPlan::Aggregate {
             group_expr, schema, ..
         } => Ok(LogicalPlan::Aggregate {

diff --git a/datafusion/src/physical_plan/planner.rs b/datafusion/src/physical_plan/planner.rs
@@ -142,10 +142,7 @@ impl DefaultPhysicalPlanner {
                 ..
             } => source.scan(projection, batch_size, filters, *limit),
             LogicalPlan::Window {
-                input,
-                window_expr,
-                order_by_expr,
-                ..
+                input, window_expr, ..
             } => {
                 // Initially need to perform the aggregate and then merge the partitions
                 let input_exec = self.create_initial_plan(input, ctx_state)?;
@@ -166,12 +163,6 @@ impl DefaultPhysicalPlanner {
                     })
                     .collect::<Result<Vec<_>>>()?;
 
-                if !order_by_expr.is_empty() {
-                    return Err(DataFusionError::NotImplemented(
-                        "Window function with order by is not yet implemented".to_owned(),
-                    ));
-                }
-
                 Ok(Arc::new(WindowAggExec::try_new(
                     window_expr,
                     input_exec.clone(),

diff --git a/datafusion/src/sql/planner.rs b/datafusion/src/sql/planner.rs
@@ -684,7 +684,7 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
         groups.reverse();
         for (sort_keys, exprs) in groups {
             if !sort_keys.is_empty() {
-                let sort_keys: Vec<Expr> = sort_keys.into_iter().cloned().collect();
+                let sort_keys: Vec<Expr> = sort_keys.to_vec();
                 plan = LogicalPlanBuilder::from(&plan).sort(sort_keys)?.build()?;
             }
             let window_exprs: Vec<Expr> = exprs.into_iter().cloned().collect();