chore(planner): fix distributed merge into (#16228)

* chore(planner): fix distributed merge into

* chore(code): make lint

* chore(planner): add is_broadcast check for BroadcastToShuffleOptimizer

* chore(test): fix sqllogictest

* chore(test): add distributed merge into test

* chore(test): change to distributed plan

* chore(test): remove explain

Author: Dousir9

src/query/settings/src/settings_default.rs

@@ -296,12 +296,6 @@ impl DefaultSettings {
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=u64::MAX)),
                 }),
-                ("disable_merge_into_join_reorder", DefaultSettingValue {
-                    value: UserSettingValue::UInt64(0),
-                    desc: "Disable merge into join reorder optimization.",
-                    mode: SettingMode::Both,
-                    range: Some(SettingRange::Numeric(0..=1)),
-                }),
                 ("enable_merge_into_row_fetch", DefaultSettingValue {
                     value: UserSettingValue::UInt64(1),
                     desc: "Enable merge into row fetch optimization.",
@@ -356,6 +350,12 @@ impl DefaultSettings {
                     mode: SettingMode::Both,
                     range: Some(SettingRange::Numeric(0..=1)),
                 }),
+                ("enforce_shuffle_join", DefaultSettingValue {
+                    value: UserSettingValue::UInt64(0),
+                    desc: "Enforce shuffle join.",
+                    mode: SettingMode::Both,
+                    range: Some(SettingRange::Numeric(0..=1)),
+                }),
                 ("storage_fetch_part_num", DefaultSettingValue {
                     value: UserSettingValue::UInt64(2),
                     desc: "Sets the number of partitions that are fetched in parallel from storage during query execution.",

src/query/settings/src/settings_getter_setter.rs

@@ -302,8 +302,8 @@ impl Settings {
         Ok(self.try_get_u64("enforce_broadcast_join")? != 0)
     }
 
-    pub fn get_disable_merge_into_join_reorder(&self) -> Result<bool> {
-        Ok(self.try_get_u64("disable_merge_into_join_reorder")? != 0)
+    pub fn get_enforce_shuffle_join(&self) -> Result<bool> {
+        Ok(self.try_get_u64("enforce_shuffle_join")? != 0)
     }
 
     pub fn get_enable_merge_into_row_fetch(&self) -> Result<bool> {

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

@@ -115,7 +115,7 @@ impl PhysicalPlanBuilder {
             distributed,
             predicate_column_index,
             row_id_index,
-            change_join_order,
+            row_id_shuffle,
             can_try_update_column_only,
             ..
         } = mutation;
@@ -230,9 +230,7 @@ impl PhysicalPlanBuilder {
         // For distributed merge, we shuffle data blocks by block_id (drived from row_id) to avoid
         // different nodes update the same physical block simultaneously, data blocks that are needed
         // to insert just keep in local node.
-        let source_is_broadcast =
-            matches!(strategy, MutationStrategy::MatchedOnly) && !change_join_order;
-        if *distributed && !is_not_matched_only && !source_is_broadcast {
+        if *distributed && *row_id_shuffle && !is_not_matched_only {
             plan = PhysicalPlan::Exchange(build_block_id_shuffle_exchange(
                 plan,
                 bind_context,

src/query/sql/src/planner/binder/bind_mutation/bind.rs

@@ -306,7 +306,7 @@ impl Binder {
             field_index_map,
             strategy: mutation_strategy.clone(),
             distributed: false,
-            change_join_order: false,
+            row_id_shuffle: true,
             row_id_index: target_table_row_id_index,
             can_try_update_column_only: self.can_try_update_column_only(&matched_clauses),
             lock_guard,

src/query/sql/src/planner/binder/bind_mutation/merge.rs

@@ -30,9 +30,9 @@ use crate::plans::Plan;
 use crate::BindContext;
 
 // Merge into strategies:
-// 1. Insert only: RIGHT ANTI join.
-// 2. Matched and unmatched: RIGHT OUTER join.
-// 3. Matched only: INNER join.
+// 1. Insert only: target right-anti join source.
+// 2. Matched and unmatched: target right-outer join source.
+// 3. Matched only: target inner join source.
 impl Binder {
     #[allow(warnings)]
     #[async_backtrace::framed]

src/query/sql/src/planner/binder/bind_mutation/mod.rs

@@ -20,4 +20,4 @@ mod update;
 
 pub use bind::MutationStrategy;
 pub use bind::MutationType;
-pub use mutation_expression::target_table_position;
+pub use mutation_expression::target_probe;

src/query/sql/src/planner/binder/bind_mutation/mutation_expression.rs

@@ -18,9 +18,7 @@ use std::sync::Arc;
 
 use databend_common_ast::ast::Expr;
 use databend_common_ast::ast::JoinCondition;
-use databend_common_ast::ast::JoinOperator::Inner;
-use databend_common_ast::ast::JoinOperator::RightAnti;
-use databend_common_ast::ast::JoinOperator::RightOuter;
+use databend_common_ast::ast::JoinOperator;
 use databend_common_ast::ast::TableReference;
 use databend_common_catalog::plan::InternalColumn;
 use databend_common_catalog::plan::InternalColumnType;
@@ -164,9 +162,9 @@ impl MutationExpression {
                         target_s_expr,
                         source_s_expr,
                         match mutation_strategy {
-                            MutationStrategy::MatchedOnly => Inner,
-                            MutationStrategy::NotMatchedOnly => RightAnti,
-                            MutationStrategy::MixedMatched => RightOuter,
+                            MutationStrategy::MatchedOnly => JoinOperator::Inner,
+                            MutationStrategy::NotMatchedOnly => JoinOperator::RightAnti,
+                            MutationStrategy::MixedMatched => JoinOperator::RightOuter,
                             MutationStrategy::Direct => unreachable!(),
                         },
                         JoinCondition::On(Box::new(match_expr.clone())),
@@ -568,10 +566,11 @@ pub struct MutationExpressionBindResult {
     pub direct_filter: Option<ScalarExpr>,
 }
 
-pub fn target_table_position(s_expr: &SExpr, target_table_index: usize) -> Result<usize> {
+pub fn target_probe(s_expr: &SExpr, target_table_index: usize) -> Result<bool> {
     if !matches!(s_expr.plan(), RelOperator::Join(_)) {
-        return Ok(0);
+        return Ok(false);
     }
+
     fn contains_target_table(s_expr: &SExpr, target_table_index: usize) -> bool {
         if let RelOperator::Scan(ref scan) = s_expr.plan() {
             scan.table_index == target_table_index
@@ -582,9 +581,5 @@ pub fn target_table_position(s_expr: &SExpr, target_table_index: usize) -> Resul
         }
     }
 
-    if contains_target_table(s_expr.child(0)?, target_table_index) {
-        Ok(0)
-    } else {
-        Ok(1)
-    }
+    Ok(contains_target_table(s_expr.child(0)?, target_table_index))
 }

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

@@ -57,7 +57,7 @@ mod window;
 
 pub use aggregate::AggregateInfo;
 pub use bind_context::*;
-pub use bind_mutation::target_table_position;
+pub use bind_mutation::target_probe;
 pub use bind_mutation::MutationStrategy;
 pub use bind_mutation::MutationType;
 pub use bind_query::bind_values;

src/query/sql/src/planner/optimizer/distributed/distributed_merge.rs

@@ -18,21 +18,29 @@ use databend_common_exception::Result;
 
 use crate::optimizer::extract::Matcher;
 use crate::optimizer::SExpr;
-use crate::plans::Exchange::Hash;
+use crate::plans::Exchange;
 use crate::plans::Join;
 use crate::plans::RelOp;
 use crate::plans::RelOperator;
-pub struct MergeOptimizer {
-    pub merge_matcher: Matcher,
+pub struct BroadcastToShuffleOptimizer {
+    pub matcher: Matcher,
 }
 
-impl MergeOptimizer {
+impl BroadcastToShuffleOptimizer {
     pub fn create() -> Self {
         Self {
-            merge_matcher: Self::merge_matcher(),
+            matcher: Self::matcher(),
         }
     }
 
+    pub fn is_broadcast(&self, s_expr: &SExpr) -> Result<bool> {
+        let right = s_expr.child(1)?;
+        Ok(matches!(
+            right.plan(),
+            RelOperator::Exchange(Exchange::Broadcast)
+        ))
+    }
+
     pub fn optimize(&self, s_expr: &SExpr) -> Result<SExpr> {
         let left_exchange_input = s_expr.child(0)?;
 
@@ -51,11 +59,11 @@ impl MergeOptimizer {
 
         let new_join_children = vec![
             Arc::new(SExpr::create_unary(
-                Arc::new(RelOperator::Exchange(Hash(left_conditions))),
+                Arc::new(RelOperator::Exchange(Exchange::Hash(left_conditions))),
                 Arc::new(left_exchange_input.clone()),
             )),
             Arc::new(SExpr::create_unary(
-                Arc::new(RelOperator::Exchange(Hash(right_conditions))),
+                Arc::new(RelOperator::Exchange(Exchange::Hash(right_conditions))),
                 Arc::new(right_exchange_input.clone()),
             )),
         ];
@@ -64,7 +72,7 @@ impl MergeOptimizer {
         Ok(join_s_expr)
     }
 
-    fn merge_matcher() -> Matcher {
+    fn matcher() -> Matcher {
         // Input:
         //         Join
         //         /  \

src/query/sql/src/planner/optimizer/distributed/mod.rs

@@ -18,5 +18,5 @@ mod distributed_merge;
 mod sort_and_limit;
 
 pub use distributed::optimize_distributed_query;
-pub use distributed_merge::MergeOptimizer;
+pub use distributed_merge::BroadcastToShuffleOptimizer;
 pub use sort_and_limit::SortAndLimitPushDownOptimizer;