From dbef3dfc55d19e1d3e182ef61f23629fe7261eff Mon Sep 17 00:00:00 2001
From: Cary Xu <klxu@taosdata.com>
Date: Thu, 16 Jun 2022 15:34:33 +0800
Subject: [PATCH] fix: not submit for rsma without rollup

---
 source/common/src/tdatablock.c         | 44 ++++++++++++++++----------
 source/dnode/vnode/src/inc/vnodeInt.h  |  1 +
 source/dnode/vnode/src/sma/smaRollup.c | 14 ++++++--
 3 files changed, 39 insertions(+), 20 deletions(-)

diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c
index 5a2aaed74e..3ec1b98b16 100644
--- a/source/common/src/tdatablock.c
+++ b/source/common/src/tdatablock.c
@@ -1583,6 +1583,11 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
     int32_t      rowSize = pDataBlock->info.rowSize;
     int64_t      groupId = pDataBlock->info.groupId;
 
+    if (colNum <= 1) {
+      // invalid if only with TS col
+      continue;
+    }
+
     if (rb.nCols != colNum) {
       tdSRowSetTpInfo(&rb, colNum, pTSchema->flen);
     }
@@ -1679,23 +1684,28 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
     msgLen += pSubmitBlk->dataLen;
   }
 
-  (*pReq)->length = msgLen;
-
-  (*pReq)->header.vgId = htonl(vgId);
-  (*pReq)->header.contLen = htonl(msgLen);
-  (*pReq)->length = (*pReq)->header.contLen;
-  (*pReq)->numOfBlocks = htonl(numOfBlks);
-  SSubmitBlk* blk = (SSubmitBlk*)((*pReq) + 1);
-  while (numOfBlks--) {
-    int32_t dataLen = blk->dataLen;
-    blk->uid = htobe64(blk->uid);
-    blk->suid = htobe64(blk->suid);
-    blk->padding = htonl(blk->padding);
-    blk->sversion = htonl(blk->sversion);
-    blk->dataLen = htonl(blk->dataLen);
-    blk->schemaLen = htonl(blk->schemaLen);
-    blk->numOfRows = htons(blk->numOfRows);
-    blk = (SSubmitBlk*)(blk->data + dataLen);
+  if (numOfBlks > 0) {
+    (*pReq)->length = msgLen;
+
+    (*pReq)->header.vgId = htonl(vgId);
+    (*pReq)->header.contLen = htonl(msgLen);
+    (*pReq)->length = (*pReq)->header.contLen;
+    (*pReq)->numOfBlocks = htonl(numOfBlks);
+    SSubmitBlk* blk = (SSubmitBlk*)((*pReq) + 1);
+    while (numOfBlks--) {
+      int32_t dataLen = blk->dataLen;
+      blk->uid = htobe64(blk->uid);
+      blk->suid = htobe64(blk->suid);
+      blk->padding = htonl(blk->padding);
+      blk->sversion = htonl(blk->sversion);
+      blk->dataLen = htonl(blk->dataLen);
+      blk->schemaLen = htonl(blk->schemaLen);
+      blk->numOfRows = htons(blk->numOfRows);
+      blk = (SSubmitBlk*)(blk->data + dataLen);
+    }
+  } else {
+    // no valid rows
+    taosMemoryFreeClear(*pReq);
   }
 
   return TSDB_CODE_SUCCESS;
diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h
index 2fd0815181..553c6a40ab 100644
--- a/source/dnode/vnode/src/inc/vnodeInt.h
+++ b/source/dnode/vnode/src/inc/vnodeInt.h
@@ -260,6 +260,7 @@ struct SSma {
 
 #define SMA_CFG(s)        (&(s)->pVnode->config)
 #define SMA_TSDB_CFG(s)   (&(s)->pVnode->config.tsdbCfg)
+#define SMA_RETENTION(s)  ((SRetention *)&(s)->pVnode->config.tsdbCfg.retentions)
 #define SMA_LOCKED(s)     ((s)->locked)
 #define SMA_META(s)       ((s)->pVnode->pMeta)
 #define SMA_VID(s)        TD_VID((s)->pVnode)
diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c
index b2dcce8f4c..0c372dfa70 100644
--- a/source/dnode/vnode/src/sma/smaRollup.c
+++ b/source/dnode/vnode/src/sma/smaRollup.c
@@ -400,22 +400,24 @@ static FORCE_INLINE int32_t tdExecuteRSmaImpl(SSma *pSma, const void *pMsg, int3
   }
 
   if (taosArrayGetSize(pResult) > 0) {
-#if 1
+#if 0
     char flag[10] = {0};
     snprintf(flag, 10, "level %" PRIi8, level);
     blockDebugShowData(pResult, flag);
 #endif
     STsdb      *sinkTsdb = (level == TSDB_RETENTION_L1 ? pSma->pRSmaTsdb1 : pSma->pRSmaTsdb2);
     SSubmitReq *pReq = NULL;
-    if (buildSubmitReqFromDataBlock(&pReq, pResult, pTSchema, SMA_VID(pSma), suid) != 0) {
+    if (buildSubmitReqFromDataBlock(&pReq, pResult, pTSchema, SMA_VID(pSma), suid) < 0) {
       taosArrayDestroy(pResult);
       return TSDB_CODE_FAILED;
     }
-    if (tdProcessSubmitReq(sinkTsdb, INT64_MAX, pReq) != 0) {
+    
+    if (pReq && tdProcessSubmitReq(sinkTsdb, INT64_MAX, pReq) < 0) {
       taosArrayDestroy(pResult);
       taosMemoryFreeClear(pReq);
       return TSDB_CODE_FAILED;
     }
+
     taosMemoryFreeClear(pReq);
   } else {
     smaDebug("vgId:%d, no rsma % " PRIi8 " data generated since %s", SMA_VID(pSma), level, tstrerror(terrno));
@@ -469,6 +471,12 @@ int32_t tdProcessRSmaSubmit(SSma *pSma, void *pMsg, int32_t inputType) {
     return TSDB_CODE_SUCCESS;
   }
 
+  SRetention *pRetention = SMA_RETENTION(pSma);
+  if (!RETENTION_VALID(pRetention + 1)) {
+    // return directly if retention level 1 is invalid
+    return TSDB_CODE_SUCCESS;
+  }
+
   if (inputType == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
     STbUidStore uidStore = {0};
     tdFetchSubmitReqSuids(pMsg, &uidStore);
-- 
GitLab