Consider table scan filter during analysis of optimize projections

datafusion/optimizer/src/optimize_projections.rs

@@ -330,6 +330,12 @@ fn optimize_projections(
             vec![(left_child_indices, true), (right_child_indices, true)]
         }
         LogicalPlan::TableScan(table_scan) => {
+            // Get required field indices by filter expressions
+            let referred_indices = indices_referred_by_exprs(
+                &table_scan.projected_schema,
+                table_scan.filters.iter(),
+            )?;
+            let indices = merge_slices(indices, &referred_indices);
             let schema = table_scan.source.schema();
             // Get indices referred to in the original (schema with all fields)
             // given projected indices.
@@ -903,9 +909,12 @@ mod tests {
     use std::sync::Arc;
 
     use crate::optimize_projections::OptimizeProjections;
-    use crate::test::{assert_optimized_plan_eq, test_table_scan};
+    use crate::test::{
+        assert_optimized_plan_eq, test_table_scan, test_table_scan_fields,
+    };
     use arrow::datatypes::{DataType, Field, Schema};
     use datafusion_common::{Result, TableReference};
+    use datafusion_expr::builder::table_scan_with_filters;
     use datafusion_expr::{
         binary_expr, col, count, lit, logical_plan::builder::LogicalPlanBuilder, not,
         table_scan, try_cast, when, Expr, Like, LogicalPlan, Operator,
@@ -1193,4 +1202,23 @@ mod tests {
         \n    TableScan: test projection=[a]";
         assert_optimized_plan_equal(&plan, expected)
     }
+
+    #[test]
+    fn filter_with_table_optimize_projection() -> Result<()> {
+        let schema = Schema::new(test_table_scan_fields());
+        let table_scan = table_scan_with_filters(
+            Some("test"),
+            &schema,
+            None,
+            vec![col("b").is_not_null()],
+        )?
+        .build()?;
+        let plan = LogicalPlanBuilder::from(table_scan)
+            .project(vec![col("a")])?
+            .build()?;
+
+        let expected = "Projection: test.a\
+        \n  TableScan: test projection=[a, b], full_filters=[b IS NOT NULL]";
+        assert_optimized_plan_equal(&plan, expected)
+    }
 }

datafusion/optimizer/src/push_down_projection.rs

@@ -486,7 +486,7 @@ mod tests {
 
         let expected = "\
         Projection: Int32(1) AS a\
-        \n  TableScan: test projection=[], full_filters=[b = Int32(1)]";
+        \n  TableScan: test projection=[b], full_filters=[b = Int32(1)]";
 
         assert_optimized_plan_eq(&plan, expected)
     }