[UT] [BugFix] Trigger to refresh related mvs when base tables has deleted rows (backport #48106) (#48132)

Signed-off-by: shuming.li <[email protected]>
Co-authored-by: shuming.li <[email protected]>

Author: mergify[bot]

fe/fe-core/src/main/java/com/starrocks/listener/GlobalLoadJobListenerBus.java

@@ -88,4 +88,16 @@ public void onStreamJobTransactionFinish(TransactionState transactionState) {
         }
         listeners.stream().forEach(listener -> listener.onStreamLoadTransactionFinish(transactionState));
     }
+
+    /**
+     * Do all callbacks after `delete` transaction is finished.
+     * @param db database of the target table
+     * @param table table of the target table
+     */
+    public void onDeleteJobTransactionFinish(Database db, Table table) {
+        if (db == null || table == null) {
+            return;
+        }
+        listeners.stream().forEach(listener -> listener.onDeleteJobTransactionFinish(db, table));
+    }
 }

fe/fe-core/src/main/java/com/starrocks/listener/LoadJobListener.java

@@ -50,4 +50,11 @@ public interface LoadJobListener {
      * @param table target table that has changed
      */
     void onInsertOverwriteJobCommitFinish(Database db, Table table);
+
+    /**
+     * Listener after `Delete` transaction is finished, which is only triggered without an error.
+     * @param db database of the target table
+     * @param table target table that has changed
+     */
+    void onDeleteJobTransactionFinish(Database db, Table table);
 }

fe/fe-core/src/main/java/com/starrocks/listener/LoadJobMVListener.java

@@ -156,4 +156,9 @@ private List<PartitionCommitInfo> getPartitionCommitInfos(TransactionState txnSt
         }
         return new ArrayList<>(tableCommitInfo.getIdToPartitionCommitInfo().values());
     }
+
+    @Override
+    public void onDeleteJobTransactionFinish(Database db, Table table) {
+        triggerToRefreshRelatedMVs(db, table);
+    }
 }

fe/fe-core/src/main/java/com/starrocks/listener/LoadJobStatsListener.java

@@ -90,4 +90,9 @@ private void onTransactionFinish(TransactionState transactionState, boolean sync
             LOG.warn("refresh mv after publish version failed:", DebugUtil.getStackTrace(t));
         }
     }
+
+    @Override
+    public void onDeleteJobTransactionFinish(Database db, Table table) {
+        // do nothing
+    }
 }

fe/fe-core/src/main/java/com/starrocks/load/DeleteMgr.java

@@ -95,6 +95,7 @@
 import com.starrocks.persist.metablock.SRMetaBlockWriter;
 import com.starrocks.planner.PartitionColumnFilter;
 import com.starrocks.planner.RangePartitionPruner;
+import com.starrocks.qe.QueryState;
 import com.starrocks.qe.QueryStateException;
 import com.starrocks.server.GlobalStateMgr;
 import com.starrocks.service.FrontendOptions;
@@ -175,8 +176,8 @@ public void process(DeleteStmt stmt) throws DdlException, QueryStateException {
         }
 
         DeleteJob deleteJob = null;
+        Table table = null;
         try {
-            Table table = null;
             long transactionId = -1L;
             List<Partition> partitions = Lists.newArrayList();
             db.readLock();
@@ -211,6 +212,13 @@ public void process(DeleteStmt stmt) throws DdlException, QueryStateException {
             }
 
             deleteJob.run(stmt, db, table, partitions);
+        } catch (QueryStateException e) {
+            // If delete success, it will throw QueryStateException(QueryState.MysqlStateType.OK, sb.toString()).
+            if (e.getQueryState().getStateType() == QueryState.MysqlStateType.OK) {
+                // trigger after a delete job finished
+                GlobalStateMgr.getCurrentState().getOperationListenerBus().onDeleteJobTransactionFinish(db, table);
+            }
+            throw e;
         } finally {
             if (!FeConstants.runningUnitTest) {
                 clearJob(deleteJob);

fe/fe-core/src/test/java/com/starrocks/utframe/StarRocksAssert.java

@@ -764,6 +764,46 @@ public StarRocksAssert refreshMvPartition(String sql) throws Exception {
         return this;
     }
 
+    /**
+     * Wait the input mv refresh task finished.
+     * @param mvId: mv id
+     * @return true if the mv refresh task finished, otherwise false.
+     */
+    public boolean waitRefreshFinished(long mvId) {
+        TaskManager tm = GlobalStateMgr.getCurrentState().getTaskManager();
+        Task task = tm.getTask(TaskBuilder.getMvTaskName(mvId));
+        Assert.assertTrue(task != null);
+        TaskRunManager taskRunManager = tm.getTaskRunManager();
+        TaskRunScheduler taskRunScheduler = taskRunManager.getTaskRunScheduler();
+        TaskRun taskRun = taskRunScheduler.getRunnableTaskRun(task.getId());
+        int maxTimes = 1200;
+        int count = 0;
+        while (taskRun != null && count < maxTimes) {
+            ThreadUtil.sleepAtLeastIgnoreInterrupts(500L);
+            taskRun = taskRunScheduler.getRunnableTaskRun(task.getId());
+            count += 1;
+        }
+        return taskRun == null;
+    }
+
+    /**
+     * Refresh materialized view asynchronously.
+     * @param ctx connnect context
+     * @param mvName mv's name
+     */
+    public StarRocksAssert refreshMV(ConnectContext ctx, String mvName) throws Exception {
+        String sql = "REFRESH MATERIALIZED VIEW " + mvName;
+        StatementBase stmt = UtFrameUtils.parseStmtWithNewParser(sql, ctx);
+        RefreshMaterializedViewStatement refreshMaterializedViewStatement = (RefreshMaterializedViewStatement) stmt;
+        TableName tableName = refreshMaterializedViewStatement.getMvName();
+        Database db = GlobalStateMgr.getCurrentState().getDb(tableName.getDb());
+        Table table = db.getTable(tableName.getTbl());
+        Assert.assertNotNull(table);
+        Assert.assertTrue(table instanceof MaterializedView);
+        ctx.executeSql(sql);
+        return this;
+    }
+
     public StarRocksAssert refreshMV(String sql) throws Exception {
         StatementBase stmt = UtFrameUtils.parseStmtWithNewParser(sql, ctx);
         RefreshMaterializedViewStatement refreshMaterializedViewStatement = (RefreshMaterializedViewStatement) stmt;

test/sql/test_materialized_view_refresh/R/test_mv_refresh_with_delete

@@ -0,0 +1,89 @@
+-- name: test_mv_refresh_with_delete
+create database db_${uuid0};
+-- result:
+-- !result
+use db_${uuid0};
+-- result:
+-- !result
+CREATE TABLE `t1` (
+    `k1` date,
+    `k2` int,
+    `k3` int
+)
+DUPLICATE KEY(`k1`)
+PARTITION BY RANGE (k1) (
+START ("2020-10-01") END ("2022-03-04") EVERY (INTERVAL 15 day)
+)
+DISTRIBUTED BY HASH(`k1`) BUCKETS 3;
+-- result:
+-- !result
+CREATE TABLE `t2` (
+    `k1` date,
+    `k2` int,
+    `k3` int
+)
+DUPLICATE KEY(`k1`)
+DISTRIBUTED BY HASH(`k1`) BUCKETS 3;
+-- result:
+-- !result
+INSERT INTO t1 VALUES ("2020-11-10",1,1);
+-- result:
+-- !result
+INSERT INTO t2 VALUES ("2020-11-10",2,2);
+-- result:
+-- !result
+CREATE MATERIALIZED VIEW test_mv1
+PARTITION BY k3
+DISTRIBUTED BY HASH(k1) BUCKETS 10
+REFRESH ASYNC
+PROPERTIES ("partition_refresh_number"="-1")
+AS SELECT t2.k1 as k1, t2.k2 as k2, t1.k1 as k3, t1.k2 as k4
+FROM t1 join t2 on t1.k1=t2.k1;
+-- result:
+-- !result
+function: wait_async_materialized_view_finish("db_${uuid0}", "test_mv1")
+-- result:
+None
+-- !result
+select * from test_mv1 order by k1, k2;
+-- result:
+2020-11-10	2	2020-11-10	1
+-- !result
+INSERT INTO t2 VALUES ("2020-11-10",3, 3);
+-- result:
+-- !result
+INSERT INTO t1 VALUES ("2020-11-10",4,4);
+-- result:
+-- !result
+function: wait_async_materialized_view_finish("db_${uuid0}", "test_mv1")
+-- result:
+None
+-- !result
+select * from test_mv1 order by k1, k2;
+-- result:
+2020-11-10	2	2020-11-10	1
+2020-11-10	2	2020-11-10	4
+2020-11-10	3	2020-11-10	1
+2020-11-10	3	2020-11-10	4
+-- !result
+DELETE FROM t2 WHERE k2=2;
+-- result:
+-- !result
+function: wait_async_materialized_view_finish("db_${uuid0}", "test_mv1")
+-- result:
+None
+-- !result
+select * from test_mv1 order by k1, k2;
+-- result:
+2020-11-10	3	2020-11-10	1
+2020-11-10	3	2020-11-10	4
+-- !result
+drop table t1;
+-- result:
+-- !result
+drop table t2;
+-- result:
+-- !result
+drop database db_${uuid0} force;
+-- result:
+-- !result

test/sql/test_materialized_view_refresh/T/test_mv_refresh_with_delete

@@ -0,0 +1,47 @@
+-- name: test_mv_refresh_with_delete
+
+create database db_${uuid0};
+use db_${uuid0};
+CREATE TABLE `t1` (
+    `k1` date,
+    `k2` int,
+    `k3` int
+)
+DUPLICATE KEY(`k1`)
+PARTITION BY RANGE (k1) (
+START ("2020-10-01") END ("2022-03-04") EVERY (INTERVAL 15 day)
+)
+DISTRIBUTED BY HASH(`k1`) BUCKETS 3;
+CREATE TABLE `t2` (
+    `k1` date,
+    `k2` int,
+    `k3` int
+)
+DUPLICATE KEY(`k1`)
+DISTRIBUTED BY HASH(`k1`) BUCKETS 3;
+
+INSERT INTO t1 VALUES ("2020-11-10",1,1);
+INSERT INTO t2 VALUES ("2020-11-10",2,2);
+
+CREATE MATERIALIZED VIEW test_mv1
+PARTITION BY k3
+DISTRIBUTED BY HASH(k1) BUCKETS 10
+REFRESH ASYNC
+PROPERTIES ("partition_refresh_number"="-1")
+AS SELECT t2.k1 as k1, t2.k2 as k2, t1.k1 as k3, t1.k2 as k4
+FROM t1 join t2 on t1.k1=t2.k1;
+function: wait_async_materialized_view_finish("db_${uuid0}", "test_mv1")
+select * from test_mv1 order by k1, k2;
+
+INSERT INTO t2 VALUES ("2020-11-10",3, 3);
+INSERT INTO t1 VALUES ("2020-11-10",4,4);
+function: wait_async_materialized_view_finish("db_${uuid0}", "test_mv1")
+select * from test_mv1 order by k1, k2;
+
+DELETE FROM t2 WHERE k2=2;
+function: wait_async_materialized_view_finish("db_${uuid0}", "test_mv1")
+select * from test_mv1 order by k1, k2;
+
+drop table t1;
+drop table t2;
+drop database db_${uuid0} force;