Fix

Dandandan · Dec 1, 2023 · 189ca07 · 189ca07
1 parent 137fc68
commit 189ca07
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 23 deletions.
diff --git a/datafusion/core/src/physical_optimizer/projection_pushdown.rs b/datafusion/core/src/physical_optimizer/projection_pushdown.rs
@@ -1933,7 +1933,7 @@ mod tests {
             )),
         ));
         let filter: Arc<dyn ExecutionPlan> =
-            Arc::new(FilterExec::try_new(predicate, csv)?);
+            Arc::new(FilterExec::try_new(predicate, None, csv)?);
         let projection: Arc<dyn ExecutionPlan> = Arc::new(ProjectionExec::try_new(
             vec![
                 (Arc::new(Column::new("a", 0)), "a_new".to_string()),

diff --git a/datafusion/optimizer/src/optimize_projections.rs b/datafusion/optimizer/src/optimize_projections.rs
@@ -21,9 +21,9 @@
 //! - Adds projection to decrease table column size before operators that benefits from less memory at its input.
 //! - Removes unnecessary [LogicalPlan::Projection] from the [LogicalPlan].
 use crate::optimizer::ApplyOrder;
-use datafusion_common::{Column, DFSchema, DFSchemaRef, JoinType, Result, DFField};
-use datafusion_expr::Filter;
+use datafusion_common::{Column, DFField, DFSchema, DFSchemaRef, JoinType, Result};
 use datafusion_expr::expr::{Alias, ScalarFunction};
+use datafusion_expr::Filter;
 use datafusion_expr::{
     logical_plan::LogicalPlan, projection_schema, Aggregate, BinaryExpr, Cast, Distinct,
     Expr, Projection, ScalarFunctionDefinition, TableScan, Window,
@@ -352,25 +352,23 @@ fn optimize_projections(
             LogicalPlan::Filter(f) => {
                 let schema = res.schema();
                 let fields: Vec<DFField> = indices
-                .iter()
-                .map(|c| {
-                    plan.schema().field(*c).clone()
-                })
-                .collect();
-                let new_schema = DFSchema::new_with_metadata(fields, schema.metadata().clone())?;
-                let new_schema = if !schema.logically_equivalent_names_and_types(&new_schema) {
-                    Some(Arc::new(new_schema))
-                } else {
-                    None
-                };
-
-                Ok(Some(Filter::try_new(
-                    f.predicate.clone(),
-                    f.input.clone(),
-                    new_schema,
-                ).map(LogicalPlan::Filter)?))
-
-            },
+                    .iter()
+                    .map(|c| plan.schema().field(*c).clone())
+                    .collect();
+                let new_schema =
+                    DFSchema::new_with_metadata(fields, schema.metadata().clone())?;
+                let new_schema =
+                    if !schema.logically_equivalent_names_and_types(&new_schema) {
+                        Some(Arc::new(new_schema))
+                    } else {
+                        None
+                    };
+
+                Ok(Some(
+                    Filter::try_new(f.predicate.clone(), f.input.clone(), new_schema)
+                        .map(LogicalPlan::Filter)?,
+                ))
+            }
             _ => Ok(Some(res)),
         }
     }

diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs
@@ -1016,7 +1016,7 @@ mod tests {
             Arc::new(Literal::new(ScalarValue::Int32(Some(10)))),
         ));
         let filter: Arc<dyn ExecutionPlan> =
-            Arc::new(FilterExec::try_new(predicate, input)?);
+            Arc::new(FilterExec::try_new(predicate, None, input)?);
         let filter_statistics = filter.statistics()?;
         // First column is "a", and it is a column with only one value after the filter.
         assert!(filter_statistics.column_statistics[0].is_singleton());