feat: stream state optimization for rsma

include/util/taoserror.h

@@ -619,6 +619,8 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_RSMA_EMPTY_INFO                TAOS_DEF_ERROR_CODE(0, 0x3156)
 #define TSDB_CODE_RSMA_INVALID_SCHEMA            TAOS_DEF_ERROR_CODE(0, 0x3157)
 #define TSDB_CODE_RSMA_REGEX_MATCH               TAOS_DEF_ERROR_CODE(0, 0x3158)
+#define TSDB_CODE_RSMA_STREAM_STATE_OPEN         TAOS_DEF_ERROR_CODE(0, 0x3159)
+#define TSDB_CODE_RSMA_STREAM_STATE_COMMIT       TAOS_DEF_ERROR_CODE(0, 0x3160)
 
 //index
 #define TSDB_CODE_INDEX_REBUILDING               TAOS_DEF_ERROR_CODE(0, 0x3200)

source/dnode/vnode/src/inc/sma.h

@@ -146,6 +146,7 @@ struct SRSmaInfoItem {
   uint16_t nScanned;
   int32_t  maxDelay;  // ms
   tmr_h    tmrId;
+  void    *pStreamState;
 };
 
 struct SRSmaInfo {
@@ -224,8 +225,10 @@ int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con
 int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type);
 int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
 int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer);
-void    tdRSmaQTaskInfoGetFileName(int32_t vid, int64_t version, char *outputName);
-void    tdRSmaQTaskInfoGetFullName(int32_t vid, int64_t version, const char *path, char *outputName);
+void    tdRSmaQTaskInfoGetFileName(int32_t vgId, int64_t version, char *outputName);
+void    tdRSmaQTaskInfoGetFullName(int32_t vgId, int64_t version, const char *path, char *outputName);
+void    tdRSmaQTaskInfoGetFullPath(int32_t vgId, int8_t level, const char *path, char *outputName);
+void    tdRSmaQTaskInfoGetFullPathEx(int32_t vgId, tb_uid_t suid, int8_t level, const char *path, char *outputName);
 
 static FORCE_INLINE void tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   int32_t ref = T_REF_INC(pRSmaInfo);

source/dnode/vnode/src/sma/smaRollup.c

@@ -92,6 +92,18 @@ void tdRSmaQTaskInfoGetFullName(int32_t vgId, int64_t version, const char *path,
   tdGetVndFileName(vgId, path, VNODE_RSMA_DIR, TD_QTASKINFO_FNAME_PREFIX, version, outputName);
 }
 
+void tdRSmaQTaskInfoGetFullPath(int32_t vgId, int8_t level, const char *path, char *outputName) {
+  tdGetVndDirName(vgId, path, VNODE_RSMA_DIR, true, outputName);
+  int32_t rsmaLen = strlen(outputName);
+  snprintf(outputName + rsmaLen, TSDB_FILENAME_LEN - rsmaLen, "%" PRIi8, level);
+}
+
+void tdRSmaQTaskInfoGetFullPathEx(int32_t vgId, tb_uid_t suid, int8_t level, const char *path, char *outputName) {
+  tdGetVndDirName(vgId, path, VNODE_RSMA_DIR, true, outputName);
+  int32_t rsmaLen = strlen(outputName);
+  snprintf(outputName + rsmaLen, TSDB_FILENAME_LEN - rsmaLen, "%" PRIi64 "%s%" PRIi8, suid, TD_DIRSEP, level);
+}
+
 static FORCE_INLINE int32_t tdRSmaQTaskInfoContLen(int32_t lenWithHead) {
   return lenWithHead - RSMA_QTASKINFO_HEAD_LEN;
 }
@@ -130,6 +142,10 @@ void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree) {
         taosTmrStopA(&pItem->tmrId);
       }
 
+      if (isDeepFree && pItem->pStreamState) {
+        streamStateClose(pItem->pStreamState);
+      }
+
       if (isDeepFree && pInfo->taskInfo[i]) {
         tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], SMA_VID(pSma), i + 1);
       } else {
@@ -290,12 +306,33 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
     SRetention *pRetention = SMA_RETENTION(pSma);
     STsdbCfg   *pTsdbCfg = SMA_TSDB_CFG(pSma);
     SVnode     *pVnode = pSma->pVnode;
+    char        taskInfDir[TSDB_FILENAME_LEN] = {0};
+    void       *pStreamState = NULL;
+
+    // set the backend of stream state
+    tdRSmaQTaskInfoGetFullPathEx(TD_VID(pVnode), pRSmaInfo->suid, idx + 1, tfsGetPrimaryPath(pVnode->pTfs), taskInfDir);
+    if (!taosCheckExistFile(taskInfDir)) {
+      char *s = strdup(taskInfDir);
+      if (taosMulMkDir(taosDirName(s)) != 0) {
+        terrno = TAOS_SYSTEM_ERROR(errno);
+        taosMemoryFree(s);
+        return TSDB_CODE_FAILED;
+      }
+      taosMemoryFree(s);
+    }
+    pStreamState = streamStateOpen(taskInfDir, NULL, true);
+    if (!pStreamState) {
+      terrno = TSDB_CODE_RSMA_STREAM_STATE_OPEN;
+      return TSDB_CODE_FAILED;
+    }
+    
+
     SReadHandle handle = {
         .meta = pVnode->pMeta,
         .vnode = pVnode,
         .initTqReader = 1,
+        .pStateBackend = pStreamState,
     };
-
     pRSmaInfo->taskInfo[idx] = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle);
     if (!pRSmaInfo->taskInfo[idx]) {
       terrno = TSDB_CODE_RSMA_QTASKINFO_CREATE;
@@ -303,6 +340,7 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
     }
     SRSmaInfoItem *pItem = &(pRSmaInfo->items[idx]);
     pItem->triggerStat = TASK_TRIGGER_STAT_ACTIVE;  // fetch the data when reboot
+    pItem->pStreamState = pStreamState;
     if (param->maxdelay[idx] < TSDB_MIN_ROLLUP_MAX_DELAY) {
       int64_t msInterval =
           convertTimeFromPrecisionToUnit(pRetention[idx + 1].freq, pTsdbCfg->precision, TIME_UNIT_MILLISECOND);
@@ -322,7 +360,6 @@ static int32_t tdSetRSmaInfoItemParams(SSma *pSma, SRSmaParam *param, SRSmaStat
 
     pItem->fetchLevel = pItem->level;
     taosTmrReset(tdRSmaFetchTrigger, RSMA_FETCH_INTERVAL, pItem, smaMgmt.tmrHandle, &pItem->tmrId);
-    
 
     smaInfo("vgId:%d, item:%p table:%" PRIi64 " level:%" PRIi8 " maxdelay:%" PRIi64 " watermark:%" PRIi64
             ", finally maxdelay:%" PRIi32,
@@ -1226,16 +1263,17 @@ int32_t tdRSmaProcessRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer)
   if (tdRSmaRestoreQTaskInfoInit(pSma, &nTables) < 0) {
     goto _err;
   }
-
   if (nTables <= 0) {
     smaDebug("vgId:%d, no need to restore rsma task %" PRIi8 " since no tables", SMA_VID(pSma), type);
     return TSDB_CODE_SUCCESS;
   }
 
+#if 0
   // step 2: retrieve qtaskinfo items from the persistence file(rsma/qtaskinfo) and restore
   if (tdRSmaRestoreQTaskInfoReload(pSma, type, qtaskFileVer) < 0) {
     goto _err;
   }
+#endif
 
   // step 3: reload ts data from checkpoint
   if (tdRSmaRestoreTSDataReload(pSma) < 0) {
@@ -1440,6 +1478,50 @@ static int32_t tdRSmaQTaskInfoRestore(SSma *pSma, int8_t type, SRSmaQTaskInfoIte
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
+  SSma   *pSma = pRSmaStat->pSma;
+  SVnode *pVnode = pSma->pVnode;
+  int32_t vid = SMA_VID(pSma);
+
+  if (taosHashGetSize(pInfoHash) <= 0) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  int64_t fsMaxVer = tdRSmaFSMaxVer(pSma, pRSmaStat);
+  if (pRSmaStat->commitAppliedVer <= fsMaxVer) {
+    smaDebug("vgId:%d, rsma persist, no need as applied %" PRIi64 " not larger than fsMaxVer %" PRIi64, vid,
+             pRSmaStat->commitAppliedVer, fsMaxVer);
+    return TSDB_CODE_SUCCESS;
+  }
+
+  void *infoHash = NULL;
+  while ((infoHash = taosHashIterate(pInfoHash, infoHash))) {
+    SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
+
+    if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
+      continue;
+    }
+
+    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
+      SRSmaInfoItem *pItem = RSMA_INFO_ITEM(pRSmaInfo, i);
+      if (pItem && pItem->pStreamState) {
+        if (streamStateCommit(pItem->pStreamState) < 0) {
+          terrno = TSDB_CODE_RSMA_STREAM_STATE_COMMIT;
+          goto _err;
+        }
+        smaDebug("vgId:%d, rsma persist, stream state commit success, table %" PRIi64 " level %d", vid, pRSmaInfo->suid,
+                 i + 1);
+      }
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+_err:
+  smaError("vgId:%d, rsma persist failed since %s", vid, terrstr());
+  return TSDB_CODE_FAILED;
+}
+
+#if 0
 int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
   SSma   *pSma = pRSmaStat->pSma;
   SVnode *pVnode = pSma->pVnode;
@@ -1459,7 +1541,7 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash) {
   int64_t fsMaxVer = tdRSmaFSMaxVer(pSma, pRSmaStat);
   if (pRSmaStat->commitAppliedVer <= fsMaxVer) {
     smaDebug("vgId:%d, rsma persist, no need as applied %" PRIi64 " not larger than fsMaxVer %" PRIi64, vid,
-            pRSmaStat->commitAppliedVer, fsMaxVer);
+             pRSmaStat->commitAppliedVer, fsMaxVer);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -1579,6 +1661,8 @@ _err:
   return TSDB_CODE_FAILED;
 }
 
+#endif
+
 /**
  * @brief trigger to get rsma result in async mode
  *
@@ -1926,7 +2010,7 @@ int32_t tdRSmaProcessExecImpl(SSma *pSma, ERsmaExecType type) {
 
       if ((pEnv->flag & SMA_ENV_FLG_CLOSE) && (atomic_load_64(&pRSmaStat->nBufItems) <= 0)) {
         smaDebug("vgId:%d, exec task end, flag:%" PRIi8 ", nBufItems:%" PRIi64, SMA_VID(pSma), pEnv->flag,
-                atomic_load_64(&pRSmaStat->nBufItems));
+                 atomic_load_64(&pRSmaStat->nBufItems));
         break;
       }
     }

source/util/src/terror.c

@@ -621,6 +621,8 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP,        "Rsma fetch msg is m
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_EMPTY_INFO,                "Rsma info is empty")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_INVALID_SCHEMA,            "Rsma invalid schema")
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_REGEX_MATCH,               "Rsma regex match")
+TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_STREAM_STATE_OPEN,         "Rsma stream state open")
+TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_STREAM_STATE_COMMIT,       "Rsma stream state commit")
 
 //index
 TAOS_DEFINE_ERROR(TSDB_CODE_INDEX_REBUILDING,               "Index is rebuilding")