WIP

datafusion-examples/examples/rewrite_expr.rs

@@ -98,7 +98,7 @@ impl MyAnalyzerRule {
                     Transformed::Yes(LogicalPlan::Filter(Filter::try_new(
                         predicate,
                         filter.input,
-                        filter.projection.clone(),
+                        filter.projected_schema.clone(),
                     )?))
                 }
                 _ => Transformed::No(plan),
@@ -143,7 +143,7 @@ impl OptimizerRule for MyOptimizerRule {
                 Ok(Some(LogicalPlan::Filter(Filter::try_new(
                     predicate,
                     filter.input,
-                    filter.projection.clone(),
+                    filter.projected_schema.clone(),
                 )?)))
             }
             Some(optimized_plan) => Ok(Some(optimized_plan)),
@@ -153,7 +153,7 @@ impl OptimizerRule for MyOptimizerRule {
                     Ok(Some(LogicalPlan::Filter(Filter::try_new(
                         predicate,
                         filter.input.clone(),
-                        filter.projection.clone(),
+                        filter.projected_schema.clone(),
                     )?)))
                 }
                 _ => Ok(None),

datafusion/expr/src/logical_plan/plan.rs

@@ -163,8 +163,10 @@ impl LogicalPlan {
             }) => projected_schema,
             LogicalPlan::Projection(Projection { schema, .. }) => schema,
             LogicalPlan::Filter(Filter {
-                projected_schema, ..
-            }) => projected_schema,
+                projected_schema,
+                input,
+                ..
+            }) => projected_schema.as_ref().unwrap_or(input.schema()),
             LogicalPlan::Distinct(Distinct { input }) => input.schema(),
             LogicalPlan::Window(Window { schema, .. }) => schema,
             LogicalPlan::Aggregate(Aggregate { schema, .. }) => schema,
@@ -1549,14 +1551,14 @@ impl LogicalPlan {
                     }
                     LogicalPlan::Filter(Filter {
                         predicate: ref expr,
-                        projection,
                         projected_schema,
                         ..
                     }) => {
-                        if let Some(indices) = projection {
-                            let names: Vec<&str> = indices
+                        if let Some(projected_schema) = projected_schema {
+                            let names: Vec<&str> = projected_schema
+                                .fields()
                                 .iter()
-                                .map(|i| projected_schema.field(*i).name().as_str())
+                                .map(|i| i.name().as_str())
                                 .collect();
 
                             write!(f, "Filter: {expr}, projection=[{}]", names.join(", "))
@@ -1853,18 +1855,16 @@ pub struct Filter {
     pub predicate: Expr,
     /// The incoming logical plan
     pub input: Arc<LogicalPlan>,
-    /// Optional projection
-    pub projection: Option<Vec<usize>>,
     /// Schema representing the data after the optional projection is applied
-    pub projected_schema: DFSchemaRef,
+    pub projected_schema: Option<DFSchemaRef>,
 }
 
 impl Filter {
     /// Create a new filter operator.
     pub fn try_new(
         predicate: Expr,
         input: Arc<LogicalPlan>,
-        projection: Option<Vec<usize>>,
+        projected_schema: Option<DFSchemaRef>,
     ) -> Result<Self> {
         // Filter predicates must return a boolean value so we try and validate that here.
         // Note that it is not always possible to resolve the predicate expression during plan
@@ -1889,20 +1889,31 @@ impl Filter {
 
         let func_dependencies = input.schema().functional_dependencies();
 
-        let projected_schema = projection
+        let projected_schema = projected_schema
             .as_ref()
-            .map(|p| {
-                let projected_func_dependencies =
-                    func_dependencies.project_functional_dependencies(p, p.len());
+            .map(|p| -> Result<_> {
+                let indices: Vec<usize> = p
+                    .fields()
+                    .iter()
+                    .filter_map(|x| {
+                        input
+                            .schema()
+                            .index_of_column_by_name(x.qualifier(), x.name())
+                            .transpose()
+                    })
+                    .collect::<Result<_>>()?;
+                let projected_func_dependencies = func_dependencies
+                    .project_functional_dependencies(&indices, indices.len());
                 let schema = input.schema().as_ref().clone();
-                Arc::new(schema.with_functional_dependencies(projected_func_dependencies))
+                Result::Ok(Arc::new(
+                    schema.with_functional_dependencies(projected_func_dependencies),
+                ))
             })
-            .unwrap_or_else(|| input.schema().clone());
+            .transpose()?;
 
         Ok(Self {
             predicate,
             input,
-            projection,
             projected_schema,
         })
     }

datafusion/optimizer/src/analyzer/count_wildcard_rule.rs

@@ -102,12 +102,14 @@ fn analyze_internal(plan: LogicalPlan) -> Result<Transformed<LogicalPlan>> {
         LogicalPlan::Filter(Filter {
             predicate,
             input,
-            projection,
+            projected_schema,
             ..
         }) => {
             let predicate = rewrite_preserving_name(predicate, &mut rewriter)?;
             Ok(Transformed::Yes(LogicalPlan::Filter(Filter::try_new(
-                predicate, input, projection,
+                predicate,
+                input,
+                projected_schema.clone(),
             )?)))
         }
 

datafusion/optimizer/src/analyzer/inline_table_scan.rs

@@ -78,7 +78,7 @@ fn analyze_internal(plan: LogicalPlan) -> Result<Transformed<LogicalPlan>> {
             Transformed::Yes(LogicalPlan::Filter(Filter::try_new(
                 new_expr,
                 filter.input,
-                filter.projection,
+                filter.projected_schema.clone(),
             )?))
         }
         _ => Transformed::No(plan),

datafusion/optimizer/src/push_down_projection.rs

@@ -471,10 +471,18 @@ fn push_down_filter(
         })
         .collect::<Result<Vec<usize>>>()?;
 
+    let new_schema = DFSchema::new_with_metadata(
+        projection
+            .into_iter()
+            .map(|i| schema.fields()[i].clone())
+            .collect(),
+        schema.metadata().clone(),
+    )?;
+
     Filter::try_new(
         filter.predicate.clone(),
         filter.input.clone(),
-        Some(projection),
+        Some(Arc::new(new_schema)),
     )
     .map(LogicalPlan::Filter)
 }

datafusion/optimizer/src/rewrite_disjunctive_predicate.rs

@@ -137,7 +137,7 @@ impl OptimizerRule for RewriteDisjunctivePredicate {
                 Ok(Some(LogicalPlan::Filter(Filter::try_new(
                     rewritten_expr,
                     filter.input.clone(),
-                    filter.projection.clone(),
+                    filter.projected_schema.clone(),
                 )?)))
             }
             _ => Ok(None),