diff --git a/include/common/tmsg.h b/include/common/tmsg.h
index 4ef4273631..86d34502c6 100644
--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@@ -3774,6 +3774,7 @@ typedef struct {
   int64_t suid;
   SArray* deleteReqs;  // SArray<SSingleDeleteReq>
   int64_t ctimeMs;     // fill by vnode
+  int8_t  level;       // 0 tsdb(default), 1 rsma1 , 2 rsma2
 } SBatchDeleteReq;
 
 int32_t tEncodeSBatchDeleteReq(SEncoder* pCoder, const SBatchDeleteReq* pReq);
diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c
index bf21b2eda0..054cff560f 100644
--- a/source/common/src/tdatablock.c
+++ b/source/common/src/tdatablock.c
@@ -2121,6 +2121,7 @@ _end:
 char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
   char* pBuf = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
   if (!pBuf) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
     return NULL;
   }
   int32_t code = buildCtbNameByGroupIdImpl(stbFullName, groupId, pBuf);
@@ -2133,6 +2134,7 @@ char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
 
 int32_t buildCtbNameByGroupIdImpl(const char* stbFullName, uint64_t groupId, char* cname) {
   if (stbFullName[0] == 0) {
+    terrno = TSDB_CODE_INVALID_PARA;
     return TSDB_CODE_FAILED;
   }
 
@@ -2142,6 +2144,7 @@ int32_t buildCtbNameByGroupIdImpl(const char* stbFullName, uint64_t groupId, cha
   }
 
   if (cname == NULL) {
+    terrno = TSDB_CODE_INVALID_PARA;
     taosArrayDestroy(tags);
     return TSDB_CODE_FAILED;
   }
diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c
index d2e4e7b845..4a1ba9e391 100644
--- a/source/common/src/tglobal.c
+++ b/source/common/src/tglobal.c
@@ -610,7 +610,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
       0)
     return -1;
 
-  tsNumOfVnodeRsmaThreads = tsNumOfCores / 2;
+  tsNumOfVnodeRsmaThreads = tsNumOfCores / 4;
   tsNumOfVnodeRsmaThreads = TMAX(tsNumOfVnodeRsmaThreads, 4);
   if (cfgAddInt32(pCfg, "numOfVnodeRsmaThreads", tsNumOfVnodeRsmaThreads, 1, 1024, CFG_SCOPE_SERVER, CFG_DYN_NONE) != 0)
     return -1;
diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c
index 34b789fef8..dc3ba7934f 100644
--- a/source/common/src/tmsg.c
+++ b/source/common/src/tmsg.c
@@ -8337,6 +8337,7 @@ int32_t tEncodeSBatchDeleteReq(SEncoder *pEncoder, const SBatchDeleteReq *pReq)
     if (tEncodeSSingleDeleteReq(pEncoder, pOneReq) < 0) return -1;
   }
   if (tEncodeI64(pEncoder, pReq->ctimeMs) < 0) return -1;
+  if (tEncodeI8(pEncoder, pReq->level) < 0) return -1;
   return 0;
 }
 
@@ -8361,6 +8362,9 @@ int32_t tDecodeSBatchDeleteReq(SDecoder *pDecoder, SBatchDeleteReq *pReq) {
   if (!tDecodeIsEnd(pDecoder)) {
     if (tDecodeI64(pDecoder, &pReq->ctimeMs) < 0) return -1;
   }
+  if (!tDecodeIsEnd(pDecoder)) {
+    if (tDecodeI8(pDecoder, &pReq->level) < 0) return -1;
+  }
   return 0;
 }
 
diff --git a/source/common/src/tname.c b/source/common/src/tname.c
index c6210ca8c9..4fe2beb6aa 100644
--- a/source/common/src/tname.c
+++ b/source/common/src/tname.c
@@ -296,7 +296,10 @@ static int compareKv(const void* p1, const void* p2) {
 void buildChildTableName(RandTableName* rName) {
   SStringBuilder sb = {0};
   taosStringBuilderAppendStringLen(&sb, rName->stbFullName, rName->stbFullNameLen);
-  if (sb.buf == NULL) return;
+  if (sb.buf == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return;
+  }
   taosArraySort(rName->tags, compareKv);
   for (int j = 0; j < taosArrayGetSize(rName->tags); ++j) {
     taosStringBuilderAppendChar(&sb, ',');
diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h
index 675bfa334a..fdd449bf36 100644
--- a/source/dnode/vnode/src/inc/tq.h
+++ b/source/dnode/vnode/src/inc/tq.h
@@ -147,7 +147,7 @@ int32_t         tqOffsetDelete(STqOffsetStore* pStore, const char* subscribeKey)
 int32_t         tqOffsetCommitFile(STqOffsetStore* pStore);
 
 // tqSink
-int32_t tqBuildDeleteReq(const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
+int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
                          const char* pIdStr);
 void    tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data);
 
@@ -160,7 +160,7 @@ int32_t tqResetStreamTaskStatus(STQ* pTq);
 int32_t tqStopStreamTasks(STQ* pTq);
 
 // tq util
-int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStreamRefDataBlock** pRefBlock);
+int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, void** pRefBlock, int32_t type);
 int32_t tqExtractDataForMq(STQ* pTq, STqHandle* pHandle, const SMqPollReq* pRequest, SRpcMsg* pMsg);
 int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp* pRsp, int32_t epoch, int64_t consumerId,
                         int32_t type, int64_t sver, int64_t ever);
diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h
index 264c64e0fc..df1720d4a7 100644
--- a/source/dnode/vnode/src/inc/vnodeInt.h
+++ b/source/dnode/vnode/src/inc/vnodeInt.h
@@ -286,7 +286,8 @@ int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
 int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);
 
 int32_t tdProcessRSmaCreate(SSma* pSma, SVCreateStbReq* pReq);
-int32_t tdProcessRSmaSubmit(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len, int32_t inputType);
+int32_t tdProcessRSmaSubmit(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len);
+int32_t tdProcessRSmaDelete(SSma* pSma, int64_t version, void* pReq, void* pMsg, int32_t len);
 int32_t tdProcessRSmaDrop(SSma* pSma, SVDropStbReq* pReq);
 int32_t tdFetchTbUidList(SSma* pSma, STbUidStore** ppStore, tb_uid_t suid, tb_uid_t uid);
 int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd);
diff --git a/source/dnode/vnode/src/sma/smaCommit.c b/source/dnode/vnode/src/sma/smaCommit.c
index 92b8c09fbc..3512f1476f 100644
--- a/source/dnode/vnode/src/sma/smaCommit.c
+++ b/source/dnode/vnode/src/sma/smaCommit.c
@@ -217,10 +217,7 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma, SCommitInfo *pInfo) {
   int32_t lino = 0;
   SVnode *pVnode = pSma->pVnode;
 
-  SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
-  if (!pSmaEnv) {
-    goto _exit;
-  }
+  if (!SMA_RSMA_ENV(pSma)) goto _exit;
 
   code = tsdbCommitBegin(VND_RSMA1(pVnode), pInfo);
   TSDB_CHECK_CODE(code, lino, _exit);
diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c
index 252a3ade36..5dc29509a0 100644
--- a/source/dnode/vnode/src/sma/smaRollup.c
+++ b/source/dnode/vnode/src/sma/smaRollup.c
@@ -17,12 +17,17 @@
 #include "tq.h"
 #include "tstream.h"
 
-#define RSMA_QTASKEXEC_SMOOTH_SIZE (100)     // cnt
-#define RSMA_SUBMIT_BATCH_SIZE     (1024)    // cnt
-#define RSMA_FETCH_DELAY_MAX       (120000)  // ms
-#define RSMA_FETCH_ACTIVE_MAX      (1000)    // ms
-#define RSMA_FETCH_INTERVAL        (5000)    // ms
-#define RSMA_TASK_FLAG             "rsma"
+#define RSMA_EXEC_SMOOTH_SIZE   (100)     // cnt
+#define RSMA_EXEC_BATCH_SIZE    (1024)    // cnt
+#define RSMA_FETCH_DELAY_MAX    (120000)  // ms
+#define RSMA_FETCH_ACTIVE_MAX   (1000)    // ms
+#define RSMA_FETCH_INTERVAL     (5000)    // ms
+#define RSMA_EXEC_TASK_FLAG     "rsma"
+#define RSMA_EXEC_MSG_HLEN      (13)  // type(int8_t) + len(int32_t) + version(int64_t)
+#define RSMA_EXEC_MSG_TYPE(msg) (*(int8_t *)(msg))
+#define RSMA_EXEC_MSG_LEN(msg)  (*(int32_t *)POINTER_SHIFT((msg), sizeof(int8_t)))
+#define RSMA_EXEC_MSG_VER(msg)  (*(int64_t *)POINTER_SHIFT((msg), sizeof(int8_t) + sizeof(int32_t)))
+#define RSMA_EXEC_MSG_BODY(msg) (POINTER_SHIFT((msg), RSMA_EXEC_MSG_HLEN))
 
 #define RSMA_NEED_FETCH(r) (RSMA_INFO_ITEM((r), 0)->fetchLevel || RSMA_INFO_ITEM((r), 1)->fetchLevel)
 
@@ -38,11 +43,11 @@ static void       tdUidStoreDestory(STbUidStore *pStore);
 static int32_t    tdUpdateTbUidListImpl(SSma *pSma, tb_uid_t *suid, SArray *tbUids, bool isAdd);
 static int32_t    tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat *pStat, SRSmaInfo *pRSmaInfo,
                                           int8_t idx);
-static int32_t    tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize, int32_t inputType, SRSmaInfo *pInfo,
-                                    ERsmaExecType type, int8_t level);
+static int32_t    tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize, int64_t version, int32_t inputType,
+                                    SRSmaInfo *pInfo, ERsmaExecType type, int8_t level);
 static SRSmaInfo *tdAcquireRSmaInfoBySuid(SSma *pSma, int64_t suid);
 static void       tdReleaseRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
-static void       tdFreeRSmaSubmitItems(SArray *pItems);
+static void       tdFreeRSmaSubmitItems(SArray *pItems, int32_t type);
 static int32_t    tdRSmaFetchAllResult(SSma *pSma, SRSmaInfo *pInfo);
 static int32_t    tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, SRSmaInfo *pInfo,
                                             int32_t execType, int8_t *streamFlushed);
@@ -288,8 +293,8 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
     pStreamTask->id.streamId = pRSmaInfo->suid + idx;
     pStreamTask->chkInfo.startTs = taosGetTimestampMs();
     pStreamTask->pMeta = pVnode->pTq->pStreamMeta;
-    pStreamTask->exec.qmsg = taosMemoryMalloc(strlen(RSMA_TASK_FLAG) + 1);
-    sprintf(pStreamTask->exec.qmsg, "%s", RSMA_TASK_FLAG);
+    pStreamTask->exec.qmsg = taosMemoryMalloc(strlen(RSMA_EXEC_TASK_FLAG) + 1);
+    sprintf(pStreamTask->exec.qmsg, "%s", RSMA_EXEC_TASK_FLAG);
     pStreamTask->chkInfo.checkpointId = streamMetaGetLatestCheckpointId(pStreamTask->pMeta);
     tdRSmaTaskInit(pStreamTask->pMeta, pItem, &pStreamTask->id);
     pStreamState = streamStateOpen(taskInfDir, pStreamTask, true, -1, -1);
@@ -624,6 +629,45 @@ _end:
   return code;
 }
 
+static int32_t tdRSmaProcessDelReq(SSma *pSma, int64_t suid, int8_t level, SBatchDeleteReq *pDelReq) {
+  int32_t code = 0;
+  int32_t lino = 0;
+
+  if (taosArrayGetSize(pDelReq->deleteReqs) > 0) {
+    int32_t len = 0;
+    tEncodeSize(tEncodeSBatchDeleteReq, pDelReq, len, code);
+    TSDB_CHECK_CODE(code, lino, _exit);
+
+    void *pBuf = rpcMallocCont(len + sizeof(SMsgHead));
+    if (!pBuf) {
+      code = terrno;
+      TSDB_CHECK_CODE(code, lino, _exit);
+    }
+
+    SEncoder encoder;
+    tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SMsgHead)), len);
+    tEncodeSBatchDeleteReq(&encoder, pDelReq);
+    tEncoderClear(&encoder);
+
+    ((SMsgHead *)pBuf)->vgId = TD_VID(pSma->pVnode);
+
+    SRpcMsg delMsg = {.msgType = TDMT_VND_BATCH_DEL,
+                      .pCont = pBuf,
+                      .contLen = len + sizeof(SMsgHead)};
+    code = tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &delMsg);
+    TSDB_CHECK_CODE(code, lino, _exit);
+  }
+
+_exit:
+  taosArrayDestroy(pDelReq->deleteReqs);
+  if (code) {
+    smaError("vgId:%d, failed at line %d to process delete req for table:%" PRIi64 ", level:%" PRIi8 " since %s",
+             SMA_VID(pSma), lino, suid, level, tstrerror(code));
+  }
+
+  return code;
+}
+
 static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSmaInfoItem *pItem, SRSmaInfo *pInfo,
                                          int32_t execType, int8_t *streamFlushed) {
   int32_t      code = 0;
@@ -652,9 +696,42 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
       if (output->info.type == STREAM_CHECKPOINT) {
         if (streamFlushed) *streamFlushed = 1;
         continue;
+      } else if (output->info.type == STREAM_DELETE_RESULT) {
+        SBatchDeleteReq deleteReq = {.suid = suid, .level = pItem->level};
+        deleteReq.deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
+        if (!deleteReq.deleteReqs) {
+          code = terrno;
+          TSDB_CHECK_CODE(code, lino, _exit);
+        }
+        code = tqBuildDeleteReq(pSma->pVnode->pTq, NULL, output, &deleteReq, "");
+        TSDB_CHECK_CODE(code, lino, _exit);
+        code = tdRSmaProcessDelReq(pSma, suid, pItem->level, &deleteReq);
+        TSDB_CHECK_CODE(code, lino, _exit);
+        continue;
+      }
+
+      smaDebug("vgId:%d, result block, execType:%d, ver:%" PRIi64 ", submitReqVer:%" PRIi64 ", fetchResultVer:%" PRIi64
+               ", suid:%" PRIi64 ", level:%" PRIi8 ", uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%" PRIi64,
+               SMA_VID(pSma), execType, output->info.version, pItem->submitReqVer, pItem->fetchResultVer, suid,
+               pItem->level, output->info.id.uid, output->info.id.groupId, output->info.rows);
+
+      if (STREAM_GET_ALL == execType) {
+        /**
+         * 1. reset the output version when reboot
+         * 2. delete msg version not updated from the result
+         */
+        if (output->info.version < pItem->submitReqVer) {
+          // submitReqVer keeps unchanged since tdExecuteRSmaImpl and tdRSmaFetchAllResult are executed synchronously
+          output->info.version = pItem->submitReqVer;
+        } else if (output->info.version == pItem->fetchResultVer) {
+          smaWarn("vgId:%d, result block, skip dup version, execType:%d, ver:%" PRIi64 ", submitReqVer:%" PRIi64
+                  ", fetchResultVer:%" PRIi64 ", suid:%" PRIi64 ", level:%" PRIi8 ", uid:%" PRIu64 ", groupid:%" PRIu64
+                  ", rows:%" PRIi64,
+                  SMA_VID(pSma), execType, output->info.version, pItem->submitReqVer, pItem->fetchResultVer, suid,
+                  pItem->level, output->info.id.uid, output->info.id.groupId, output->info.rows);
+          continue;
+        }
       }
-      smaDebug("vgId:%d, result block, uid:%" PRIu64 ", groupid:%" PRIu64 ", rows:%" PRIi64, SMA_VID(pSma),
-               output->info.id.uid, output->info.id.groupId, output->info.rows);
 
       STsdb       *sinkTsdb = (pItem->level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb[0] : pSma->pRSmaTsdb[1]);
       SSubmitReq2 *pReq = NULL;
@@ -664,12 +741,6 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
         TSDB_CHECK_CODE(code, lino, _exit);
       }
 
-      // reset the output version to handle reboot
-      if (STREAM_GET_ALL == execType && output->info.version == 0) {
-        // the submitReqVer keeps unchanged since tdExecuteRSmaImpl and tdRSmaFetchAllResult are executed synchronously
-        output->info.version = pItem->submitReqVer;
-      }
-
       if (pReq && tdProcessSubmitReq(sinkTsdb, output->info.version, pReq) < 0) {
         if (terrno == TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE) {
           // TODO: reconfigure SSubmitReq2
@@ -686,8 +757,9 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
         atomic_store_64(&pItem->fetchResultVer, output->info.version);
       }
 
-      smaDebug("vgId:%d, process submit req for rsma suid:%" PRIu64 ",uid:%" PRIu64 ", level %" PRIi8 " ver %" PRIi64,
-               SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, output->info.version);
+      smaDebug("vgId:%d, process submit req for rsma suid:%" PRIu64 ",uid:%" PRIu64 ", level:%" PRIi8
+               ", execType:%d, ver:%" PRIi64,
+               SMA_VID(pSma), suid, output->info.id.groupId, pItem->level, execType, output->info.version);
 
       if (pReq) {
         tDestroySubmitReq(pReq, TSDB_MSG_FLG_ENCODE);
@@ -722,7 +794,7 @@ _exit:
  */
 static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *pMsg, int32_t len, int32_t inputType,
                                       SRSmaInfo *pInfo, tb_uid_t suid) {
-  int32_t size = sizeof(int32_t) + sizeof(int64_t) + len;
+  int32_t size = RSMA_EXEC_MSG_HLEN + len;  // header + payload
   void   *qItem = taosAllocateQitem(size, DEF_QITEM, 0);
 
   if (!qItem) {
@@ -731,6 +803,8 @@ static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *p
 
   void *pItem = qItem;
 
+  *(int8_t *)pItem = (int8_t)inputType;
+  pItem = POINTER_SHIFT(pItem, sizeof(int8_t));
   *(int32_t *)pItem = len;
   pItem = POINTER_SHIFT(pItem, sizeof(int32_t));
   *(int64_t *)pItem = version;
@@ -749,7 +823,7 @@ static int32_t tdExecuteRSmaImplAsync(SSma *pSma, int64_t version, const void *p
   }
 
   // smoothing consume
-  int32_t n = nItems / RSMA_QTASKEXEC_SMOOTH_SIZE;
+  int32_t n = nItems / RSMA_EXEC_SMOOTH_SIZE;
   if (n > 1) {
     if (n > 10) {
       n = 10;
@@ -796,7 +870,7 @@ static int32_t tdRsmaPrintSubmitReq(SSma *pSma, SSubmitReq *pReq) {
  * @param level
  * @return int32_t
  */
-static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize, int32_t inputType, SRSmaInfo *pInfo,
+static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize, int64_t version, int32_t inputType, SRSmaInfo *pInfo,
                                  ERsmaExecType type, int8_t level) {
   int32_t        idx = level - 1;
   void          *qTaskInfo = RSMA_INFO_QTASK(pInfo, idx);
@@ -813,25 +887,15 @@ static int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int32_t msgSize,
     return TSDB_CODE_FAILED;
   }
 
-  smaDebug("vgId:%d, execute rsma %" PRIi8 " task for qTaskInfo:%p suid:%" PRIu64 " nMsg:%d", SMA_VID(pSma), level,
-           RSMA_INFO_QTASK(pInfo, idx), pInfo->suid, msgSize);
+  smaDebug("vgId:%d, execute rsma %" PRIi8 " task for qTaskInfo:%p, suid:%" PRIu64 ", nMsg:%d, submitReqVer:%" PRIi64 ", inputType:%d", SMA_VID(pSma), level,
+           RSMA_INFO_QTASK(pInfo, idx), pInfo->suid, msgSize, version, inputType);
 
-#if 0
-  for (int32_t i = 0; i < msgSize; ++i) {
-    SSubmitReq *pReq = *(SSubmitReq **)((char *)pMsg + i * sizeof(void *));
-    smaDebug("vgId:%d, [%d][%d] version %" PRIi64, SMA_VID(pSma), msgSize, i, pReq->version);
-    tdRsmaPrintSubmitReq(pSma, pReq);
-  }
-#endif
   if ((terrno = qSetSMAInput(qTaskInfo, pMsg, msgSize, inputType)) < 0) {
     smaError("vgId:%d, rsma %" PRIi8 " qSetStreamInput failed since %s", SMA_VID(pSma), level, tstrerror(terrno));
     return TSDB_CODE_FAILED;
   }
 
-  if (STREAM_INPUT__MERGED_SUBMIT == inputType) {
-    SPackedData *packData = POINTER_SHIFT(pMsg, sizeof(SPackedData) * (msgSize - 1));
-    atomic_store_64(&pItem->submitReqVer, packData->ver);
-  }
+  atomic_store_64(&pItem->submitReqVer, version);
 
   terrno = tdRSmaExecAndSubmitResult(pSma, qTaskInfo, pItem, pInfo, STREAM_NORMAL, NULL);
 
@@ -910,7 +974,7 @@ static int32_t tdExecuteRSmaAsync(SSma *pSma, int64_t version, const void *pMsg,
     return TSDB_CODE_SUCCESS;
   }
 
-  if (inputType == STREAM_INPUT__DATA_SUBMIT) {
+  if (inputType == STREAM_INPUT__DATA_SUBMIT || inputType == STREAM_INPUT__REF_DATA_BLOCK) {
     if (tdExecuteRSmaImplAsync(pSma, version, pMsg, len, inputType, pRSmaInfo, suid) < 0) {
       tdReleaseRSmaInfo(pSma, pRSmaInfo);
       return TSDB_CODE_FAILED;
@@ -937,12 +1001,8 @@ static int32_t tdExecuteRSmaAsync(SSma *pSma, int64_t version, const void *pMsg,
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len, int32_t inputType) {
-  SSmaEnv *pEnv = SMA_RSMA_ENV(pSma);
-  if (!pEnv) {
-    // only applicable when rsma env exists
-    return TDB_CODE_SUCCESS;
-  }
+int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len) {
+  if (!SMA_RSMA_ENV(pSma)) return TSDB_CODE_SUCCESS;
 
   if ((terrno = atomic_load_32(&SMA_RSMA_STAT(pSma)->execStat))) {
     smaError("vgId:%d, failed to process rsma submit since invalid exec code: %s", SMA_VID(pSma), terrstr());
@@ -951,27 +1011,25 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, int64_t version, void *pReq, void *pMsg,
 
   STbUidStore uidStore = {0};
 
-  if (inputType == STREAM_INPUT__DATA_SUBMIT) {
-    if (tdFetchSubmitReqSuids(pReq, &uidStore) < 0) {
-      smaError("vgId:%d, failed to process rsma submit fetch suid since: %s", SMA_VID(pSma), terrstr());
+  if (tdFetchSubmitReqSuids(pReq, &uidStore) < 0) {
+    smaError("vgId:%d, failed to process rsma submit fetch suid since: %s", SMA_VID(pSma), terrstr());
+    goto _err;
+  }
+
+  if (uidStore.suid != 0) {
+    if (tdExecuteRSmaAsync(pSma, version, pMsg, len, STREAM_INPUT__DATA_SUBMIT, uidStore.suid) < 0) {
+      smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr());
       goto _err;
     }
 
-    if (uidStore.suid != 0) {
-      if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, uidStore.suid) < 0) {
-        smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr());
+    void *pIter = NULL;
+    while ((pIter = taosHashIterate(uidStore.uidHash, pIter))) {
+      tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
+      if (tdExecuteRSmaAsync(pSma, version, pMsg, len, STREAM_INPUT__DATA_SUBMIT, *pTbSuid) < 0) {
+        smaError("vgId:%d, failed to process rsma submit exec 2 since: %s", SMA_VID(pSma), terrstr());
+        taosHashCancelIterate(uidStore.uidHash, pIter);
         goto _err;
       }
-
-      void *pIter = NULL;
-      while ((pIter = taosHashIterate(uidStore.uidHash, pIter))) {
-        tb_uid_t *pTbSuid = (tb_uid_t *)taosHashGetKey(pIter, NULL);
-        if (tdExecuteRSmaAsync(pSma, version, pMsg, len, inputType, *pTbSuid) < 0) {
-          smaError("vgId:%d, failed to process rsma submit exec 2 since: %s", SMA_VID(pSma), terrstr());
-          taosHashCancelIterate(uidStore.uidHash, pIter);
-          goto _err;
-        }
-      }
     }
   }
   tdUidStoreDestory(&uidStore);
@@ -981,6 +1039,24 @@ _err:
   return terrno;
 }
 
+int32_t tdProcessRSmaDelete(SSma *pSma, int64_t version, void *pReq, void *pMsg, int32_t len) {
+  if (!SMA_RSMA_ENV(pSma)) return TSDB_CODE_SUCCESS;
+
+  if ((terrno = atomic_load_32(&SMA_RSMA_STAT(pSma)->execStat))) {
+    smaError("vgId:%d, failed to process rsma delete since invalid exec code: %s", SMA_VID(pSma), terrstr());
+    goto _err;
+  }
+
+  SDeleteRes *pDelRes = pReq;
+  if (tdExecuteRSmaAsync(pSma, version, pMsg, len, STREAM_INPUT__REF_DATA_BLOCK, pDelRes->suid) < 0) {
+    smaError("vgId:%d, failed to process rsma submit exec 1 since: %s", SMA_VID(pSma), terrstr());
+    goto _err;
+  }
+  return TSDB_CODE_SUCCESS;
+_err:
+  return terrno;
+}
+
 /**
  * @brief retrieve rsma meta and init
  *
@@ -1359,10 +1435,20 @@ _end:
   tdReleaseSmaRef(smaMgmt.rsetId, pRSmaRef->refId);
 }
 
-static void tdFreeRSmaSubmitItems(SArray *pItems) {
-  for (int32_t i = 0; i < taosArrayGetSize(pItems); ++i) {
-    SPackedData *packData = taosArrayGet(pItems, i);
-    taosFreeQitem(POINTER_SHIFT(packData->msgStr, -sizeof(int32_t) - sizeof(int64_t)));
+static void tdFreeRSmaSubmitItems(SArray *pItems, int32_t type) {
+  int32_t arrSize = taosArrayGetSize(pItems);
+  if (type == STREAM_INPUT__MERGED_SUBMIT) {
+    for (int32_t i = 0; i < arrSize; ++i) {
+      SPackedData *packData = TARRAY_GET_ELEM(pItems, i);
+      taosFreeQitem(POINTER_SHIFT(packData->msgStr, -RSMA_EXEC_MSG_HLEN));
+    }
+  } else if (type == STREAM_INPUT__REF_DATA_BLOCK) {
+    for (int32_t i = 0; i < arrSize; ++i) {
+      SPackedData *packData = TARRAY_GET_ELEM(pItems, i);
+      blockDataDestroy(packData->pDataBlock);
+    }
+  } else {
+    ASSERTS(0, "unknown type:%d", type);
   }
   taosArrayClear(pItems);
 }
@@ -1427,40 +1513,98 @@ _err:
 }
 
 static int32_t tdRSmaBatchExec(SSma *pSma, SRSmaInfo *pInfo, STaosQall *qall, SArray *pSubmitArr, ERsmaExecType type) {
+  void   *msg = NULL;
+  int8_t  resume = 0;
+  int32_t nSubmit = 0;
+  int32_t nDelete = 0;
+  int64_t version = 0;
+
+  SPackedData packData;
+
   taosArrayClear(pSubmitArr);
+
+  // the submitReq/deleteReq msg may exsit alternately in the msg queue, consume them sequentially in batch mode
   while (1) {
-    void *msg = NULL;
     taosGetQitem(qall, (void **)&msg);
     if (msg) {
-      SPackedData packData = {.msgLen = *(int32_t *)msg,
-                              .ver = *(int64_t *)POINTER_SHIFT(msg, sizeof(int32_t)),
-                              .msgStr = POINTER_SHIFT(msg, sizeof(int32_t) + sizeof(int64_t))};
+      int8_t inputType = RSMA_EXEC_MSG_TYPE(msg);
+      if (inputType == STREAM_INPUT__DATA_SUBMIT) {
+        if (nDelete > 0) {
+          resume = 1;
+          break;
+        }
+      _resume_submit:
+        packData.msgLen = RSMA_EXEC_MSG_LEN(msg);
+        packData.ver = RSMA_EXEC_MSG_VER(msg);
+        packData.msgStr = RSMA_EXEC_MSG_BODY(msg);
+        version = packData.ver;
+        if (!taosArrayPush(pSubmitArr, &packData)) {
+          taosFreeQitem(msg);
+          terrno = TSDB_CODE_OUT_OF_MEMORY;
+          goto _err;
+        }
+        ++nSubmit;
+      } else if (inputType == STREAM_INPUT__REF_DATA_BLOCK) {
+        if (nSubmit > 0) {
+          resume = 2;
+          break;
+        }
+      _resume_delete:
+        version = RSMA_EXEC_MSG_VER(msg);
+        if ((terrno = extractDelDataBlock(RSMA_EXEC_MSG_BODY(msg), RSMA_EXEC_MSG_LEN(msg), version,
+                                          &packData.pDataBlock, 1))) {
+          taosFreeQitem(msg);
+          goto _err;
+        }
 
-      if (!taosArrayPush(pSubmitArr, &packData)) {
-        terrno = TSDB_CODE_OUT_OF_MEMORY;
-        tdFreeRSmaSubmitItems(pSubmitArr);
-        goto _err;
+        if (packData.pDataBlock && !taosArrayPush(pSubmitArr, &packData)) {
+          taosFreeQitem(msg);
+          terrno = TSDB_CODE_OUT_OF_MEMORY;
+          goto _err;
+        }
+        taosFreeQitem(msg);
+        if (packData.pDataBlock) {
+          // packData.pDataBlock is NULL if delete affects 0 row
+          ++nDelete;
+        }
+      } else {
+        ASSERTS(0, "unknown msg type:%d", inputType);
+        break;
       }
+    }
+
+    if (nSubmit > 0 || nDelete > 0) {
+      int32_t size = TARRAY_SIZE(pSubmitArr);
+      ASSERTS(size > 0, "size is %d", size);
+      int32_t inputType = nSubmit > 0 ? STREAM_INPUT__MERGED_SUBMIT : STREAM_INPUT__REF_DATA_BLOCK;
+      for (int32_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
+        if (tdExecuteRSmaImpl(pSma, pSubmitArr->pData, size, version, inputType, pInfo, type, i) < 0) {
+          goto _err;
+        }
+      }
+      tdFreeRSmaSubmitItems(pSubmitArr, inputType);
+      nSubmit = 0;
+      nDelete = 0;
     } else {
-      break;
+      goto _rtn;
+    }
+
+    if (resume == 1) {
+      resume = 0;
+      goto _resume_submit;
+    } else if (resume == 2) {
+      resume = 0;
+      goto _resume_delete;
     }
   }
 
-  int32_t size = taosArrayGetSize(pSubmitArr);
-  if (size > 0) {
-    for (int32_t i = 1; i <= TSDB_RETENTION_L2; ++i) {
-      if (tdExecuteRSmaImpl(pSma, pSubmitArr->pData, size, STREAM_INPUT__MERGED_SUBMIT, pInfo, type, i) < 0) {
-        goto _err;
-      }
-    }
-    tdFreeRSmaSubmitItems(pSubmitArr);
-  }
+_rtn:
   return TSDB_CODE_SUCCESS;
 _err:
   atomic_store_32(&SMA_RSMA_STAT(pSma)->execStat, terrno);
   smaError("vgId:%d, batch exec for suid:%" PRIi64 " execType:%d size:%d failed since %s", SMA_VID(pSma), pInfo->suid,
            type, (int32_t)taosArrayGetSize(pSubmitArr), terrstr());
-  tdFreeRSmaSubmitItems(pSubmitArr);
+  tdFreeRSmaSubmitItems(pSubmitArr, nSubmit ? STREAM_INPUT__MERGED_SUBMIT : STREAM_INPUT__REF_DATA_BLOCK);
   while (1) {
     void *msg = NULL;
     taosGetQitem(qall, (void **)&msg);
@@ -1497,7 +1641,7 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
   }
 
   if (!(pSubmitArr =
-            taosArrayInit(TMIN(RSMA_SUBMIT_BATCH_SIZE, atomic_load_64(&pRSmaStat->nBufItems)), sizeof(SPackedData)))) {
+            taosArrayInit(TMIN(RSMA_EXEC_BATCH_SIZE, atomic_load_64(&pRSmaStat->nBufItems)), sizeof(SPackedData)))) {
     code = TSDB_CODE_OUT_OF_MEMORY;
     TSDB_CHECK_CODE(code, lino, _exit);
   }
diff --git a/source/dnode/vnode/src/sma/smaTimeRange.c b/source/dnode/vnode/src/sma/smaTimeRange.c
index 08ddc4bd7b..289986e01f 100644
--- a/source/dnode/vnode/src/sma/smaTimeRange.c
+++ b/source/dnode/vnode/src/sma/smaTimeRange.c
@@ -188,7 +188,8 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
     if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
       pDeleteReq->suid = suid;
       pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
-      tqBuildDeleteReq(stbFullName, pDataBlock, pDeleteReq, "");
+      code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, pDeleteReq, "");
+      TSDB_CHECK_CODE(code, lino, _exit);
       continue;
     }
 
diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c
index bd2a591a98..41b1aa7bd1 100644
--- a/source/dnode/vnode/src/tq/tqRead.c
+++ b/source/dnode/vnode/src/tq/tqRead.c
@@ -343,7 +343,7 @@ int32_t extractMsgFromWal(SWalReader* pReader, void** pItem, int64_t maxVer, con
       void*   pBody = POINTER_SHIFT(pCont->body, sizeof(SMsgHead));
       int32_t len = pCont->bodyLen - sizeof(SMsgHead);
 
-      code = extractDelDataBlock(pBody, len, ver, (SStreamRefDataBlock**)pItem);
+      code = extractDelDataBlock(pBody, len, ver, (void**)pItem, 0);
       if (code == TSDB_CODE_SUCCESS) {
         if (*pItem == NULL) {
           tqDebug("s-task:%s empty delete msg, discard it, len:%d, ver:%" PRId64, id, len, ver);
diff --git a/source/dnode/vnode/src/tq/tqSink.c b/source/dnode/vnode/src/tq/tqSink.c
index 4b64737936..c2e48d5d92 100644
--- a/source/dnode/vnode/src/tq/tqSink.c
+++ b/source/dnode/vnode/src/tq/tqSink.c
@@ -43,7 +43,7 @@ static SArray* createDefaultTagColName();
 static void    setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName,
                                    int64_t gid);
 
-int32_t tqBuildDeleteReq(const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
+int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq,
                          const char* pIdStr) {
   int32_t          totalRows = pDataBlock->info.rows;
   SColumnInfoData* pStartTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX);
@@ -58,8 +58,9 @@ int32_t tqBuildDeleteReq(const char* stbFullName, const SSDataBlock* pDataBlock,
     int64_t ekey = *(int64_t*)colDataGetData(pEndTsCol, row);
     int64_t groupId = *(int64_t*)colDataGetData(pGidCol, row);
 
-    char*   name;
-    void*   varTbName = NULL;
+    char* name = NULL;
+    char* originName = NULL;
+    void* varTbName = NULL;
     if (!colDataIsNull(pTbNameCol, totalRows, row, NULL)) {
       varTbName = colDataGetVarData(pTbNameCol, row);
     }
@@ -67,18 +68,29 @@ int32_t tqBuildDeleteReq(const char* stbFullName, const SSDataBlock* pDataBlock,
     if (varTbName != NULL && varTbName != (void*)-1) {
       name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
       memcpy(name, varDataVal(varTbName), varDataLen(varTbName));
-    } else {
+    } else if (stbFullName) {
       name = buildCtbNameByGroupId(stbFullName, groupId);
+    } else {
+      originName = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE);
+      if (metaGetTableNameByUid(pTq->pVnode, groupId, originName) == 0) {
+        name = varDataVal(originName);
+      }
     }
 
-    tqDebug("s-task:%s build delete msg groupId:%" PRId64 ", name:%s, skey:%" PRId64 " ekey:%" PRId64,
-            pIdStr, groupId, name, skey, ekey);
+    if (!name || *name == '\0') {
+      tqWarn("s-task:%s failed to build delete msg groupId:%" PRId64 ", skey:%" PRId64 " ekey:%" PRId64
+             " since invalid tbname:%s",
+             pIdStr, groupId, skey, ekey, name ? name : "NULL");
+    } else {
+      tqDebug("s-task:%s build delete msg groupId:%" PRId64 ", name:%s, skey:%" PRId64 " ekey:%" PRId64, pIdStr,
+              groupId, name, skey, ekey);
 
-    SSingleDeleteReq req = { .startTs = skey, .endTs = ekey};
-    strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN - 1);
-    taosMemoryFree(name);
-
-    taosArrayPush(deleteReq->deleteReqs, &req);
+      SSingleDeleteReq req = {.startTs = skey, .endTs = ekey};
+      strncpy(req.tbname, name, TSDB_TABLE_NAME_LEN - 1);
+      taosArrayPush(deleteReq->deleteReqs, &req);
+    }
+    if (originName) name = originName;
+    taosMemoryFreeClear(name);
   }
 
   return 0;
@@ -345,7 +357,7 @@ int32_t doBuildAndSendDeleteMsg(SVnode* pVnode, char* stbFullName, SSDataBlock*
                                 int64_t suid) {
   SBatchDeleteReq deleteReq = {.suid = suid, .deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq))};
 
-  int32_t code = tqBuildDeleteReq(stbFullName, pDataBlock, &deleteReq, pTask->id.idStr);
+  int32_t code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, &deleteReq, pTask->id.idStr);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
   }
diff --git a/source/dnode/vnode/src/tq/tqUtil.c b/source/dnode/vnode/src/tq/tqUtil.c
index d8fe899bf6..8f62928d22 100644
--- a/source/dnode/vnode/src/tq/tqUtil.c
+++ b/source/dnode/vnode/src/tq/tqUtil.c
@@ -399,7 +399,7 @@ int32_t tqDoSendDataRsp(const SRpcHandleInfo* pRpcHandleInfo, const SMqDataRsp*
   return 0;
 }
 
-int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStreamRefDataBlock** pRefBlock) {
+int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, void** pRefBlock, int32_t type) {
   SDecoder*   pCoder = &(SDecoder){0};
   SDeleteRes* pRes = &(SDeleteRes){0};
 
@@ -442,14 +442,21 @@ int32_t extractDelDataBlock(const void* pData, int32_t len, int64_t ver, SStream
   }
 
   taosArrayDestroy(pRes->uidList);
-  *pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0);
-  if (*pRefBlock == NULL) {
-    blockDataCleanup(pDelBlock);
-    taosMemoryFree(pDelBlock);
-    return TSDB_CODE_OUT_OF_MEMORY;
+  if (type == 0) {
+    *pRefBlock = taosAllocateQitem(sizeof(SStreamRefDataBlock), DEF_QITEM, 0);
+    if (*pRefBlock == NULL) {
+      blockDataCleanup(pDelBlock);
+      taosMemoryFree(pDelBlock);
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+
+    ((SStreamRefDataBlock*)(*pRefBlock))->type = STREAM_INPUT__REF_DATA_BLOCK;
+    ((SStreamRefDataBlock*)(*pRefBlock))->pBlock = pDelBlock;
+  } else if (type == 1) {
+    *pRefBlock = pDelBlock;
+  } else {
+    ASSERTS(0, "unknown type:%d", type);
   }
 
-  (*pRefBlock)->type = STREAM_INPUT__REF_DATA_BLOCK;
-  (*pRefBlock)->pBlock = pDelBlock;
   return TSDB_CODE_SUCCESS;
 }
diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c
index eadfd39d0b..29db7d602f 100644
--- a/source/dnode/vnode/src/vnd/vnodeSvr.c
+++ b/source/dnode/vnode/src/vnd/vnodeSvr.c
@@ -380,7 +380,7 @@ static int32_t vnodePreProcessDeleteMsg(SVnode *pVnode, SRpcMsg *pMsg) {
   SEncoder  *pCoder = &(SEncoder){0};
   SDeleteRes res = {0};
 
-  SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb};
+  SReadHandle handle = {.vnode = pVnode, .pMsgCb = &pVnode->msgCb, .skipRollup = 1};
   initStorageAPI(&handle.api);
 
   code = qWorkerProcessDeleteMsg(&handle, pVnode->pQuery, pMsg, &res);
@@ -1674,7 +1674,7 @@ _exit:
   atomic_add_fetch_64(&pVnode->statis.nBatchInsert, 1);
   if (code == 0) {
     atomic_add_fetch_64(&pVnode->statis.nBatchInsertSuccess, 1);
-    code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len, STREAM_INPUT__DATA_SUBMIT);
+    code = tdProcessRSmaSubmit(pVnode->pSma, ver, pSubmitReq, pReq, len);
   }
 
   // clear
@@ -1891,6 +1891,11 @@ static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t ver, void *pRe
 
   SMetaReader mr = {0};
   metaReaderDoInit(&mr, pVnode->pMeta, META_READER_NOLOCK);
+  STsdb *pTsdb = pVnode->pTsdb;
+
+  if (deleteReq.level) {
+    pTsdb = deleteReq.level == 1 ? VND_RSMA1(pVnode) : VND_RSMA2(pVnode);
+  }
 
   int32_t sz = taosArrayGetSize(deleteReq.deleteReqs);
   for (int32_t i = 0; i < sz; i++) {
@@ -1903,21 +1908,22 @@ static int32_t vnodeProcessBatchDeleteReq(SVnode *pVnode, int64_t ver, void *pRe
 
     int64_t uid = mr.me.uid;
 
-    int32_t code = tsdbDeleteTableData(pVnode->pTsdb, ver, deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
+    int32_t code = tsdbDeleteTableData(pTsdb, ver, deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
     if (code < 0) {
       terrno = code;
       vError("vgId:%d, delete error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64 ", end ts:%" PRId64,
              TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
     }
 
-    code = metaUpdateChangeTimeWithLock(pVnode->pMeta, uid, deleteReq.ctimeMs);
-    if (code < 0) {
-      terrno = code;
-      vError("vgId:%d, update change time error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64
-             ", end ts:%" PRId64,
-             TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
+    if (deleteReq.level == 0) {
+      code = metaUpdateChangeTimeWithLock(pVnode->pMeta, uid, deleteReq.ctimeMs);
+      if (code < 0) {
+        terrno = code;
+        vError("vgId:%d, update change time error since %s, suid:%" PRId64 ", uid:%" PRId64 ", start ts:%" PRId64
+               ", end ts:%" PRId64,
+               TD_VID(pVnode), terrstr(), deleteReq.suid, uid, pOneReq->startTs, pOneReq->endTs);
+      }
     }
-
     tDecoderClear(&mr.coder);
   }
   metaReaderClear(&mr);
@@ -1952,6 +1958,8 @@ static int32_t vnodeProcessDeleteReq(SVnode *pVnode, int64_t ver, void *pReq, in
     if (code) goto _err;
   }
 
+  code = tdProcessRSmaDelete(pVnode->pSma, ver, pRes, pReq, len);
+
   tDecoderClear(pCoder);
   taosArrayDestroy(pRes->uidList);
 
diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c
index a0e35fff87..1f82a9477b 100644
--- a/source/libs/executor/src/executor.c
+++ b/source/libs/executor/src/executor.c
@@ -77,6 +77,12 @@ static int32_t doSetSMABlock(SOperatorInfo* pOperator, void* input, size_t numOf
       SPackedData tmp = {.pDataBlock = input};
       taosArrayPush(pInfo->pBlockLists, &tmp);
       pInfo->blockType = STREAM_INPUT__CHECKPOINT;
+    } else if (type == STREAM_INPUT__REF_DATA_BLOCK) {
+      for (int32_t i = 0; i < numOfBlocks; ++i) {       
+        SPackedData* pReq = POINTER_SHIFT(input, i * sizeof(SPackedData));
+        taosArrayPush(pInfo->pBlockLists, pReq);
+      }
+      pInfo->blockType = STREAM_INPUT__DATA_BLOCK;
     }
 
     return TSDB_CODE_SUCCESS;
diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task
index 36204f553a..6a7c0b47ec 100644
--- a/tests/parallel_test/cases.task
+++ b/tests/parallel_test/cases.task
@@ -1183,6 +1183,7 @@ e
 ,,y,script,./test.sh -f tsim/sma/sma_leak.sim
 ,,y,script,./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
 ,,y,script,./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
+,,y,script,./test.sh -f tsim/sma/rsmaCreateInsertQueryDelete.sim
 ,,y,script,./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim
 ,,y,script,./test.sh -f tsim/sync/vnodesnapshot-rsma-test.sim
 ,,n,script,./test.sh -f tsim/valgrind/checkError1.sim
diff --git a/tests/script/tsim/sma/rsmaCreateInsertQueryDelete.sim b/tests/script/tsim/sma/rsmaCreateInsertQueryDelete.sim
new file mode 100644
index 0000000000..b93f6f0c44
--- /dev/null
+++ b/tests/script/tsim/sma/rsmaCreateInsertQueryDelete.sim
@@ -0,0 +1,540 @@
+system sh/stop_dnodes.sh
+system sh/deploy.sh -n dnode1 -i 1
+system sh/exec.sh -n dnode1 -s start
+sleep 50
+sql connect
+
+print =============== create database with retentions
+sql create database d0 retentions -:7d,10s:21d,15s:365d vgroups 1;
+sql use d0
+
+print =============== create super table and register rsma
+sql create table if not exists stb (ts timestamp, c1 float, c2 double) tags (city binary(20),district binary(20)) rollup(sum) max_delay 1s,1s;
+
+sql show stables
+if $rows != 1 then
+  return -1
+endi
+
+print =============== create child table
+sql create table ct1 using stb tags("BeiJing", "ChaoYang");
+
+sql show tables
+if $rows != 1 then
+  return -1
+endi
+
+print =============== insert data and trigger rollup
+sql insert into ct1 values(now, 10, NULL);
+sql insert into ct1 values(now+60m, 1, NULL);
+sql insert into ct1 values(now+120m, 100, NULL);
+
+print =============== wait 7 seconds for results
+sleep 7000
+
+print =============== select * from retention level 2 from memory
+sql select * from ct1;
+print $data00 $data01 $data02
+print $data10 $data11 $data12
+print $data20 $data21 $data22
+if $rows != 3 then
+  print retention level 2 file rows $rows != 3
+  return -1
+endi
+
+if $data01 != 10.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+if $data11 != 1.00000 then
+  return -1
+endi
+if $data12 != NULL then
+  return -1
+endi
+if $data21 != 100.00000 then
+  return -1
+endi
+if $data22 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 1 from memory
+sql select * from ct1 where ts > now-8d;
+print $data00 $data01 $data02
+print $data10 $data11 $data12
+print $data20 $data21 $data22
+
+if $rows != 3 then
+  print retention level 2 file rows $rows != 3
+  return -1
+endi
+
+if $data01 != 10.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+if $data11 != 1.00000 then
+  return -1
+endi
+if $data12 != NULL then
+  return -1
+endi
+if $data21 != 100.00000 then
+  return -1
+endi
+if $data22 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 0 from memory
+sql select * from ct1 where ts > now-3d;
+print $data00 $data01 $data02
+print $data10 $data11 $data12
+print $data20 $data21 $data22
+
+if $rows != 3 then
+  print retention level 2 file rows $rows != 3
+  return -1
+endi
+
+if $data01 != 10.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+if $data11 != 1.00000 then
+  return -1
+endi
+if $data12 != NULL then
+  return -1
+endi
+if $data21 != 100.00000 then
+  return -1
+endi
+if $data22 != NULL then
+  return -1
+endi
+
+print =============== delete row 0
+sql delete from ct1 where ts < now;
+sql delete from ct1 where ts < now;
+sql delete from ct1 where ts < now;
+print =============== wait 7 seconds for results
+sleep 7000
+
+print =============== select * from retention level 2 from memory after delete row 0
+sql select * from ct1;
+print $data00 $data01 $data02
+print $data10 $data11 $data12
+
+if $rows != 2 then
+  print retention level 2 file rows $rows != 2
+  return -1
+endi
+
+if $data01 != 1.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+if $data11 != 100.00000 then
+  return -1
+endi
+if $data12 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 1 from memory after delete row 0
+sql select * from ct1 where ts > now-8d;
+print $data00 $data01 $data02
+print $data10 $data11 $data12
+
+if $rows != 2 then
+  print retention level 2 file rows $rows != 2
+  return -1
+endi
+
+if $data01 != 1.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+if $data11 != 100.00000 then
+  return -1
+endi
+if $data12 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 0 from memory after delete row 0
+sql select * from ct1 where ts > now-3d;
+print $data00 $data01 $data02
+print $data10 $data11 $data12
+
+if $rows != 2 then
+  print retention level 2 file rows $rows != 2
+  return -1
+endi
+
+if $data01 != 1.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+if $data11 != 100.00000 then
+  return -1
+endi
+if $data12 != NULL then
+  return -1
+endi
+
+print =============== delete row 1
+sql delete from ct1 where ts < now;
+sql delete from ct1 where ts < now;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+
+print =============== wait 7 seconds for results
+sleep 7000
+
+print =============== select * from retention level 2 from memory after delete row 1
+sql select * from ct1;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 1 from memory after delete row 1
+sql select * from ct1 where ts > now-8d;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 0 from memory after delete row 1
+sql select * from ct1 where ts > now-3d;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+#===================================================================
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+system sh/exec.sh -n dnode1 -s start
+print =============== wait 7 seconds for results after reboot
+sleep 7000
+
+print =============== select * from retention level 2 from memory after reboot
+sql select * from ct1;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 1 from memory after reboot
+sql select * from ct1 where ts > now-8d;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 0 from memory after reboot
+sql select * from ct1 where ts > now-3d;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+#==================== flush database to trigger commit data to file
+sql flush database d0;
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+system sh/exec.sh -n dnode1 -s start
+
+print =============== select * from retention level 2 from file
+sql select * from ct1 where ts > now-365d;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 1 from file
+sql select * from ct1 where ts > now-8d;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+print =============== select * from retention level 0 from file
+sql select * from ct1 where ts > now-3d;
+print $data00 $data01 $data02
+
+if $rows != 1 then
+  print retention level 2 file rows $rows != 1
+  return -1
+endi
+
+if $data01 != 100.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+
+print =============== delete row 2
+sql delete from ct1 where ts < now;
+sql delete from ct1 where ts < now;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now + 120m;
+sql delete from ct1 where ts < now + 200m;
+sql delete from ct1 where ts < now + 300m;
+sql delete from ct1 where ts < now + 60m;
+sql delete from ct1 where ts < now;
+
+print =============== wait 7 seconds for results
+sleep 7000
+
+print =============== select * from retention level 2 from memory after delete row 2
+sql select * from ct1;
+print $data00 $data01 $data02
+
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+print =============== select * from retention level 1 from memory after delete row 2
+sql select * from ct1 where ts > now-8d;
+print $data00 $data01 $data02
+
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+print =============== select * from retention level 0 from memory after delete row 2
+sql select * from ct1 where ts > now-3d;
+print $data00 $data01 $data02
+
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+#===================================================================
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+system sh/exec.sh -n dnode1 -s start
+print =============== wait 7 seconds for results after reboot
+sleep 7000
+
+print =============== select * from retention level 2 from memory after delete row 2
+sql select * from ct1;
+print $data00 $data01 $data02
+
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+print =============== select * from retention level 1 from memory after delete row 2
+sql select * from ct1 where ts > now-8d;
+print $data00 $data01 $data02
+
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+print =============== select * from retention level 0 from memory after delete row 2
+sql select * from ct1 where ts > now-3d;
+print $data00 $data01 $data02
+
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+print =============== check delete multiple tables
+sql create table ct2 using stb tags("BeiJing", "HaiDian");
+sql create table ct3 using stb tags("ShangHai", "PuDong");
+
+sql insert into ct2 values(now, 10, NULL);
+sql insert into ct2 values(now+60m, 1, NULL);
+sql insert into ct2 values(now+120m, 100, NULL);
+sql insert into ct3 values(now, 10, NULL);
+sql insert into ct3 values(now+60m, 1, NULL);
+sql insert into ct3 values(now+120m, 100, NULL);
+
+print =============== wait 7 seconds for results
+sleep 7000
+
+print =============== select * from retention level 2 from memory
+sql select * from ct2;
+print $data00 $data01 $data02
+print $data10 $data11 $data12
+print $data20 $data21 $data22
+if $rows != 3 then
+  print retention level 2 file rows $rows != 3
+  return -1
+endi
+
+if $data01 != 10.00000 then
+  return -1
+endi
+if $data02 != NULL then
+  return -1
+endi
+if $data11 != 1.00000 then
+  return -1
+endi
+if $data12 != NULL then
+  return -1
+endi
+if $data21 != 100.00000 then
+  return -1
+endi
+if $data22 != NULL then
+  return -1
+endi
+
+sql delete from ct1 where ts < now + 120m;
+sql delete from ct3 where ts < now;
+sql delete from ct2 where ts < now + 60m;
+sql delete from ct2 where ts < now + 120m;
+sql delete from ct3 where ts < now + 60m;
+sql delete from ct3 where ts < now + 120m;
+sql delete from ct3 where ts < now;
+
+print =============== wait 7 seconds for results
+sleep 7000
+
+print =============== select * from retention level 2 from memory after delete ct2
+sql select * from ct2;
+print $data00 $data01 $data02
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+print =============== select * from retention level 1 from memory after delete ct3
+sql select * from ct3 where ts > now - 8d;
+print $data00 $data01 $data02
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+
+#===================================================================
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+#===================================================================
+system sh/exec.sh -n dnode1 -s start
+print =============== wait 7 seconds for results after reboot
+sleep 7000
+
+print =============== select * from retention level 1 from memory after delete ct2
+sql select * from ct2  where ts > now - 8d;
+print $data00 $data01 $data02
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+print =============== select * from retention level 2 from memory after delete ct3
+sql select * from ct3 where ts > now - 365d;
+print $data00 $data01 $data02
+if $rows != 0 then
+  print retention level 2 file rows $rows != 0
+  return -1
+endi
+
+#===================================================================
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+#===================================================================
\ No newline at end of file
diff --git a/tests/script/tsim/testsuit.sim b/tests/script/tsim/testsuit.sim
index 0abe56ab3c..c208a07488 100644
--- a/tests/script/tsim/testsuit.sim
+++ b/tests/script/tsim/testsuit.sim
@@ -130,5 +130,6 @@ run tsim/sync/3Replica1VgElect.sim
 run tsim/sync/threeReplica1VgElectWihtInsert.sim
 run tsim/sma/tsmaCreateInsertQuery.sim
 run tsim/sma/rsmaCreateInsertQuery.sim
+run tsim/sma/rsmaCreateInsertQueryDelete.sim
 run tsim/valgrind/basic.sim
 run tsim/valgrind/checkError.sim
\ No newline at end of file
diff --git a/tests/script/win-test-file b/tests/script/win-test-file
index fe5f5c39e3..b2d50ade8a 100644
--- a/tests/script/win-test-file
+++ b/tests/script/win-test-file
@@ -319,6 +319,7 @@
 ./test.sh -f tsim/sma/sma_leak.sim
 ./test.sh -f tsim/sma/tsmaCreateInsertQuery.sim
 ./test.sh -f tsim/sma/rsmaCreateInsertQuery.sim
+./test.sh -f tsim/sma/rsmaCreateInsertQueryDelete.sim
 ./test.sh -f tsim/sma/rsmaPersistenceRecovery.sim
 ./test.sh -f tsim/sync/vnodesnapshot-rsma-test.sim
 ./test.sh -f tsim/valgrind/checkError1.sim