push down filter: consider already pushed filters

If optimizer runs several times it can try to push filters into
table provider several times. But, `supports_filters_pushdown`
may be context-dependent. For example, it may support any filter
like `column=value` but not the both at the same time.

Consider the following optimizer run:

1. Try to push `a = 1, b = 1`.
   `supports_filters_pushdown` returns [Exact, Inexact]
   Ok, will remember that `a=1` is pushed and make 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 pushed filters. So, there is nothing
   to left but to answer: "Exact".
   Now, the optimizer thinks that conjunction `a = 1 AND b = 1` is supported
   exactly, but it's not.

To prevent this problem, this patch adds already pushed filters to the
`supports_filters_pushdown` call.

Author: askalt

datafusion/optimizer/src/push_down_filter.rs

@@ -872,7 +872,13 @@ impl OptimizerRule for PushDownFilter {
                 convert_to_cross_join_if_beneficial(plan.data)
             }
             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 results = scan
                     .source
                     .supports_filters_pushdown(filter_predicates.as_slice())?;
@@ -889,13 +895,8 @@ impl OptimizerRule for PushDownFilter {
                     .clone()
                     .filter(|(_, res)| res != &TableProviderFilterPushDown::Unsupported)
                     .map(|(pred, _)| pred);
-                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();
                 let new_predicate: Vec<Expr> = zip
                     .filter(|(_, res)| res != &TableProviderFilterPushDown::Exact)
                     .map(|(pred, _)| pred.clone())

datafusion/sqllogictest/test_files/predicates.slt

@@ -39,7 +39,7 @@ CREATE EXTERNAL TABLE aggregate_test_100 (
   c13 VARCHAR NOT NULL
 )
 STORED AS CSV
-LOCATION '../../testing/data/csv/aggregate_test_100.csv' 
+LOCATION '../../testing/data/csv/aggregate_test_100.csv'
 OPTIONS ('format.has_header' 'true');
 
 statement ok
@@ -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 = Utf8("Brand#12") AND part.p_size <= Int32(5) OR part.p_brand = Utf8("Brand#23") AND part.p_size <= Int32(10) OR part.p_brand = Utf8("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 = Utf8("Brand#12") AND part.p_size <= Int32(5) OR part.p_brand = Utf8("Brand#23") AND part.p_size <= Int32(10) OR part.p_brand = Utf8("Brand#34") AND part.p_size <= Int32(15)]
+06)------TableScan: part projection=[p_partkey, p_brand, p_size], partial_filters=[part.p_brand = Utf8("Brand#12") AND part.p_size <= Int32(5) OR part.p_brand = Utf8("Brand#23") AND part.p_size <= Int32(10) OR part.p_brand = Utf8("Brand#34") AND part.p_size <= Int32(15), part.p_size >= Int32(1)]
 physical_plan
 01)CoalesceBatchesExec: target_batch_size=8192
 02)--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]