feat: support star "*" for merge into (#12906)

* add test first

* support * and add more tests

* fix clippy

* fix check

* fix check

Author: JackTan25

src/query/ast/src/ast/statements/merge_into.rs

@@ -52,7 +52,10 @@ impl Display for MergeUpdateExpr {
 
 #[derive(Debug, Clone, PartialEq)]
 pub enum MatchOperation {
-    Update { update_list: Vec<MergeUpdateExpr> },
+    Update {
+        update_list: Vec<MergeUpdateExpr>,
+        is_star: bool,
+    },
     Delete,
 }
 
@@ -66,6 +69,7 @@ pub struct MatchedClause {
 pub struct InsertOperation {
     pub columns: Option<Vec<Identifier>>,
     pub values: Vec<Expr>,
+    pub is_star: bool,
 }
 
 #[derive(Debug, Clone, PartialEq)]
@@ -116,9 +120,16 @@ impl Display for MergeIntoStmt {
                     write!(f, " THEN ")?;
 
                     match &match_clause.operation {
-                        MatchOperation::Update { update_list } => {
-                            write!(f, " UPDATE SET ")?;
-                            write_comma_separated_list(f, update_list)?;
+                        MatchOperation::Update {
+                            update_list,
+                            is_star,
+                        } => {
+                            if *is_star {
+                                write!(f, " UPDATE * ")?;
+                            } else {
+                                write!(f, " UPDATE SET ")?;
+                                write_comma_separated_list(f, update_list)?;
+                            }
                         }
                         MatchOperation::Delete => {
                             write!(f, " DELETE ")?;

src/query/ast/src/parser/statement.rs

@@ -2267,40 +2267,75 @@ fn match_operation(i: Input) -> IResult<MatchOperation> {
             rule! {
                 UPDATE ~ SET ~ ^#comma_separated_list1(merge_update_expr)
             },
-            |(_, _, update_list)| MatchOperation::Update { update_list },
+            |(_, _, update_list)| MatchOperation::Update {
+                update_list,
+                is_star: false,
+            },
+        ),
+        map(
+            rule! {
+                UPDATE ~ "*"
+            },
+            |(_, _)| MatchOperation::Update {
+                update_list: Vec::new(),
+                is_star: true,
+            },
         ),
     ))(i)
 }
 
 pub fn unmatch_clause(i: Input) -> IResult<MergeOption> {
-    map(
-        rule! {
-            WHEN ~ NOT ~ MATCHED ~ (AND ~ ^#expr)?  ~ THEN ~ INSERT ~ ( "(" ~ ^#comma_separated_list1(ident) ~ ^")" )?
-            ~ VALUES ~ ^#row_values
-        },
-        |(_, _, _, expr_op, _, _, columns_op, _, values)| {
-            let selection = match expr_op {
-                Some(e) => Some(e.1),
-                None => None,
-            };
-            match columns_op {
-                Some(columns) => MergeOption::Unmatch(UnmatchedClause {
-                    insert_operation: InsertOperation {
-                        columns: Some(columns.1),
-                        values,
-                    },
-                    selection,
-                }),
-                None => MergeOption::Unmatch(UnmatchedClause {
+    alt((
+        map(
+            rule! {
+                WHEN ~ NOT ~ MATCHED ~ (AND ~ ^#expr)?  ~ THEN ~ INSERT ~ ( "(" ~ ^#comma_separated_list1(ident) ~ ^")" )?
+                ~ VALUES ~ ^#row_values
+            },
+            |(_, _, _, expr_op, _, _, columns_op, _, values)| {
+                let selection = match expr_op {
+                    Some(e) => Some(e.1),
+                    None => None,
+                };
+                match columns_op {
+                    Some(columns) => MergeOption::Unmatch(UnmatchedClause {
+                        insert_operation: InsertOperation {
+                            columns: Some(columns.1),
+                            values,
+                            is_star: false,
+                        },
+                        selection,
+                    }),
+                    None => MergeOption::Unmatch(UnmatchedClause {
+                        insert_operation: InsertOperation {
+                            columns: None,
+                            values,
+                            is_star: false,
+                        },
+                        selection,
+                    }),
+                }
+            },
+        ),
+        map(
+            rule! {
+                WHEN ~ NOT ~ MATCHED ~ (AND ~ ^#expr)?  ~ THEN ~ INSERT ~ "*"
+            },
+            |(_, _, _, expr_op, _, _, _)| {
+                let selection = match expr_op {
+                    Some(e) => Some(e.1),
+                    None => None,
+                };
+                MergeOption::Unmatch(UnmatchedClause {
                     insert_operation: InsertOperation {
                         columns: None,
-                        values,
+                        values: Vec::new(),
+                        is_star: true,
                     },
                     selection,
-                }),
-            }
-        },
-    )(i)
+                })
+            },
+        ),
+    ))(i)
 }
 
 pub fn add_column_option(i: Input) -> IResult<AddColumnOption> {

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

@@ -179,6 +179,7 @@ impl MergeIntoInterpreter {
             } else {
                 None
             };
+
             let mut values_exprs = Vec::<RemoteExpr>::with_capacity(item.values.len());
 
             for scalar_expr in &item.values {
@@ -208,6 +209,7 @@ impl MergeIntoInterpreter {
             } else {
                 None
             };
+
             // update
             let update_list = if let Some(update_list) = &item.update {
                 // use update_plan to get exprs
@@ -224,7 +226,7 @@ impl MergeIntoInterpreter {
                 let col_indices = if item.condition.is_none() {
                     vec![]
                 } else {
-                    // we don't need to real col_indices here, just give a
+                    // we don't need real col_indices here, just give a
                     // dummy index, that's ok.
                     vec![DUMMY_COL_INDEX]
                 };
@@ -235,7 +237,6 @@ impl MergeIntoInterpreter {
                         col_indices,
                         Some(join_output_schema.num_fields()),
                     )?;
-
                 let update_list = update_list
                     .iter()
                     .map(|(idx, remote_expr)| {