fix: Fix alias of target (#13030)

* fix target table alias

* add test

src/query/service/src/interpreters/interpreter_merge_into.rs

@@ -102,6 +102,7 @@ impl MergeIntoInterpreter {
             catalog,
             database,
             table,
+            target_alias,
             matched_evaluators,
             unmatched_evaluators,
             target_table_idx,
@@ -217,7 +218,10 @@ impl MergeIntoInterpreter {
                     selection: None,
                     subquery_desc: vec![],
                     database: database.clone(),
-                    table: table_name.clone(),
+                    table: match target_alias {
+                        None => table_name.clone(),
+                        Some(alias) => alias.name.to_string(),
+                    },
                     update_list: update_list.clone(),
                     bind_context: bind_context.clone(),
                     metadata: self.plan.meta_data.clone(),
@@ -236,6 +240,7 @@ impl MergeIntoInterpreter {
                         fuse_table.schema().into(),
                         col_indices,
                         Some(join_output_schema.num_fields()),
+                        target_alias.is_some(),
                     )?;
                 let update_list = update_list
                     .iter()

src/query/service/src/interpreters/interpreter_update.rs

@@ -157,6 +157,7 @@ impl Interpreter for UpdateInterpreter {
             tbl.schema().into(),
             col_indices.clone(),
             None,
+            false,
         )?;
 
         let computed_list = self

src/query/service/src/test_kits/table_test_fixture.rs

@@ -676,6 +676,7 @@ pub async fn do_update(
         table.schema().into(),
         col_indices.clone(),
         None,
+        false,
     )?;
     let computed_list =
         plan.generate_stored_computed_list(ctx.clone(), Arc::new(table.schema().into()))?;

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

@@ -300,6 +300,7 @@ impl Binder {
             catalog: catalog_name.to_string(),
             database: database_name.to_string(),
             table: table_name,
+            target_alias: target_alias.clone(),
             table_id,
             bind_context: Box::new(bind_ctx.clone()),
             meta_data: self.metadata.clone(),

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

@@ -15,6 +15,7 @@
 use std::collections::HashMap;
 use std::collections::HashSet;
 
+use common_ast::ast::TableAlias;
 use common_expression::DataSchemaRef;
 use common_expression::FieldIndex;
 use common_meta_types::MetaId;
@@ -47,6 +48,7 @@ pub struct MergeInto {
     pub catalog: String,
     pub database: String,
     pub table: String,
+    pub target_alias: Option<TableAlias>,
     pub table_id: MetaId,
     pub input: Box<SExpr>,
     pub bind_context: Box<BindContext>,

src/query/sql/src/planner/plans/update.rs

@@ -65,6 +65,7 @@ impl UpdatePlan {
         schema: DataSchema,
         col_indices: Vec<usize>,
         use_column_name_index: Option<usize>,
+        has_alias: bool,
     ) -> Result<Vec<(FieldIndex, RemoteExpr<String>)>> {
         let column = ColumnBindingBuilder::new(
             PREDICATE_COLUMN_NAME.to_string(),
@@ -94,7 +95,11 @@ impl UpdatePlan {
                     let mut right = None;
                     for column_binding in self.bind_context.columns.iter() {
                         if BindContext::match_column_binding(
-                            Some(&self.database),
+                            if has_alias {
+                                None
+                            } else {
+                                Some(&self.database)
+                            },
                             Some(&self.table),
                             field.name(),
                             column_binding,
@@ -106,6 +111,7 @@ impl UpdatePlan {
                             break;
                         }
                     }
+
                     let right = right.ok_or_else(|| ErrorCode::Internal("It's a bug"))?;
                     ScalarExpr::FunctionCall(FunctionCall {
                         span: None,

tests/sqllogictests/suites/base/09_fuse_engine/09_0026_merge_into

@@ -408,5 +408,33 @@ select * from t1 order by a,b,c;
 NULL b_1 c_1
 NULL b_3 c_3
 
+statement ok
+truncate table t1;
+
+statement ok
+truncate table t2;
+
+query I
+select count(*) from t1;
+----
+0
+
+query I
+select count(*) from t2;
+----
+0
+
+## test target table alias
+statement ok
+insert into t2 values(1,'a1','b1');
+
+statement ok
+merge into t1 as t3 using (select * from t2 as t2) on t3.a = t2.a  when not matched  then insert (a,b,c) values(t2.a,t2.b,t2.c);
+
+query TTT
+select * from t1 order by a,b,c;
+----
+1 a1 b1
+
 statement ok
 set enable_experimental_merge_into = 0;