Merge pull request #7072 from leiysky/enable-push-down-filter-scan

fix(planner): Enable push down filter scan

common/storages/hive/src/hive_table.rs

@@ -213,12 +213,6 @@ impl HiveTable {
         if let Some(partition_keys) = &self.table_options.partition_keys {
             if !partition_keys.is_empty() {
                 let filter_expression = if let Some(extras) = push_downs {
-                    if extras.filters.len() > 1 {
-                        return Err(ErrorCode::UnImplement(format!(
-                            "more than one filters, {:?}",
-                            extras.filters
-                        )));
-                    };
                     extras.filters.get(0).cloned()
                 } else {
                     None

query/src/sql/optimizer/heuristic/mod.rs

@@ -54,6 +54,7 @@ pub static DEFAULT_REWRITE_RULES: Lazy<Vec<RuleID>> = Lazy::new(|| {
         RuleID::PushDownFilterProject,
         RuleID::PushDownFilterJoin,
         RuleID::SplitAggregate,
+        RuleID::PushDownFilterScan,
     ]
 });
 

query/src/sql/planner/format/display_rel_operator.rs

@@ -231,8 +231,19 @@ pub fn format_physical_scan(
     let table = metadata.read().table(op.table_index).clone();
     write!(
         f,
-        "Scan: {}.{}.{}",
-        &table.catalog, &table.database, &table.name
+        "Scan: {}.{}.{}, filters: [{}]",
+        &table.catalog,
+        &table.database,
+        &table.name,
+        op.push_down_predicates.as_ref().map_or_else(
+            || "".to_string(),
+            |predicates| {
+                predicates
+                    .iter()
+                    .map(|pred| format_scalar(metadata, pred))
+                    .join(", ")
+            }
+        )
     )
 }
 

query/tests/it/sql/optimizer/heuristic/testdata/exchange.test

@@ -3,9 +3,9 @@ select * from numbers(1) t, numbers(2) t1 where t.number = t1.number
 Exchange(Merge)
     HashJoin: INNER, build keys: [t1.number (#1)], probe keys: [t.number (#0)], join filters: []
         Exchange(Hash): keys: [t.number (#0)]
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: []
         Exchange(Hash): keys: [t1.number (#1)]
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: []
 
 
 # Result of t1 join t is distributed on t.number
@@ -15,11 +15,11 @@ Exchange(Merge)
     HashJoin: INNER, build keys: [t2.number (#2)], probe keys: [t.number (#0)], join filters: []
         HashJoin: INNER, build keys: [t1.number (#1)], probe keys: [t.number (#0)], join filters: []
             Exchange(Hash): keys: [t.number (#0)]
-                Scan: default.system.numbers
+                Scan: default.system.numbers, filters: []
             Exchange(Hash): keys: [t1.number (#1)]
-                Scan: default.system.numbers
+                Scan: default.system.numbers, filters: []
         Exchange(Hash): keys: [t2.number (#2)]
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: []
 
 
 select * from (select number as a, number+1 as b from numbers(1)) t, numbers(2) t1, numbers(3) t2 where a = t1.number and b = t2.number
@@ -30,11 +30,11 @@ Exchange(Merge)
             HashJoin: INNER, build keys: [t1.number (#3)], probe keys: [t.a (#0)], join filters: []
                 Exchange(Hash): keys: [t.a (#0)]
                     EvalScalar: [+(numbers.number (#0), 1)]
-                        Scan: default.system.numbers
+                        Scan: default.system.numbers, filters: []
                 Exchange(Hash): keys: [t1.number (#3)]
-                    Scan: default.system.numbers
+                    Scan: default.system.numbers, filters: []
         Exchange(Hash): keys: [t2.number (#4)]
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: []
 
 
 select * from (select sum(number) as number from numbers(1) group by number) t, numbers(2) t1 where t.number = t1.number
@@ -47,8 +47,8 @@ Exchange(Merge)
                     Aggregate(Final): group items: [numbers.number (#0)], aggregate functions: [sum(number)]
                         Aggregate(Partial): group items: [numbers.number (#0)], aggregate functions: [sum(number)]
                             Exchange(Hash): keys: [numbers.number (#0)]
-                                Scan: default.system.numbers
+                                Scan: default.system.numbers, filters: []
         Exchange(Hash): keys: [t1.number (#4)]
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: []
 
 

query/tests/it/sql/optimizer/heuristic/testdata/join.test

@@ -3,16 +3,16 @@ select t.number from numbers(1) as t, numbers(1) as t1 where t.number = t1.numbe
 ----
 Project: [number (#0)]
     HashJoin: INNER, build keys: [t1.number (#1)], probe keys: [t.number (#0)], join filters: []
-        Scan: default.system.numbers
-        Scan: default.system.numbers
+        Scan: default.system.numbers, filters: []
+        Scan: default.system.numbers, filters: []
 
 
 select t.number from numbers(1) as t, numbers(1) as t1 where t.number = t1.number and t.number = t1.number + 1
 ----
 Project: [number (#0)]
     HashJoin: INNER, build keys: [t1.number (#1), +(t1.number (#1), 1)], probe keys: [t.number (#0), t.number (#0)], join filters: []
-        Scan: default.system.numbers
-        Scan: default.system.numbers
+        Scan: default.system.numbers, filters: []
+        Scan: default.system.numbers, filters: []
 
 
 # Push predicates down through join
@@ -21,18 +21,18 @@ select t.number from numbers(1) as t, numbers(1) as t1 where t.number > 1 and 1
 Project: [number (#0)]
     CrossJoin
         Filter: [t.number (#0) > 1]
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: [t.number (#0) > 1]
         Filter: [1 < t1.number (#1)]
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: [1 < t1.number (#1)]
 
 
 select t.number from numbers(1) as t, numbers(1) as t1 where t.number + t1.number = 1
 ----
 Project: [number (#0)]
     Filter: [+(t.number (#0), t1.number (#1)) = 1]
         CrossJoin
-            Scan: default.system.numbers
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: []
+            Scan: default.system.numbers, filters: []
 
 
 # Incompatible join keys, cannot push into join
@@ -41,8 +41,8 @@ select t.number from numbers(1) as t, numbers(1) as t1 where t.number = cast(t1.
 Project: [number (#0)]
     Filter: [t.number (#0) = CAST(t1.number (#1) AS VARCHAR)]
         CrossJoin
-            Scan: default.system.numbers
-            Scan: default.system.numbers
+            Scan: default.system.numbers, filters: []
+            Scan: default.system.numbers, filters: []
 
 
 # Join multiple tables
@@ -52,8 +52,8 @@ Project: [number (#0)]
     HashJoin: INNER, build keys: [t2.number (#2)], probe keys: [t1.number (#1)], join filters: []
         CrossJoin
             Filter: [t.number (#0) = 1]
-                Scan: default.system.numbers
-            Scan: default.system.numbers
-        Scan: default.system.numbers
+                Scan: default.system.numbers, filters: [t.number (#0) = 1]
+            Scan: default.system.numbers, filters: []
+        Scan: default.system.numbers, filters: []
 
 

query/tests/it/sql/optimizer/heuristic/testdata/prune_columns.test

@@ -7,7 +7,7 @@ Project: [number (#0)]
             EvalScalar: [numbers.a (#0)]
                 Project: [number (#0)]
                     EvalScalar: [numbers.number (#0)]
-                        Scan: default.system.numbers
+                        Scan: default.system.numbers, filters: []
 
 
 # Prune unused columns from Aggregate
@@ -19,7 +19,7 @@ Project: [number (#0)]
             EvalScalar: [group_item (#0)]
                 Aggregate(Initial): group items: [numbers.number (#0)], aggregate functions: []
                     EvalScalar: [numbers.number (#0)]
-                        Scan: default.system.numbers
+                        Scan: default.system.numbers, filters: []
 
 
 # Prune unused columns for simple plan nodes (Project, Filter, Aggregate...)
@@ -34,7 +34,7 @@ Limit: [1], Offset: [0]
                         EvalScalar: [group_item (#0)]
                             Aggregate(Initial): group items: [numbers.number (#0), numbers.number (#0), numbers.number (#0), numbers.number (#0)], aggregate functions: []
                                 EvalScalar: [numbers.number (#0), numbers.number (#0), numbers.number (#0), numbers.number (#0), numbers.number (#0)]
-                                    Scan: default.system.numbers
+                                    Scan: default.system.numbers, filters: []
 
 
 # Prune unused columns for join plan nodes (LogicalInnerJoin ...)
@@ -48,10 +48,10 @@ Project: [a (#1)]
                     CrossJoin
                         Project: [a (#1),b (#2),c (#3)]
                             EvalScalar: [+(numbers.number (#0), 1), +(numbers.number (#0), 1), +(numbers.number (#0), 1)]
-                                Scan: default.system.numbers
+                                Scan: default.system.numbers, filters: []
                         Project: [b (#11)]
                             EvalScalar: [+(numbers.number (#9), 1)]
-                                Scan: default.system.numbers
+                                Scan: default.system.numbers, filters: []
 
 
 # Prune unused columns for correlated query
@@ -63,7 +63,7 @@ Project: [a (#1)]
             HashJoin: SINGLE, build keys: [], probe keys: [], join filters: []
                 Project: [a (#1)]
                     EvalScalar: [+(numbers.number (#0), 1)]
-                        Scan: default.system.numbers
+                        Scan: default.system.numbers, filters: []
                 Project: [COUNT(*) (#21)]
                     EvalScalar: [COUNT(*) (#22)]
                         Aggregate(Initial): group items: [], aggregate functions: [COUNT(*)]
@@ -73,10 +73,10 @@ Project: [a (#1)]
                                         CrossJoin
                                             Project: [a (#6),b (#7),c (#8)]
                                                 EvalScalar: [+(numbers.number (#5), 1), +(numbers.number (#5), 1), +(numbers.number (#5), 1)]
-                                                    Scan: default.system.numbers
+                                                    Scan: default.system.numbers, filters: []
                                             Project: [b (#16)]
                                                 EvalScalar: [+(numbers.number (#14), 1)]
-                                                    Scan: default.system.numbers
+                                                    Scan: default.system.numbers, filters: []
 
 
 # Prune unused columns with order by
@@ -85,7 +85,7 @@ select name from system.functions order by example
 Project: [name (#0)]
     EvalScalar: [functions.name (#0)]
         Sort: [example (#7) ASC]
-            Scan: default.system.functions
+            Scan: default.system.functions, filters: []
 
 
 # Prune unused columns with cross join
@@ -95,13 +95,13 @@ Project: [number (#0)]
     EvalScalar: [t.number (#0)]
         Filter: [subquery_3 (#3)]
             CrossJoin
-                Scan: default.system.numbers
+                Scan: default.system.numbers, filters: []
                 Project: [subquery (#3)]
                     EvalScalar: [count(*) (#2) = 1]
                         Aggregate(Initial): group items: [], aggregate functions: [count(*)]
                             Limit: [1], Offset: [0]
                                 Project: [number (#1)]
                                     EvalScalar: [numbers.number (#1)]
-                                        Scan: default.system.numbers
+                                        Scan: default.system.numbers, filters: []
 
 

query/tests/it/sql/optimizer/heuristic/testdata/select.test

@@ -1,67 +1,67 @@
 select * from numbers(1)
 ----
-Scan: default.system.numbers
+Scan: default.system.numbers, filters: []
 
 
 select * from (select * from numbers(1)) as t1 where number = 1
 ----
 Filter: [t1.number (#0) = 1]
-    Scan: default.system.numbers
+    Scan: default.system.numbers, filters: [t1.number (#0) = 1]
 
 
 # `b = 1` can not be pushed down
 select * from (select number as a, number + 1 as b from numbers(1)) as t1 where a = 1 and b = 1
 ----
 Filter: [t1.a (#0) = 1, t1.b (#1) = 1]
     EvalScalar: [+(numbers.number (#0), 1)]
-        Scan: default.system.numbers
+        Scan: default.system.numbers, filters: []
 
 
 select * from (select number as a, number + 1 as b from numbers(1)) as t1 where a = 1
 ----
 EvalScalar: [+(numbers.number (#0), 1)]
     Filter: [t1.a (#0) = 1]
-        Scan: default.system.numbers
+        Scan: default.system.numbers, filters: [t1.a (#0) = 1]
 
 
 select * from numbers(1) where number = pow(1, 1 + 1)
 ----
 Filter: [numbers.number (#0) = 1]
-    Scan: default.system.numbers
+    Scan: default.system.numbers, filters: [numbers.number (#0) = 1]
 
 
 select * from numbers(1) where TRUE and 1 = 1
 ----
-Scan: default.system.numbers
+Scan: default.system.numbers, filters: []
 
 
 select * from numbers(1) where number = 0 and false
 ----
 Filter: [false]
-    Scan: default.system.numbers
+    Scan: default.system.numbers, filters: [false]
 
 
 select * from numbers(1) where number = 0 and null
 ----
 Filter: [false]
-    Scan: default.system.numbers
+    Scan: default.system.numbers, filters: [false]
 
 
 # If there is only one conjunction and the value is null, then we won't rewrite it
 select * from numbers(1) where null
 ----
 Filter: [NULL]
-    Scan: default.system.numbers
+    Scan: default.system.numbers, filters: [NULL]
 
 
 select a from (select number as a, number as b from numbers(1))
 ----
-Scan: default.system.numbers
+Scan: default.system.numbers, filters: []
 
 
 select a from (select number as a, number+1 as b from numbers(1))
 ----
 Project: [number (#0)]
-    Scan: default.system.numbers
+    Scan: default.system.numbers, filters: []