fix

Author: xudong963

src/query/service/src/interpreters/interpreter_explain.rs

@@ -218,7 +218,7 @@ impl Interpreter for ExplainInterpreter {
                     )
                     .await?
                 }
-                Plan::MergeInto(merge_into) => self.explain_merge_fragments(&merge_into).await?,
+                Plan::MergeInto(merge_into) => self.explain_merge_fragments(merge_into).await?,
                 Plan::Update(update) => self.explain_update_fragments(update.as_ref()).await?,
                 _ => {
                     return Err(ErrorCode::Unimplemented("Unsupported EXPLAIN statement"));

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

@@ -483,10 +483,10 @@ async fn optimize_merge_into(mut opt_ctx: OptimizerContext, plan: Box<MergeInto>
             join_s_expr = merge_source_optimizer.optimize(&join_s_expr)?;
             enable_right_broadcast = true;
         }
-
+        let distributed = !join_s_expr.has_merge_exchange();
         Ok(Plan::MergeInto(Box::new(MergeInto {
             input: Box::new(join_s_expr),
-            distributed: true,
+            distributed,
             change_join_order,
             columns_set: new_columns_set.clone(),
             enable_right_broadcast,

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

@@ -394,6 +394,13 @@ impl SExpr {
             .collect::<Vec<_>>();
         self.children = children;
     }
+
+    pub fn has_merge_exchange(&self) -> bool {
+        if let RelOperator::Exchange(Exchange::Merge) = self.plan.as_ref() {
+            return true;
+        }
+        self.children.iter().any(|child| child.has_merge_exchange())
+    }
 }
 
 fn find_subquery(rel_op: &RelOperator) -> bool {

tests/sqllogictests/suites/base/09_fuse_engine/09_0035_merge_into_separate_pipeline_without_distributed_enable.test

@@ -149,7 +149,7 @@ merge into t1_separate as t1 using (select * from t2_separate) as t2 on t1.a = t
 2 4
 
 query T
-select count(*) from fuse_block('default','t1_separate');
+select count(*) from fuse_block('db','t1_separate');
 ----
 1
 

tests/sqllogictests/suites/base/09_fuse_engine/09_0036_merge_into_without_distributed_enable.test

@@ -874,10 +874,10 @@ insert into t11 values (1, 10),(2, 20),(3, 30),(4, 40);
 statement ok
 insert into t12 values (1, 10),(2, 20),(3, 30),(4, 40);
 
-#query T
-#MERGE INTO t11 USING(SELECT NULL AS c0 FROM t12) AS t12 ON (t11.a OR TRUE) WHEN MATCHED AND TRUE THEN DELETE;
-#----
-#4
+query T
+MERGE INTO t11 USING(SELECT NULL AS c0 FROM t12) AS t12 ON (t11.a OR TRUE) WHEN MATCHED AND TRUE THEN DELETE;
+----
+4
 
 query T
 select count(*) from t11;

tests/sqllogictests/suites/mode/cluster/merge_into_non_equal_distributed.test

@@ -58,44 +58,42 @@ explain merge into t1 using t2 on t1.a < t2.a when matched then update * when no
 ----
 MergeInto:
 target_table: default.default.t1
-├── distributed: false
+├── distributed: true
 ├── target_build_optimization: false
 ├── can_try_update_column_only: true
 ├── matched update: [condition: None,update set a = a (#0)]
 └── unmatched insert: [condition: None,insert into (a) values(CAST(a (#0) AS Int32 NULL))]
 HashJoin
-├── output columns: [t1.a (#1), t1._row_id (#2), t2.a (#0)]
+├── output columns: [t1.a (#1), t1._row_id (#2), t2.a (#0), #_row_number]
 ├── join type: RIGHT OUTER
 ├── build keys: []
 ├── probe keys: []
 ├── filters: [t1.a (#1) < t2.a (#0)]
 ├── estimated rows: 15.00
 ├── Exchange(Build)
-│   ├── output columns: [t2.a (#0)]
-│   ├── exchange type: Merge
-│   └── TableScan
-│       ├── table: default.default.t2
-│       ├── output columns: [a (#0)]
-│       ├── read rows: 1
-│       ├── 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: 1.00
-└── Exchange(Probe)
-    ├── output columns: [t1.a (#1), t1._row_id (#2)]
-    ├── exchange type: Merge
-    └── TableScan
-        ├── table: default.default.t1
-        ├── output columns: [a (#1), _row_id (#2)]
-        ├── read rows: 15
-        ├── read size: < 1 KiB
-        ├── partitions total: 3
-        ├── partitions scanned: 3
-        ├── pruning stats: [segments: <range pruning: 3 to 3>, blocks: <range pruning: 3 to 3>]
-        ├── push downs: [filters: [], limit: NONE]
-        └── estimated rows: 15.00
+│   ├── output columns: [t2.a (#0), #_row_number]
+│   ├── exchange type: Broadcast
+│   └── MergeIntoAddRowNumber
+│       └── TableScan
+│           ├── table: default.default.t2
+│           ├── output columns: [a (#0)]
+│           ├── read rows: 1
+│           ├── 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: 1.00
+└── TableScan(Probe)
+    ├── table: default.default.t1
+    ├── output columns: [a (#1), _row_id (#2)]
+    ├── read rows: 15
+    ├── read size: < 1 KiB
+    ├── partitions total: 3
+    ├── partitions scanned: 3
+    ├── pruning stats: [segments: <range pruning: 3 to 3>, blocks: <range pruning: 3 to 3>]
+    ├── push downs: [filters: [], limit: NONE]
+    └── estimated rows: 15.00
 
 query TT
 merge into t1 using t2 on t1.a < t2.a when matched then update * when not matched then insert *;
@@ -173,6 +171,15 @@ HashJoin
 │   ├── output columns: [stage._$1 (#0), #_row_number]
 │   ├── exchange type: Broadcast
 │   └── MergeIntoAddRowNumber
+│       └── TableScan
+│           ├── table: default.system.stage
+│           ├── output columns: [_$1 (#0)]
+│           ├── read rows: 6
+│           ├── read size: < 1 KiB
+│           ├── partitions total: 1
+│           ├── partitions scanned: 1
+│           ├── push downs: [filters: [], limit: NONE]
+│           └── estimated rows: 0.00
 └── TableScan(Probe)
     ├── table: default.default.t1
     ├── output columns: [a (#1), _row_id (#2)]