fix: The update in MERGE INTO was mistakenly optimized as a delete (#17571)

* fix: The update in MERGE INTO was mistakenly optimized as a delete

* fix and add logic test

Author: SkyFan2002

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

@@ -47,6 +47,7 @@ use crate::optimizer::RuleID;
 use crate::optimizer::SExpr;
 use crate::optimizer::DEFAULT_REWRITE_RULES;
 use crate::planner::query_executor::QueryExecutor;
+use crate::plans::ConstantTableScan;
 use crate::plans::CopyIntoLocationPlan;
 use crate::plans::Join;
 use crate::plans::JoinType;
@@ -558,20 +559,28 @@ async fn optimize_mutation(mut opt_ctx: OptimizerContext, s_expr: SExpr) -> Resu
     if !mutation.matched_evaluators.is_empty() {
         match inner_rel_op {
             RelOp::ConstantTableScan => {
-                mutation.no_effect = true;
+                let constant_table_scan = ConstantTableScan::try_from(input_s_expr.plan().clone())?;
+                if constant_table_scan.num_rows == 0 {
+                    mutation.no_effect = true;
+                }
             }
             RelOp::Join => {
-                let right_child = input_s_expr.child(1)?;
+                let mut right_child = input_s_expr.child(1)?;
                 let mut right_child_rel = right_child.plan.rel_op();
                 if right_child_rel == RelOp::Exchange {
                     right_child_rel = right_child.child(0)?.plan.rel_op();
+                    right_child = right_child.child(0)?;
                 }
                 if right_child_rel == RelOp::ConstantTableScan {
-                    mutation.matched_evaluators = vec![MatchedEvaluator {
-                        condition: None,
-                        update: None,
-                    }];
-                    mutation.can_try_update_column_only = false;
+                    let constant_table_scan =
+                        ConstantTableScan::try_from(right_child.plan().clone())?;
+                    if constant_table_scan.num_rows == 0 {
+                        mutation.matched_evaluators = vec![MatchedEvaluator {
+                            condition: None,
+                            update: None,
+                        }];
+                        mutation.can_try_update_column_only = false;
+                    }
                 }
             }
             _ => (),

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

@@ -0,0 +1,18 @@
+statement ok
+create or replace table test_merge(col1 varchar,col2 varchar,col3 varchar);
+
+statement ok
+insert into test_merge values(2,'abc',2),(3,'abc',3),(4,'abc',4);
+
+query II
+merge into test_merge as tba using (select * from (values('1','add','11'),('4','add','44'))) as tbb("col1","col2","col3") on tba.col1=tbb.col1 WHEN MATCHED THEN update set tba.col1=tbb.col1 ,tba.col2='update',tba.col3=tbb.col3;
+----
+1
+
+query ITI
+select * from test_merge;
+----
+2 abc 2
+3 abc 3
+4 update 44
+