feat(planner): improve push down filter join

src/query/service/src/pipelines/processors/transforms/hash_join/desc.rs

@@ -36,7 +36,7 @@ pub struct HashJoinDesc {
     pub(crate) build_keys: Vec<Expr>,
     pub(crate) probe_keys: Vec<Expr>,
     pub(crate) join_type: JoinType,
-    pub(crate) original_join_type: Option<JoinType>,
+    pub(crate) single_to_inner: Option<JoinType>,
     /// when we have non-equal conditions for hash join,
     /// for example `a = b and c = d and e > f`, we will use `and_filters`
     /// to wrap `e > f` as a other_predicate to do next step's check.
@@ -88,7 +88,7 @@ impl HashJoinDesc {
             from_correlated_subquery: join.from_correlated_subquery,
             probe_keys_rt,
             broadcast: join.broadcast,
-            original_join_type: join.original_join_type.clone(),
+            single_to_inner: join.single_to_inner.clone(),
             enable_bloom_runtime_filter: join.enable_bloom_runtime_filter,
         })
     }

src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_build_state.rs

@@ -246,7 +246,7 @@ impl HashJoinBuildState {
     pub(crate) fn add_build_block(&self, data_block: DataBlock) -> Result<()> {
         let block_outer_scan_map = if self.hash_join_state.need_outer_scan()
             || matches!(
-                self.hash_join_state.hash_join_desc.original_join_type,
+                self.hash_join_state.hash_join_desc.single_to_inner,
                 Some(JoinType::RightSingle)
             ) {
             vec![false; data_block.num_rows()]
@@ -266,7 +266,7 @@ impl HashJoinBuildState {
             let build_state = unsafe { &mut *self.hash_join_state.build_state.get() };
             if self.hash_join_state.need_outer_scan()
                 || matches!(
-                    self.hash_join_state.hash_join_desc.original_join_type,
+                    self.hash_join_state.hash_join_desc.single_to_inner,
                     Some(JoinType::RightSingle)
                 )
             {

src/query/service/src/pipelines/processors/transforms/hash_join/result_blocks.rs

@@ -56,7 +56,7 @@ impl HashJoinProbeState {
             .other_predicate
             .is_none();
         match self.hash_join_state.hash_join_desc.join_type {
-            JoinType::Inner => match self.hash_join_state.hash_join_desc.original_join_type {
+            JoinType::Inner => match self.hash_join_state.hash_join_desc.single_to_inner {
                 Some(JoinType::LeftSingle) => {
                     self.inner_join::<_, true, false>(input, keys, hash_table, probe_state)
                 }

src/query/service/src/schedulers/fragments/fragmenter.rs

@@ -249,7 +249,7 @@ impl PhysicalPlanReplacer for Fragmenter {
             probe_keys_rt: plan.probe_keys_rt.clone(),
             enable_bloom_runtime_filter: plan.enable_bloom_runtime_filter,
             broadcast: plan.broadcast,
-            original_join_type: plan.original_join_type.clone(),
+            single_to_inner: plan.single_to_inner.clone(),
         }))
     }
 

src/query/sql/src/executor/physical_plan_visitor.rs

@@ -242,7 +242,7 @@ pub trait PhysicalPlanReplacer {
             probe_keys_rt: plan.probe_keys_rt.clone(),
             enable_bloom_runtime_filter: plan.enable_bloom_runtime_filter,
             broadcast: plan.broadcast,
-            original_join_type: plan.original_join_type.clone(),
+            single_to_inner: plan.single_to_inner.clone(),
         }))
     }
 

src/query/sql/src/executor/physical_plans/physical_hash_join.rs

@@ -82,9 +82,9 @@ pub struct HashJoin {
     pub enable_bloom_runtime_filter: bool,
     // Under cluster, mark if the join is broadcast join.
     pub broadcast: bool,
-    // Original join type. Left/Right single join may be convert to inner join
-    // Record the original join type and do some special processing during runtime.
-    pub original_join_type: Option<JoinType>,
+    // When left/right single join converted to inner join, record the original join type
+    // and do some special processing during runtime.
+    pub single_to_inner: Option<JoinType>,
 }
 
 impl HashJoin {
@@ -510,7 +510,7 @@ impl PhysicalPlanBuilder {
             need_hold_hash_table: join.need_hold_hash_table,
             stat_info: Some(stat_info),
             broadcast: is_broadcast,
-            original_join_type: join.original_join_type.clone(),
+            single_to_inner: join.single_to_inner.clone(),
             enable_bloom_runtime_filter: adjust_bloom_runtime_filter(
                 self.ctx.clone(),
                 &self.metadata,

src/query/sql/src/lib.rs

@@ -19,6 +19,7 @@
 #![feature(try_blocks)]
 #![feature(extend_one)]
 #![feature(lazy_cell)]
+#![feature(if_let_guard)]
 
 pub mod evaluator;
 pub mod executor;

src/query/sql/src/planner/binder/join.rs

@@ -242,7 +242,7 @@ impl Binder {
             from_correlated_subquery: false,
             need_hold_hash_table: false,
             is_lateral,
-            original_join_type: None,
+            single_to_inner: None,
         };
         Ok(SExpr::create_binary(
             Arc::new(logical_join.into()),

src/query/sql/src/planner/binder/scalar_common.rs

@@ -124,10 +124,6 @@ impl<'a> JoinPredicate<'a> {
         left_prop: &RelationalProperty,
         right_prop: &RelationalProperty,
     ) -> Self {
-        if contain_subquery(scalar) {
-            return Self::Other(scalar);
-        }
-
         if scalar.used_columns().is_empty() {
             return Self::ALL(scalar);
         }

src/query/sql/src/planner/optimizer/cascades/tasks/apply_rule.rs

@@ -63,7 +63,7 @@ impl ApplyRuleTask {
         let group = optimizer.memo.group(self.target_group_index)?;
         let m_expr = group.m_expr(self.m_expr_index)?;
         let mut state = TransformResult::new();
-        let rule = RuleFactory::create_rule(self.rule_id, optimizer.metadata.clone(), false)?;
+        let rule = RuleFactory::create_rule(self.rule_id, optimizer.metadata.clone())?;
         m_expr.apply_rule(&optimizer.memo, &rule, &mut state)?;
         optimizer.insert_from_transform_state(self.target_group_index, state)?;
 

src/query/sql/src/planner/optimizer/decorrelate/decorrelate.rs

@@ -200,7 +200,7 @@ impl SubqueryRewriter {
             from_correlated_subquery: true,
             need_hold_hash_table: false,
             is_lateral: false,
-            original_join_type: None,
+            single_to_inner: None,
         };
 
         // Rewrite plan to semi-join.
@@ -290,7 +290,7 @@ impl SubqueryRewriter {
                     from_correlated_subquery: true,
                     need_hold_hash_table: false,
                     is_lateral: false,
-                    original_join_type: None,
+                    single_to_inner: None,
                 };
                 let s_expr = SExpr::create_binary(
                     Arc::new(join_plan.into()),
@@ -339,7 +339,7 @@ impl SubqueryRewriter {
                     from_correlated_subquery: true,
                     need_hold_hash_table: false,
                     is_lateral: false,
-                    original_join_type: None,
+                    single_to_inner: None,
                 };
                 let s_expr = SExpr::create_binary(
                     Arc::new(join_plan.into()),
@@ -403,7 +403,7 @@ impl SubqueryRewriter {
                     from_correlated_subquery: true,
                     need_hold_hash_table: false,
                     is_lateral: false,
-                    original_join_type: None,
+                    single_to_inner: None,
                 }
                 .into();
                 Ok((

src/query/sql/src/planner/optimizer/decorrelate/flatten_plan.rs

@@ -136,7 +136,7 @@ impl SubqueryRewriter {
                 from_correlated_subquery: false,
                 need_hold_hash_table: false,
                 is_lateral: false,
-                original_join_type: None,
+                single_to_inner: None,
             }
             .into();
 
@@ -469,7 +469,7 @@ impl SubqueryRewriter {
                     from_correlated_subquery: false,
                     need_hold_hash_table: false,
                     is_lateral: false,
-                    original_join_type: None,
+                    single_to_inner: None,
                 }
                 .into(),
             ),

src/query/sql/src/planner/optimizer/decorrelate/subquery_rewriter.rs

@@ -483,7 +483,7 @@ impl SubqueryRewriter {
                     from_correlated_subquery: false,
                     need_hold_hash_table: false,
                     is_lateral: false,
-                    original_join_type: None,
+                    single_to_inner: None,
                 }
                 .into();
                 Ok((
@@ -553,7 +553,7 @@ impl SubqueryRewriter {
                     from_correlated_subquery: false,
                     need_hold_hash_table: false,
                     is_lateral: false,
-                    original_join_type: None,
+                    single_to_inner: None,
                 }
                 .into();
                 let s_expr = SExpr::create_binary(
@@ -584,7 +584,7 @@ impl SubqueryRewriter {
             from_correlated_subquery: false,
             need_hold_hash_table: false,
             is_lateral: false,
-            original_join_type: None,
+            single_to_inner: None,
         }
         .into();