fix: push filter down join with materialized cte (#16349)

databendlabs · Aug 29, 2024 · 44e6b1b · 44e6b1b
1 parent a5c4129
commit 44e6b1b
Show file tree

Hide file tree

Showing 2 changed files with 80 additions and 27 deletions.
diff --git a/src/query/sql/src/planner/binder/scalar_common.rs b/src/query/sql/src/planner/binder/scalar_common.rs
@@ -94,7 +94,7 @@ pub fn split_equivalent_predicate(scalar: &ScalarExpr) -> Option<(ScalarExpr, Sc
 }
 
 pub fn satisfied_by(scalar: &ScalarExpr, prop: &RelationalProperty) -> bool {
-    scalar.used_columns().is_subset(&prop.output_columns)
+    scalar.used_columns().is_subset(&prop.output_columns) && !scalar.used_columns().is_empty()
 }
 
 /// Helper to determine join condition type from a scalar expression.
@@ -128,38 +128,40 @@ impl<'a> JoinPredicate<'a> {
             return Self::ALL(scalar);
         }
 
-        if satisfied_by(scalar, left_prop) {
-            return Self::Left(scalar);
-        }
-
-        if satisfied_by(scalar, right_prop) {
-            return Self::Right(scalar);
-        }
-
         if let ScalarExpr::FunctionCall(func) = scalar {
-            if func.arguments.len() != 2 {
+            if func.arguments.len() > 2 {
                 return Self::Other(scalar);
             }
 
-            let is_equal_op = func.func_name.as_str() == "eq";
-            let left = &func.arguments[0];
-            let right = &func.arguments[1];
-
-            if satisfied_by(left, left_prop) && satisfied_by(right, right_prop) {
-                return Self::Both {
-                    left,
-                    right,
-                    is_equal_op,
-                };
+            if func.arguments.len() == 2 {
+                let is_equal_op = func.func_name.as_str() == "eq";
+                let left = &func.arguments[0];
+                let right = &func.arguments[1];
+
+                if satisfied_by(left, left_prop) && satisfied_by(right, right_prop) {
+                    return Self::Both {
+                        left,
+                        right,
+                        is_equal_op,
+                    };
+                }
+
+                if satisfied_by(right, left_prop) && satisfied_by(left, right_prop) {
+                    return Self::Both {
+                        left: right,
+                        right: left,
+                        is_equal_op,
+                    };
+                }
             }
+        }
 
-            if satisfied_by(right, left_prop) && satisfied_by(left, right_prop) {
-                return Self::Both {
-                    left: right,
-                    right: left,
-                    is_equal_op,
-                };
-            }
+        if satisfied_by(scalar, left_prop) {
+            return Self::Left(scalar);
+        }
+
+        if satisfied_by(scalar, right_prop) {
+            return Self::Right(scalar);
         }
 
         Self::Other(scalar)

diff --git a/...standalone/explain/push_down_filter/push_down_filter_join/push_down_filter_self_join.test b/...standalone/explain/push_down_filter/push_down_filter_join/push_down_filter_self_join.test
@@ -0,0 +1,51 @@
+statement ok
+create table t1 as select number as a, number as b from numbers(10);
+
+statement ok
+create table t2 as select number as a, number as b from numbers(10);
+
+query T
+explain with A as  materialized (select * from t1 union all select * from t2),
+B as (select * from A),
+C as (select * from B as b1 left outer join B as b2 on b1.a = b2.a where b1.b < b2.b),
+D as (select * from C)
+select * from D;
+----
+MaterializedCTE
+├── output columns: [t1.a (#0), t1.b (#1)]
+├── UnionAll
+│   ├── output columns: [t1.a (#0), t1.b (#1)]
+│   ├── estimated rows: 20.00
+│   ├── TableScan
+│   │   ├── table: default.default.t1
+│   │   ├── output columns: [a (#0), b (#1)]
+│   │   ├── read rows: 10
+│   │   ├── read size: < 1 KiB
+│   │   ├── partitions total: 1
+│   │   ├── partitions scanned: 1
+│   │   ├── pruning stats: [segments: <range pruning: 1 to 1>, blocks: <range pruning: 1 to 1>]
+│   │   ├── push downs: [filters: [], limit: NONE]
+│   │   └── estimated rows: 10.00
+│   └── TableScan
+│       ├── table: default.default.t2
+│       ├── output columns: [a (#2), b (#3)]
+│       ├── read rows: 10
+│       ├── read size: < 1 KiB
+│       ├── partitions total: 1
+│       ├── partitions scanned: 1
+│       ├── pruning stats: [segments: <range pruning: 1 to 1>, blocks: <range pruning: 1 to 1>]
+│       ├── push downs: [filters: [], limit: NONE]
+│       └── estimated rows: 10.00
+└── HashJoin
+    ├── output columns: [t1.a (#0), t1.b (#1)]
+    ├── join type: INNER
+    ├── build keys: [b2.a (#0)]
+    ├── probe keys: [b1.a (#0)]
+    ├── filters: [b1.b (#1) < b2.b (#1)]
+    ├── estimated rows: 10.00
+    ├── CTEScan(Build)
+    │   ├── CTE index: 0, sub index: 2
+    │   └── estimated rows: 10.00
+    └── CTEScan(Probe)
+        ├── CTE index: 0, sub index: 1
+        └── estimated rows: 10.00