fix: pass filters pushed earlier to supports_filter_pushdown

Part of apache#19929

Let "t" be a table provider that supports exactly any single filter
but not a conjunction. Consider the following optimizer pipeline:

1. Try to push `a = 1, b = 1`.
   `supports_filters_pushdown` returns [Exact, Inexact]
   OK: the optimizer records that a = 1 is pushed and creates a filter node for b = 1.

...
Another optimization iteration.

2. Try to push `b = 1`.
   `supports_filters_pushdown` returns [Exact]. Of course, the table provider can't
   remember all previously pushed filters, so it has no choice but to answer `Exact`.
   Now, the optimizer thinks the conjunction a = 1 AND b = 1 is supported exactly, but
   it is not.

To prevent this problem, this patch passes filters that were already pushed into the scan
earlier to `supports_filters_pushdown`.

Author: askalt

datafusion/optimizer/src/push_down_filter.rs

@@ -1126,7 +1126,12 @@ impl OptimizerRule for PushDownFilter {
             }
             LogicalPlan::Join(join) => push_down_join(join, Some(&filter.predicate)),
             LogicalPlan::TableScan(scan) => {
-                let filter_predicates = split_conjunction(&filter.predicate);
+                let filter_predicates: Vec<_> = split_conjunction(&filter.predicate)
+                    .into_iter()
+                    // Add already pushed filters.
+                    .chain(scan.filters.iter())
+                    .unique()
+                    .collect();
 
                 let (volatile_filters, non_volatile_filters): (Vec<&Expr>, Vec<&Expr>) =
                     filter_predicates
@@ -1154,13 +1159,8 @@ impl OptimizerRule for PushDownFilter {
                     .map(|(pred, _)| pred);
 
                 // Add new scan filters
-                let new_scan_filters: Vec<Expr> = scan
-                    .filters
-                    .iter()
-                    .chain(new_scan_filters)
-                    .unique()
-                    .cloned()
-                    .collect();
+                let new_scan_filters: Vec<Expr> =
+                    new_scan_filters.unique().cloned().collect();
 
                 // Compose predicates to be of `Unsupported` or `Inexact` pushdown type, and also include volatile filters
                 let new_predicate: Vec<Expr> = zip
@@ -1438,8 +1438,8 @@ mod tests {
     use datafusion_expr::{
         ColumnarValue, ExprFunctionExt, Extension, LogicalPlanBuilder,
         ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, TableSource, TableType,
-        UserDefinedLogicalNodeCore, Volatility, WindowFunctionDefinition, col, in_list,
-        in_subquery, lit,
+        UserDefinedLogicalNodeCore, Volatility, WindowFunctionDefinition, binary_expr,
+        col, in_list, in_subquery, lit,
     };
 
     use crate::OptimizerContext;
@@ -1459,7 +1459,17 @@ mod tests {
             $plan:expr,
             @ $expected:literal $(,)?
         ) => {{
-            let optimizer_ctx = OptimizerContext::new().with_max_passes(1);
+            assert_optimized_plan_equal_with_pases_num!($plan, 1, @ $expected,)
+        }};
+    }
+
+    macro_rules! assert_optimized_plan_equal_with_pases_num {
+        (
+            $plan:expr,
+            $max_pases: expr,
+            @ $expected:literal $(,)?
+        ) => {{
+            let optimizer_ctx = OptimizerContext::new().with_max_passes($max_pases);
             let rules: Vec<Arc<dyn crate::OptimizerRule + Send + Sync>> = vec![Arc::new(PushDownFilter::new())];
             assert_optimized_plan_eq_snapshot!(
                 optimizer_ctx,
@@ -4223,4 +4233,65 @@ mod tests {
         "
         )
     }
+
+    struct SingleFilterSupportSource {
+        schema: SchemaRef,
+    }
+
+    #[async_trait]
+    impl TableSource for SingleFilterSupportSource {
+        fn schema(&self) -> SchemaRef {
+            Arc::clone(&self.schema)
+        }
+
+        fn table_type(&self) -> TableType {
+            TableType::Base
+        }
+
+        fn supports_filters_pushdown(
+            &self,
+            filters: &[&Expr],
+        ) -> Result<Vec<TableProviderFilterPushDown>> {
+            // Support exactly any single filter.
+            let mut res = vec![TableProviderFilterPushDown::Unsupported; filters.len()];
+            res[0] = TableProviderFilterPushDown::Exact;
+            Ok(res)
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            self
+        }
+    }
+
+    #[test]
+    fn test_pushed_filters_passed_again() -> Result<()> {
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Int32, true),
+            Field::new("b", DataType::Int32, true),
+        ]));
+
+        let plan = LogicalPlanBuilder::scan(
+            "t",
+            Arc::new(SingleFilterSupportSource {
+                schema: Arc::clone(&schema),
+            }),
+            None,
+        )?
+        .filter(binary_expr(
+            binary_expr(col("a"), Operator::Eq, lit(1)),
+            Operator::And,
+            binary_expr(col("b"), Operator::Eq, lit(4)),
+        ))?
+        .build()?;
+
+        // Verify that the only one filter is pushed.
+        assert_optimized_plan_equal_with_pases_num!(
+            plan,
+            5,
+            @r"
+        Filter: t.b = Int32(4)
+          TableScan: t, full_filters=[t.a = Int32(1)]
+        "
+        )
+    }
 }

datafusion/sqllogictest/test_files/predicates.slt

@@ -666,7 +666,7 @@ logical_plan
 03)----Filter: lineitem.l_quantity >= Decimal128(Some(100),15,2) AND lineitem.l_quantity <= Decimal128(Some(1100),15,2) OR lineitem.l_quantity >= Decimal128(Some(1000),15,2) AND lineitem.l_quantity <= Decimal128(Some(2000),15,2) OR lineitem.l_quantity >= Decimal128(Some(2000),15,2) AND lineitem.l_quantity <= Decimal128(Some(3000),15,2)
 04)------TableScan: lineitem projection=[l_partkey, l_quantity], partial_filters=[lineitem.l_quantity >= Decimal128(Some(100),15,2) AND lineitem.l_quantity <= Decimal128(Some(1100),15,2) OR lineitem.l_quantity >= Decimal128(Some(1000),15,2) AND lineitem.l_quantity <= Decimal128(Some(2000),15,2) OR lineitem.l_quantity >= Decimal128(Some(2000),15,2) AND lineitem.l_quantity <= Decimal128(Some(3000),15,2)]
 05)----Filter: (part.p_brand = Utf8View("Brand#12") AND part.p_size <= Int32(5) OR part.p_brand = Utf8View("Brand#23") AND part.p_size <= Int32(10) OR part.p_brand = Utf8View("Brand#34") AND part.p_size <= Int32(15)) AND part.p_size >= Int32(1)
-06)------TableScan: part projection=[p_partkey, p_brand, p_size], partial_filters=[part.p_size >= Int32(1), part.p_brand = Utf8View("Brand#12") AND part.p_size <= Int32(5) OR part.p_brand = Utf8View("Brand#23") AND part.p_size <= Int32(10) OR part.p_brand = Utf8View("Brand#34") AND part.p_size <= Int32(15)]
+06)------TableScan: part projection=[p_partkey, p_brand, p_size], partial_filters=[part.p_brand = Utf8View("Brand#12") AND part.p_size <= Int32(5) OR part.p_brand = Utf8View("Brand#23") AND part.p_size <= Int32(10) OR part.p_brand = Utf8View("Brand#34") AND part.p_size <= Int32(15), part.p_size >= Int32(1)]
 physical_plan
 01)HashJoinExec: mode=Partitioned, join_type=Inner, on=[(l_partkey@0, p_partkey@0)], filter=p_brand@1 = Brand#12 AND l_quantity@0 >= Some(100),15,2 AND l_quantity@0 <= Some(1100),15,2 AND p_size@2 <= 5 OR p_brand@1 = Brand#23 AND l_quantity@0 >= Some(1000),15,2 AND l_quantity@0 <= Some(2000),15,2 AND p_size@2 <= 10 OR p_brand@1 = Brand#34 AND l_quantity@0 >= Some(2000),15,2 AND l_quantity@0 <= Some(3000),15,2 AND p_size@2 <= 15, projection=[l_partkey@0]
 02)--RepartitionExec: partitioning=Hash([l_partkey@0], 4), input_partitions=4