fix

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

@@ -425,10 +425,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

@@ -392,6 +392,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() {
+            mreturn 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_0036_merge_into_without_distributed_enable.test

@@ -847,10 +847,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;