fix(storage): change tracking with alias update row_version bug in merge into (#15412)

* fix change tracking with alias update row_version bug

* Update src/query/sql/src/planner/binder/merge_into.rs

* add test

* fix test

---------

Co-authored-by: dantengsky <dantengsky@gmail.com>

Author: zhyass

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

@@ -365,13 +365,28 @@ impl Binder {
             columns_set = columns_set.union(&join_column_set).cloned().collect();
         }
 
+        // If the table alias is not None, after the binding phase, the bound columns will have
+        // a database of 'None' and the table named as the alias.
+        // Thus, we adjust them accordingly.
+        let target_name = if let Some(target_identify) = target_alias {
+            normalize_identifier(&target_identify.name, &self.name_resolution_ctx)
+                .name
+                .clone()
+        } else {
+            table_name.clone()
+        };
+
         let has_update = self.has_update(&matched_clauses);
         let update_row_version = if table.change_tracking_enabled() && has_update {
             Some(Self::update_row_version(
                 table.schema_with_stream(),
                 &bind_ctx.columns,
-                Some(&database_name),
-                Some(&table_name),
+                if target_alias.is_none() {
+                    Some(&database_name)
+                } else {
+                    None
+                },
+                Some(&target_name),
             )?)
         } else {
             None
@@ -400,14 +415,6 @@ impl Binder {
             }
         }
 
-        let target_name = if let Some(target_identify) = target_alias {
-            normalize_identifier(&target_identify.name, &self.name_resolution_ctx)
-                .name
-                .clone()
-        } else {
-            table_name.clone()
-        };
-
         // bind matched clause columns and add update fields and exprs
         for clause in &matched_clauses {
             matched_evaluators.push(

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

@@ -195,6 +195,33 @@ drop table t3 all
 # end of issue 14955 #
 ######################
 
+###############
+# issue 15412 #
+###############
+
+statement ok
+CREATE TABLE test_select (id Int64, name STRING) change_tracking = true
+
+statement ok
+insert into test_select values(1, 'Wu')
+
+query I
+merge into db1.test_select t1 using (select * from db1.test_select) t2 on t1.id = t2.id when matched then update set t1.id=t2.id, t1.name='{}'
+----
+1
+
+query ITI
+select id, name, _row_version from test_select
+----
+1 {} 1
+
+statement ok
+drop table test_select all
+
+######################
+# end of issue 15412 #
+######################
+
 statement ok
 set enable_experimental_merge_into = 0