Don't add filters to used columns (apache#7670)

Add test

WIP fix

Fix filter after scan

Totally reemove filter to column extraction

Fix test

Update tests 1

Update tests 2

Update tests 3

datafusion/core/src/dataframe.rs

@@ -1287,15 +1287,16 @@ impl TableProvider for DataFrameTableProvider {
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let mut expr = LogicalPlanBuilder::from(self.plan.clone());
-        if let Some(p) = projection {
-            expr = expr.select(p.iter().copied())?
-        }
-
         // Add filter when given
         let filter = filters.iter().cloned().reduce(|acc, new| acc.and(new));
         if let Some(filter) = filter {
             expr = expr.filter(filter)?
         }
+
+        if let Some(p) = projection {
+            expr = expr.select(p.iter().copied())?
+        }
+
         // add a limit if given
         if let Some(l) = limit {
             expr = expr.limit(0, Some(l))?

datafusion/core/src/datasource/view.rs

@@ -108,12 +108,20 @@ impl TableProvider for ViewTable {
         filters: &[Expr],
         limit: Option<usize>,
     ) -> Result<Arc<dyn ExecutionPlan>> {
-        let plan = if let Some(projection) = projection {
+        let filter = filters.iter().cloned().reduce(|acc, new| acc.and(new));
+        let plan = self.logical_plan().clone();
+        let mut plan = LogicalPlanBuilder::from(plan);
+
+        if let Some(filter) = filter {
+            plan = plan.filter(filter)?;
+        }
+
+        let mut plan = if let Some(projection) = projection {
             // avoiding adding a redundant projection (e.g. SELECT * FROM view)
             let current_projection =
-                (0..self.logical_plan.schema().fields().len()).collect::<Vec<usize>>();
+                (0..plan.schema().fields().len()).collect::<Vec<usize>>();
             if projection == &current_projection {
-                self.logical_plan().clone()
+                plan
             } else {
                 let fields: Vec<Expr> = projection
                     .iter()
@@ -123,19 +131,11 @@ impl TableProvider for ViewTable {
                         )
                     })
                     .collect();
-                LogicalPlanBuilder::from(self.logical_plan.clone())
-                    .project(fields)?
-                    .build()?
+                plan.project(fields)?
             }
         } else {
-            self.logical_plan().clone()
+            plan
         };
-        let mut plan = LogicalPlanBuilder::from(plan);
-        let filter = filters.iter().cloned().reduce(|acc, new| acc.and(new));
-
-        if let Some(filter) = filter {
-            plan = plan.filter(filter)?;
-        }
 
         if let Some(limit) = limit {
             plan = plan.limit(0, Some(limit))?;
@@ -439,6 +439,7 @@ mod tests {
             .select_columns(&["bool_col", "int_col"])?;
 
         let plan = df.explain(false, false)?.collect().await?;
+
         // Filters all the way to Parquet
         let formatted = arrow::util::pretty::pretty_format_batches(&plan)
             .unwrap()

datafusion/expr/src/logical_plan/builder.rs

@@ -26,7 +26,9 @@ use crate::expr_rewriter::{
 };
 use crate::type_coercion::binary::comparison_coercion;
 use crate::utils::{columnize_expr, compare_sort_expr};
-use crate::{and, binary_expr, DmlStatement, Operator, WriteOp};
+use crate::{
+    and, binary_expr, DmlStatement, Operator, TableProviderFilterPushDown, WriteOp,
+};
 use crate::{
     logical_plan::{
         Aggregate, Analyze, CrossJoin, Distinct, EmptyRelation, Explain, Filter, Join,
@@ -1402,6 +1404,13 @@ impl TableSource for LogicalTableSource {
     fn schema(&self) -> SchemaRef {
         self.table_schema.clone()
     }
+
+    fn supports_filters_pushdown(
+        &self,
+        filters: &[&Expr],
+    ) -> Result<Vec<crate::TableProviderFilterPushDown>> {
+        Ok(vec![TableProviderFilterPushDown::Exact; filters.len()])
+    }
 }
 
 /// Create a [`LogicalPlan::Unnest`] plan