fix: fix name with ` (databendlabs#13163)

* remove quote

* suggestion by andylok

* rename

* suggestion

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

@@ -22,7 +22,9 @@ use common_ast::ast::QualifiedName;
 use common_ast::ast::SelectTarget;
 use common_ast::parser::parse_expr;
 use common_ast::parser::tokenize_sql;
+use common_ast::walk_expr_mut;
 use common_ast::Dialect;
+use common_ast::VisitorMut;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_exception::Span;
@@ -50,6 +52,14 @@ use crate::plans::SubqueryType;
 use crate::IndexType;
 use crate::WindowChecker;
 
+struct RemoveIdentifierQuote;
+
+impl VisitorMut for RemoveIdentifierQuote {
+    fn visit_identifier(&mut self, ident: &mut Identifier) {
+        ident.quote = None
+    }
+}
+
 impl Binder {
     pub fn analyze_projection(
         &mut self,
@@ -263,7 +273,12 @@ impl Binder {
                     // If alias is not specified, we will generate a name for the scalar expression.
                     let expr_name = match alias {
                         Some(alias) => normalize_identifier(alias, &self.name_resolution_ctx).name,
-                        None => format!("{:#}", expr).to_lowercase(),
+                        None => {
+                            let mut expr = expr.clone();
+                            let mut remove_quote_visitor = RemoveIdentifierQuote;
+                            walk_expr_mut(&mut remove_quote_visitor, &mut expr);
+                            format!("{:#}", expr).to_lowercase()
+                        }
                     };
 
                     prev_aliases.push((expr_name.clone(), bound_expr.clone()));

src/query/storages/fuse/src/operations/merge_into/processors/processor_merge_into_matched_and_split.rs

@@ -248,12 +248,13 @@ impl Processor for MatchedSplitProcessor {
                             .delete_mutator
                             .delete_by_expr(current_block)?;
 
+                        // delete all
+                        if !row_ids.is_empty() {
+                            row_ids = row_ids.add_meta(Some(Box::new(RowIdKind::Delete)))?;
+                            self.output_data_row_id_data.push(row_ids);
+                        }
+
                         if stage_block.is_empty() {
-                            // delete all
-                            if !row_ids.is_empty() {
-                                row_ids = row_ids.add_meta(Some(Box::new(RowIdKind::Delete)))?;
-                                self.output_data_row_id_data.push(row_ids);
-                            }
                             return Ok(());
                         }
                         current_block = stage_block;

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

@@ -507,5 +507,44 @@ select * from cluster_target;
 3 a 3
 12 b 1
 
+## add more tests
+statement ok
+drop table if exists target_test;
+
+statement ok
+drop table if exists source_test;
+
+statement ok
+create table target_test(a int,b string);
+
+statement ok
+insert into target_test values(1,'a'),(2,'b'),(3,'c');
+
+statement ok
+create table source_test(a int,b string,is_databend_deleted bool);
+
+statement ok
+insert into source_test values(1,'d',true),(2,'e',true),(3,'f',false),(4,'e',true),(5,'f',false);
+
+statement ok
+create stage source_parquet file_format = (type = parquet);
+
+statement ok
+remove @source_parquet;
+
+statement ok
+copy into @source_parquet from (select * from source_test);
+
+statement ok
+merge into `target_test` as tt using (select `a`,`b`,`is_databend_deleted` from @source_parquet (pattern => '.*[.]parquet')) as ss on (ss.`a` = tt.`a`) 
+when matched and ss.`is_databend_deleted` = true then delete when matched then update * when not matched and ss.`is_databend_deleted` = false then insert *;
+
+query TT
+select * from target_test order by a;
+----
+3 f
+5 f
+
 statement ok
 set enable_experimental_merge_into = 0;
+