fix: add visit for merge into/update/replace/insert/delete/copy statement

src/query/ast/src/visitors/visitor.rs

@@ -418,7 +418,34 @@ pub trait Visitor<'ast>: Sized {
 
     fn visit_insert(&mut self, _insert: &'ast InsertStmt) {}
     fn visit_replace(&mut self, _replace: &'ast ReplaceStmt) {}
-    fn visit_merge_into(&mut self, _merge_into: &'ast MergeIntoStmt) {}
+    fn visit_merge_into(&mut self, merge_into: &'ast MergeIntoStmt) {
+        if let MergeSource::Select { query, .. } = &merge_into.source {
+            self.visit_query(query)
+        }
+        self.visit_expr(&merge_into.join_expr);
+        for operation in &merge_into.merge_options {
+            match operation {
+                MergeOption::Match(match_operation) => {
+                    if let Some(expr) = &match_operation.selection {
+                        self.visit_expr(expr)
+                    }
+                    if let MatchOperation::Update { update_list, .. } = &match_operation.operation {
+                        for update in update_list {
+                            self.visit_expr(&update.expr)
+                        }
+                    }
+                }
+                MergeOption::Unmatch(unmatch_operation) => {
+                    if let Some(expr) = &unmatch_operation.selection {
+                        self.visit_expr(expr)
+                    }
+                    for expr in &unmatch_operation.insert_operation.values {
+                        self.visit_expr(expr)
+                    }
+                }
+            }
+        }
+    }
     fn visit_insert_source(&mut self, _insert_source: &'ast InsertSource) {}
 
     fn visit_delete(&mut self, _delete: &'ast DeleteStmt) {}

src/query/ast/src/visitors/visitor_mut.rs

@@ -431,7 +431,36 @@ pub trait VisitorMut: Sized {
 
     fn visit_insert(&mut self, _insert: &mut InsertStmt) {}
     fn visit_replace(&mut self, _replace: &mut ReplaceStmt) {}
-    fn visit_merge_into(&mut self, _merge_into: &mut MergeIntoStmt) {}
+    fn visit_merge_into(&mut self, merge_into: &mut MergeIntoStmt) {
+        if let MergeSource::Select { query, .. } = &mut merge_into.source {
+            self.visit_query(query)
+        }
+        self.visit_expr(&mut merge_into.join_expr);
+        for operation in &mut merge_into.merge_options {
+            match operation {
+                MergeOption::Match(match_operation) => {
+                    if let Some(expr) = &mut match_operation.selection {
+                        self.visit_expr(expr)
+                    }
+                    if let MatchOperation::Update { update_list, .. } =
+                        &mut match_operation.operation
+                    {
+                        for update in update_list {
+                            self.visit_expr(&mut update.expr)
+                        }
+                    }
+                }
+                MergeOption::Unmatch(unmatch_operation) => {
+                    if let Some(expr) = &mut unmatch_operation.selection {
+                        self.visit_expr(expr)
+                    }
+                    for expr in &mut unmatch_operation.insert_operation.values {
+                        self.visit_expr(expr)
+                    }
+                }
+            }
+        }
+    }
     fn visit_insert_source(&mut self, _insert_source: &mut InsertSource) {}
 
     fn visit_delete(&mut self, _delete: &mut DeleteStmt) {}

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

@@ -901,6 +901,57 @@ WHEN NOT MATCHED THEN
     INSERT (order_id, user_id, order_type, asset_type, quantity, price, status, created_at, updated_at)
     VALUES ((SELECT MAX(order_id) FROM orders) + 1, synthetic_orders.user_id, synthetic_orders.synthetic_order_type, synthetic_orders.asset_type, synthetic_orders.total_quantity, 0, 'pending', synthetic_orders.synthetic_date, synthetic_orders.synthetic_date);
 
+## issue #13810: rewrite rule test
+statement ok
+DROP TABLE IF EXISTS orders;
+
+statement ok
+CREATE TABLE orders (
+                        order_id      INT          NOT NULL,
+                        user_id       INT          NOT NULL,
+                        order_type    VARCHAR      NOT NULL,
+                        asset_type    VARCHAR      NOT NULL,
+                        quantity      DECIMAL(18,8) NOT NULL,
+                        price         DECIMAL(18,8) NOT NULL,
+                        status        VARCHAR      NOT NULL,
+                        created_at    DATE         NOT NULL,
+                        updated_at    DATE         NOT NULL
+) row_per_block=5113;
+
+statement ok
+insert into orders values(200007,7,'buy','BTC',4.81412194,48.14121943,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200015,15,'buy','BTC',3.78463552,37.84635523,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200019,19,'buy','BTC',1.61186913,16.11869132,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200031,31,'buy','BTC',3.99013730,39.90137297,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200047,47,'buy','BTC',0.98841829,9.88418289,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200077,77,'buy','BTC',2.07360391,20.73603908,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200087,87,'sell','ETH',9.64567442,96.45674419,'pending',to_date('2021-01-01'),to_date('2021-01-01')),
+(200095,95,'buy','BTC',2.26686563,22.66865634,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200098,98,'buy','BTC',1.37252960,13.72529599,'completed',to_date('2021-01-01'),to_date('2021-01-01')),
+(200102,102,'buy','BTC',1.53596481,15.35964815,'completed',to_date('2021-01-01'),to_date('2021-01-01'));
+
+statement ok
+MERGE INTO orders USING (
+    SELECT o.order_id, o.user_id, o.order_type, o.asset_type, o.quantity + a.avg_quantity AS new_quantity, o.price, o.status, o.created_at, o.updated_at
+    FROM orders o
+             INNER JOIN (
+        SELECT user_id, asset_type, sum(quantity) AS avg_quantity
+        FROM orders
+        GROUP BY user_id, asset_type
+    ) a ON o.user_id = a.user_id AND o.asset_type = a.asset_type
+) AS joined_data ON orders.order_id = joined_data.order_id
+    WHEN MATCHED THEN
+        UPDATE SET orders.quantity = joined_data.new_quantity;
+
+query TTTT
+SELECT SUM(quantity) AS total_quantity,
+       AVG(quantity) AS average_quantity,
+       MIN(quantity) AS min_quantity,
+       MAX(quantity) AS max_quantity
+FROM orders;
+----
+64.16764110 6.416764110000 1.97683658 19.29134884
+
 statement ok
 set enable_distributed_merge_into = 0;