feat: tsma days refactor

include/common/tmsg.h

@@ -2352,19 +2352,19 @@ typedef struct {
   STSma*  tSma;
 } STSmaWrapper;
 
-static FORCE_INLINE void tdDestroyTSma(STSma* pSma) {
+static FORCE_INLINE void tDestroyTSma(STSma* pSma) {
   if (pSma) {
     taosMemoryFreeClear(pSma->expr);
     taosMemoryFreeClear(pSma->tagsFilter);
   }
 }
 
-static FORCE_INLINE void tdDestroyTSmaWrapper(STSmaWrapper* pSW, bool deepCopy) {
+static FORCE_INLINE void tDestroyTSmaWrapper(STSmaWrapper* pSW, bool deepCopy) {
   if (pSW) {
     if (pSW->tSma) {
       if (deepCopy) {
         for (uint32_t i = 0; i < pSW->number; ++i) {
-          tdDestroyTSma(pSW->tSma + i);
+          tDestroyTSma(pSW->tSma + i);
         }
       }
       taosMemoryFreeClear(pSW->tSma);
@@ -2372,8 +2372,8 @@ static FORCE_INLINE void tdDestroyTSmaWrapper(STSmaWrapper* pSW, bool deepCopy)
   }
 }
 
-static FORCE_INLINE void* tdFreeTSmaWrapper(STSmaWrapper* pSW, bool deepCopy) {
-  tdDestroyTSmaWrapper(pSW, deepCopy);
+static FORCE_INLINE void* tFreeTSmaWrapper(STSmaWrapper* pSW, bool deepCopy) {
+  tDestroyTSmaWrapper(pSW, deepCopy);
   taosMemoryFreeClear(pSW);
   return NULL;
 }

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -194,6 +194,15 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SVnodeCfg vnodeCfg = {0};
   vmGenerateVnodeCfg(&createReq, &vnodeCfg);
 
+  if (createReq.isTsma) {
+    SMsgHead *smaMsg = createReq.pTsma;
+    uint32_t  contLen = (uint32_t)(htonl(smaMsg->contLen) - sizeof(SMsgHead));
+    if (smaGetTSmaDays(&vnodeCfg, POINTER_SHIFT(smaMsg, sizeof(SMsgHead)), contLen, &vnodeCfg.tsdbCfg.days) < 0) {
+      dError("vgId:%d, failed to get tsma days since %s", createReq.vgId, terrstr());
+      return -1;
+    }
+  }
+
   SWrapperCfg wrapperCfg = {0};
   vmGenerateWrapperCfg(pMgmt, &createReq, &wrapperCfg);
 

source/dnode/vnode/CMakeLists.txt

@@ -32,7 +32,7 @@ target_sources(
     "src/sma/smaEnv.c"
     "src/sma/smaOpen.c"
     "src/sma/smaRollup.c"
-    "src/sma/smaTimeRange.c"
+    "src/sma/smaTimeRange2.c"
 
     # tsdb
     # "src/tsdb/tsdbTDBImpl.c"

source/dnode/vnode/inc/vnode.h

@@ -146,6 +146,9 @@ bool    tqNextDataBlockFilterOut(STqReadHandle *pHandle, SHashObj *filterOutUids
 int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t *pUid,
                             int32_t *pNumOfRows, int16_t *pNumOfCols);
 
+// sma
+int32_t smaGetTSmaDays(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
+
 // need to reposition
 
 // structs

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

@@ -223,6 +223,8 @@ int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version)
 // TODO: This is the basic params, and should wrap the params to a queryHandle.
 int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult);
 
+int32_t tdGetTSmaDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
+
 #ifdef __cplusplus
 }
 #endif

source/dnode/vnode/src/meta/metaQuery.c

@@ -464,7 +464,7 @@ STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) {
 _err:
   metaReaderClear(&mr);
   taosArrayDestroy(pSmaIds);
-  tdFreeTSmaWrapper(pSW, deepCopy);
+  tFreeTSmaWrapper(pSW, deepCopy);
   return NULL;
 }
 

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

@@ -52,3 +52,11 @@ int32_t tdGetTSmaData(SSma* pSma, char* pData, int64_t indexUid, TSKEY querySKey
   }
   return code;
 }
+
+int32_t smaGetTSmaDays(SVnodeCfg* pCfg, void* pCont, uint32_t contLen, int32_t *days) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if ((code = tdGetTSmaDaysImpl(pCfg, pCont, contLen, days)) < 0) {
+    smaWarn("vgId:%d get tSma days failed since %s", pCfg->vgId, tstrerror(terrno));
+  }
+  return code;
+}

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

@@ -278,7 +278,7 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType) {
 
 void *tdFreeSmaStatItem(SSmaStatItem *pSmaStatItem) {
   if (pSmaStatItem) {
-    tdDestroyTSma(pSmaStatItem->pTSma);
+    tDestroyTSma(pSmaStatItem->pTSma);
     taosMemoryFreeClear(pSmaStatItem->pTSma);
     taosHashCleanup(pSmaStatItem->expiredWindows);
     taosMemoryFreeClear(pSmaStatItem);

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

@@ -828,7 +828,7 @@ int32_t tdDropTSma(SSma *pSma, char *pMsg) {
 
   // TODO: send msg to stream computing to drop tSma
   // if ((send msg to stream computing) < 0) {
-  //   tdDestroyTSma(&vCreateSmaReq);
+  //   tDestroyTSma(&vCreateSmaReq);
   //   return -1;
   // }
   //
@@ -982,25 +982,25 @@ int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version)
 
     SSubmitBlkIter blkIter = {0};
     if (tInitSubmitBlkIter(&msgIter, pBlock, &blkIter) < 0) {
-      pSW = tdFreeTSmaWrapper(pSW, false);
+      pSW = tFreeTSmaWrapper(pSW, false);
       break;
     }
 
     while (true) {
       STSRow *row = tGetSubmitBlkNext(&blkIter);
       if (!row) {
-        pSW = tdFreeTSmaWrapper(pSW, false);
+        pSW = tFreeTSmaWrapper(pSW, false);
         break;
       }
       if (!pSW || (pTSma && (pTSma->tableUid != msgIter.suid))) {
         if (pSW) {
-          pSW = tdFreeTSmaWrapper(pSW, false);
+          pSW = tFreeTSmaWrapper(pSW, false);
         }
         if (!(pSW = metaGetSmaInfoByTable(SMA_META(pSma), msgIter.suid, false))) {
           break;
         }
         if ((pSW->number) <= 0 || !pSW->tSma) {
-          pSW = tdFreeTSmaWrapper(pSW, false);
+          pSW = tFreeTSmaWrapper(pSW, false);
           break;
         }
 
@@ -1020,7 +1020,7 @@ int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version)
       if (lastWinSKey != winSKey) {
         lastWinSKey = winSKey;
         if (tdSetExpiredWindow(pSma, pItemsHash, pTSma->indexUid, winSKey, version) < 0) {
-          pSW = tdFreeTSmaWrapper(pSW, false);
+          pSW = tFreeTSmaWrapper(pSW, false);
           tdUnRefSmaStat(pSma, pStat);
           return TSDB_CODE_FAILED;
         }

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "sma.h"
+#include "tsdb.h"
+
+typedef STsdbCfg STSmaKeepCfg;
+
+#undef _TEST_SMA_PRINT_DEBUG_LOG_
+#define SMA_STORAGE_MINUTES_MAX  86400
+#define SMA_STORAGE_MINUTES_DAY  1440
+#define SMA_STORAGE_MINUTES_MIN  1440
+#define SMA_STORAGE_TSDB_MINUTES 86400
+#define SMA_STORAGE_TSDB_TIMES   10
+#define SMA_STORAGE_SPLIT_FACTOR 14400  // least records in tsma file TODO: the feasible value?
+#define SMA_KEY_LEN              16     // TSKEY+groupId 8+8
+#define SMA_DROP_EXPIRED_TIME    10     // default is 10 seconds
+
+#define SMA_STATE_ITEM_HASH_SLOT 32
+
+typedef struct {
+  SSma         *pSma;
+  SDBFile       dFile;
+  const SArray *pDataBlocks;  // sma data
+  int64_t       interval;     // interval with the precision of DB
+} STSmaWriteH;
+
+typedef struct {
+  int32_t iter;
+  int32_t fid;
+} SmaFsIter;
+
+typedef struct {
+  STsdb    *pTsdb;
+  SSma     *pSma;
+  SDBFile   dFile;
+  int64_t   interval;   // interval with the precision of DB
+  int32_t   blockSize;  // size of SMA block item
+  int32_t   days;
+  int8_t    storageLevel;
+  SmaFsIter smaFsIter;
+} STSmaReadH;
+
+typedef enum {
+  SMA_STORAGE_LEVEL_TSDB = 0,     // use days of self-defined  e.g. vnode${N}/tsdb/tsma/sma_index_uid/v2f200.tsma
+  SMA_STORAGE_LEVEL_DFILESET = 1  // use days of TS data       e.g. vnode${N}/tsdb/tsma/sma_index_uid/v2f1906.tsma
+} ESmaStorageLevel;
+
+// static func
+
+static int64_t tdGetIntervalByPrecision(int64_t interval, uint8_t intervalUnit, int8_t precision, bool adjusted);
+static int32_t tdGetSmaStorageLevel(STSmaKeepCfg *pCfg, int64_t interval);
+static int32_t tdInitTSmaWriteH(STSmaWriteH *pSmaH, SSma *pSma, const SArray *pDataBlocks, int64_t interval,
+                                int8_t intervalUnit);
+static int32_t tdInitTSmaReadH(STSmaReadH *pSmaH, SSma *pSma, int64_t interval, int8_t intervalUnit);
+static void    tdDestroyTSmaWriteH(STSmaWriteH *pSmaH);
+static int32_t tdGetTSmaDays(SSma *pSma, int64_t interval, int32_t storageLevel);
+static int32_t tdSetTSmaDataFile(STSmaWriteH *pSmaH, int64_t indexUid, int32_t fid);
+static int32_t tdInitTSmaFile(STSmaReadH *pSmaH, int64_t indexUid, TSKEY skey);
+static bool    tdSetAndOpenTSmaFile(STSmaReadH *pReadH, TSKEY *queryKey);
+static int32_t tdInsertTSmaBlocks(STSmaWriteH *pSmaH, void *smaKey, int32_t keyLen, void *pData, int32_t dataLen,
+                                  TXN *txn);
+// expired window
+
+static int32_t tdSetExpiredWindow(SSma *pSma, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey, int64_t version);
+static int32_t tdResetExpiredWindow(SSma *pSma, SSmaStat *pStat, int64_t indexUid, TSKEY skey);
+static int32_t tdDropTSmaDataImpl(SSma *pSma, int64_t indexUid);
+
+/**
+ * @brief Judge the tsma file split days
+ *
+ * @param pCfg
+ * @param pCont
+ * @param contLen
+ * @param days unit is minute
+ * @return int32_t
+ */
+int32_t tdGetTSmaDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days) {
+  SDecoder coder = {0};
+  tDecoderInit(&coder, pCont, contLen);
+
+  STSma tsma = {0};
+  if (tDecodeSVCreateTSmaReq(&coder, &tsma) < 0) {
+    terrno = TSDB_CODE_MSG_DECODE_ERROR;
+    goto _err;
+  }
+  STsdbCfg *pTsdbCfg = &pCfg->tsdbCfg;
+  int64_t   mInterval = convertTimeFromPrecisionToUnit(tsma.interval, pTsdbCfg->precision, TIME_UNIT_MINUTE);
+  int64_t records = pTsdbCfg->days / mInterval;
+
+  if (records >= SMA_STORAGE_SPLIT_FACTOR) {
+    *days = pTsdbCfg->days;
+  } else {
+    int64_t daysPerFile = mInterval * SMA_STORAGE_MINUTES_DAY * 2;
+
+    if (daysPerFile > SMA_STORAGE_MINUTES_MAX) {
+      *days = SMA_STORAGE_MINUTES_MAX;
+    } else {
+      *days = (int32_t)daysPerFile;
+    }
+
+    if(*days < pTsdbCfg->days) {
+      *days = pTsdbCfg->days;
+    }
+  }
+  tDecoderClear(&coder);
+  return 0;
+_err:
+  tDecoderClear(&coder);
+  return -1;
+}
+
+// read data
+
+// implementation
+
+/**
+ * @brief
+ *
+ * @param pSmaH
+ * @param pSma
+ * @param interval
+ * @param intervalUnit
+ * @return int32_t
+ */
+static int32_t tdInitTSmaReadH(STSmaReadH *pSmaH, SSma *pSma, int64_t interval, int8_t intervalUnit) {
+  STSmaKeepCfg *pCfg = SMA_TSDB_CFG(pSma);
+  pSmaH->pSma = pSma;
+  pSmaH->interval = tdGetIntervalByPrecision(interval, intervalUnit, SMA_TSDB_CFG(pSma)->precision, true);
+  pSmaH->storageLevel = tdGetSmaStorageLevel(pCfg, interval);
+  pSmaH->days = tdGetTSmaDays(pSma, pSmaH->interval, pSmaH->storageLevel);
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief Init of tSma FS
+ *
+ * @param pReadH
+ * @param indexUid
+ * @param skey
+ * @return int32_t
+ */
+static int32_t tdInitTSmaFile(STSmaReadH *pSmaH, int64_t indexUid, TSKEY skey) {
+  SSma *pSma = pSmaH->pSma;
+
+  int32_t fid = (int32_t)(TSDB_KEY_FID(skey, pSmaH->days, SMA_TSDB_CFG(pSma)->precision));
+  char    tSmaFile[TSDB_FILENAME_LEN] = {0};
+  snprintf(tSmaFile, TSDB_FILENAME_LEN, "%" PRIi64 "%sv%df%d.tsma", indexUid, TD_DIRSEP, SMA_VID(pSma), fid);
+  pSmaH->dFile.path = strdup(tSmaFile);
+  pSmaH->smaFsIter.iter = 0;
+  pSmaH->smaFsIter.fid = fid;
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief Set and open tSma file if it has key locates in queryWin.
+ *
+ * @param pReadH
+ * @param param
+ * @param queryWin
+ * @return true
+ * @return false
+ */
+static bool tdSetAndOpenTSmaFile(STSmaReadH *pReadH, TSKEY *queryKey) {
+  // SArray *smaFs = pReadH->pTsdb->fs->cstatus->sf;
+  // int32_t nSmaFs = taosArrayGetSize(smaFs);
+
+  smaCloseDBF(&pReadH->dFile);
+
+#if 0
+  while (pReadH->smaFsIter.iter < nSmaFs) {
+    void *pSmaFile = taosArrayGet(smaFs, pReadH->smaFsIter.iter);
+    if (pSmaFile) {  // match(indexName, queryWindow)
+      // TODO: select the file by index_name ...
+      pReadH->dFile = pSmaFile;
+      ++pReadH->smaFsIter.iter;
+      break;
+    }
+    ++pReadH->smaFsIter.iter;
+  }
+
+  if (pReadH->pDFile) {
+    tdDebug("vg%d: smaFile %s matched", REPO_ID(pReadH->pTsdb), "[pSmaFile dir]");
+    return true;
+  }
+#endif
+
+  return false;
+}
+
+/**
+ * @brief Approximate value for week/month/year.
+ *
+ * @param interval
+ * @param intervalUnit
+ * @param precision
+ * @param adjusted Interval already adjusted according to DB precision
+ * @return int64_t
+ */
+static int64_t tdGetIntervalByPrecision(int64_t interval, uint8_t intervalUnit, int8_t precision, bool adjusted) {
+  if (adjusted) {
+    return interval;
+  }
+
+  switch (intervalUnit) {
+    case TIME_UNIT_YEAR:  // approximate value
+      interval *= 365 * 86400 * 1e3;
+      break;
+    case TIME_UNIT_MONTH:  // approximate value
+      interval *= 30 * 86400 * 1e3;
+      break;
+    case TIME_UNIT_WEEK:  // approximate value
+      interval *= 7 * 86400 * 1e3;
+      break;
+    case TIME_UNIT_DAY:  // the interval for tSma calculation must <= day
+      interval *= 86400 * 1e3;
+      break;
+    case TIME_UNIT_HOUR:
+      interval *= 3600 * 1e3;
+      break;
+    case TIME_UNIT_MINUTE:
+      interval *= 60 * 1e3;
+      break;
+    case TIME_UNIT_SECOND:
+      interval *= 1e3;
+      break;
+    default:
+      break;
+  }
+
+  switch (precision) {
+    case TSDB_TIME_PRECISION_MILLI:
+      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
+        return interval / 1e3;
+      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  //  nano second
+        return interval / 1e6;
+      } else {  // ms
+        return interval;
+      }
+      break;
+    case TSDB_TIME_PRECISION_MICRO:
+      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
+        return interval;
+      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  //  ns
+        return interval / 1e3;
+      } else {  // ms
+        return interval * 1e3;
+      }
+      break;
+    case TSDB_TIME_PRECISION_NANO:
+      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
+        return interval * 1e3;
+      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  // ns
+        return interval;
+      } else {  // ms
+        return interval * 1e6;
+      }
+      break;
+    default:                                        // ms
+      if (TIME_UNIT_MICROSECOND == intervalUnit) {  // us
+        return interval / 1e3;
+      } else if (TIME_UNIT_NANOSECOND == intervalUnit) {  //  ns
+        return interval / 1e6;
+      } else {  // ms
+        return interval;
+      }
+      break;
+  }
+  return interval;
+}
+
+static int32_t tdInitTSmaWriteH(STSmaWriteH *pSmaH, SSma *pSma, const SArray *pDataBlocks, int64_t interval,
+                                int8_t intervalUnit) {
+  pSmaH->pSma = pSma;
+  pSmaH->interval = tdGetIntervalByPrecision(interval, intervalUnit, SMA_TSDB_CFG(pSma)->precision, true);
+  pSmaH->pDataBlocks = pDataBlocks;
+  pSmaH->dFile.fid = SMA_IVLD_FID;
+  return TSDB_CODE_SUCCESS;
+}
+
+static void tdDestroyTSmaWriteH(STSmaWriteH *pSmaH) {
+  if (pSmaH) {
+    smaCloseDBF(&pSmaH->dFile);
+  }
+}
+
+static int32_t tdSetTSmaDataFile(STSmaWriteH *pSmaH, int64_t indexUid, int32_t fid) {
+  SSma *pSma = pSmaH->pSma;
+  ASSERT(!pSmaH->dFile.path && !pSmaH->dFile.pDB);
+
+  pSmaH->dFile.fid = fid;
+  char tSmaFile[TSDB_FILENAME_LEN] = {0};
+  snprintf(tSmaFile, TSDB_FILENAME_LEN, "%" PRIi64 "%sv%df%d.tsma", indexUid, TD_DIRSEP, SMA_VID(pSma), fid);
+  pSmaH->dFile.path = strdup(tSmaFile);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief
+ *
+ * @param pSma
+ * @param interval Interval calculated by DB's precision
+ * @param storageLevel
+ * @return int32_t
+ */
+static int32_t tdGetTSmaDays(SSma *pSma, int64_t interval, int32_t storageLevel) {
+  STsdbCfg *pCfg = SMA_TSDB_CFG(pSma);
+  int32_t   daysPerFile = pCfg->days;  // unit is minute
+
+  if (storageLevel == SMA_STORAGE_LEVEL_TSDB) {
+    int32_t minutes = SMA_STORAGE_TSDB_TIMES * (interval / tsTickPerMin[pCfg->precision]);
+    if (minutes > SMA_STORAGE_TSDB_MINUTES) {
+      daysPerFile = SMA_STORAGE_TSDB_MINUTES;
+    }
+  }
+
+  return daysPerFile;
+}
+
+/**
+ * @brief Judge the tSma storage level
+ *
+ * @param pCfg
+ * @param interval
+ * @return int32_t
+ */
+static int32_t tdGetSmaStorageLevel(STSmaKeepCfg *pCfg, int64_t interval) {
+  int64_t mInterval = convertTimeFromPrecisionToUnit(interval, pCfg->precision, TIME_UNIT_MINUTE);
+  if (pCfg->days / mInterval >= SMA_STORAGE_SPLIT_FACTOR) {
+    return SMA_STORAGE_LEVEL_DFILESET;
+  }
+  return SMA_STORAGE_LEVEL_TSDB;
+}
+
+/**
+ * @brief Insert/Update Time-range-wise SMA data.
+ *  - If interval < SMA_STORAGE_SPLIT_HOURS(e.g. 24), save the SMA data as a part of DFileSet to e.g.
+ * v3f1900.tsma.${sma_index_name}. The days is the same with that for TS data files.
+ *  - If interval >= SMA_STORAGE_SPLIT_HOURS, save the SMA data to e.g. vnode3/tsma/v3f632.tsma.${sma_index_name}. The
+ * days is 30 times of the interval, and the minimum days is SMA_STORAGE_TSDB_DAYS(30d).
+ *  - The destination file of one data block for some interval is determined by its start TS key.
+ *
+ * @param pSma
+ * @param msg
+ * @return int32_t
+ */
+int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
+  STsdbCfg     *pCfg = SMA_TSDB_CFG(pSma);
+  const SArray *pDataBlocks = (const SArray *)msg;
+  int64_t       testSkey = TSKEY_INITIAL_VAL;
+
+  // TODO: destroy SSDataBlocks(msg)
+
+  // For super table aggregation, the sma data is stored in vgroup calculated from the hash value of stable name. Thus
+  // the sma data would arrive ahead of the update-expired-window msg.
+  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_TIME_RANGE) != TSDB_CODE_SUCCESS) {
+    terrno = TSDB_CODE_TDB_INIT_FAILED;
+    return TSDB_CODE_FAILED;
+  }
+
+  if (!pDataBlocks) {
+    terrno = TSDB_CODE_INVALID_PTR;
+    smaWarn("vgId:%d insert tSma data failed since pDataBlocks is NULL", SMA_VID(pSma));
+    return terrno;
+  }
+
+  if (taosArrayGetSize(pDataBlocks) <= 0) {
+    terrno = TSDB_CODE_INVALID_PARA;
+    smaWarn("vgId:%d insert tSma data failed since pDataBlocks is empty", SMA_VID(pSma));
+    return TSDB_CODE_FAILED;
+  }
+
+  SSmaEnv      *pEnv = SMA_TSMA_ENV(pSma);
+  SSmaStat     *pStat = SMA_ENV_STAT(pEnv);
+  SSmaStatItem *pItem = NULL;
+
+  tdRefSmaStat(pSma, pStat);
+
+  if (pStat && SMA_STAT_ITEMS(pStat)) {
+    pItem = taosHashGet(SMA_STAT_ITEMS(pStat), &indexUid, sizeof(indexUid));
+  }
+
+  if (!pItem || !(pItem = *(SSmaStatItem **)pItem) || tdSmaStatIsDropped(pItem)) {
+    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
+    tdUnRefSmaStat(pSma, pStat);
+    return TSDB_CODE_FAILED;
+  }
+
+  STSma      *pTSma = pItem->pTSma;
+  STSmaWriteH tSmaH = {0};
+
+  if (tdInitTSmaWriteH(&tSmaH, pSma, pDataBlocks, pTSma->interval, pTSma->intervalUnit) != 0) {
+    return TSDB_CODE_FAILED;
+  }
+
+  char rPath[TSDB_FILENAME_LEN] = {0};
+  char aPath[TSDB_FILENAME_LEN] = {0};
+  snprintf(rPath, TSDB_FILENAME_LEN, "%s%s%" PRIi64, SMA_ENV_PATH(pEnv), TD_DIRSEP, indexUid);
+  tfsAbsoluteName(SMA_TFS(pSma), SMA_ENV_DID(pEnv), rPath, aPath);
+  if (!taosCheckExistFile(aPath)) {
+    if (tfsMkdirRecurAt(SMA_TFS(pSma), rPath, SMA_ENV_DID(pEnv)) != TSDB_CODE_SUCCESS) {
+      tdUnRefSmaStat(pSma, pStat);
+      return TSDB_CODE_FAILED;
+    }
+  }
+
+  // Step 1: Judge the storage level and days
+  int32_t storageLevel = tdGetSmaStorageLevel(pCfg, tSmaH.interval);
+  int32_t minutePerFile = tdGetTSmaDays(pSma, tSmaH.interval, storageLevel);
+
+  char    smaKey[SMA_KEY_LEN] = {0};  // key: skey + groupId
+  char    dataBuf[512] = {0};         // val: aggr data // TODO: handle 512 buffer?
+  void   *pDataBuf = NULL;
+  int32_t sz = taosArrayGetSize(pDataBlocks);
+  for (int32_t i = 0; i < sz; ++i) {
+    SSDataBlock *pDataBlock = taosArrayGet(pDataBlocks, i);
+    int32_t      colNum = pDataBlock->info.numOfCols;
+    int32_t      rows = pDataBlock->info.rows;
+    int32_t      rowSize = pDataBlock->info.rowSize;
+    int64_t      groupId = pDataBlock->info.groupId;
+    for (int32_t j = 0; j < rows; ++j) {
+      printf("|");
+      TSKEY skey = TSKEY_INITIAL_VAL;  //  the start key of TS window by interval
+      void *pSmaKey = &smaKey;
+      bool  isStartKey = false;
+
+      int32_t tlen = 0;     // reset the len
+      pDataBuf = &dataBuf;  // reset the buf
+      for (int32_t k = 0; k < colNum; ++k) {
+        SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
+        void            *var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
+        switch (pColInfoData->info.type) {
+          case TSDB_DATA_TYPE_TIMESTAMP:
+            if (!isStartKey) {
+              isStartKey = true;
+              skey = *(TSKEY *)var;
+              testSkey = skey;
+              printf("= skey %" PRIi64 " groupId = %" PRIi64 "|", skey, groupId);
+              tdEncodeTSmaKey(groupId, skey, &pSmaKey);
+            } else {
+              printf(" %" PRIi64 " |", *(int64_t *)var);
+              tlen += taosEncodeFixedI64(&pDataBuf, *(int64_t *)var);
+              break;
+            }
+            break;
+          case TSDB_DATA_TYPE_BOOL:
+          case TSDB_DATA_TYPE_UTINYINT:
+            printf(" %15d |", *(uint8_t *)var);
+            tlen += taosEncodeFixedU8(&pDataBuf, *(uint8_t *)var);
+            break;
+          case TSDB_DATA_TYPE_TINYINT:
+            printf(" %15d |", *(int8_t *)var);
+            tlen += taosEncodeFixedI8(&pDataBuf, *(int8_t *)var);
+            break;
+          case TSDB_DATA_TYPE_SMALLINT:
+            printf(" %15d |", *(int16_t *)var);
+            tlen += taosEncodeFixedI16(&pDataBuf, *(int16_t *)var);
+            break;
+          case TSDB_DATA_TYPE_USMALLINT:
+            printf(" %15d |", *(uint16_t *)var);
+            tlen += taosEncodeFixedU16(&pDataBuf, *(uint16_t *)var);
+            break;
+          case TSDB_DATA_TYPE_INT:
+            printf(" %15d |", *(int32_t *)var);
+            tlen += taosEncodeFixedI32(&pDataBuf, *(int32_t *)var);
+            break;
+          case TSDB_DATA_TYPE_FLOAT:
+            printf(" %15f |", *(float *)var);
+            tlen += taosEncodeBinary(&pDataBuf, var, sizeof(float));
+            break;
+          case TSDB_DATA_TYPE_UINT:
+            printf(" %15u |", *(uint32_t *)var);
+            tlen += taosEncodeFixedU32(&pDataBuf, *(uint32_t *)var);
+            break;
+          case TSDB_DATA_TYPE_BIGINT:
+            printf(" %15ld |", *(int64_t *)var);
+            tlen += taosEncodeFixedI64(&pDataBuf, *(int64_t *)var);
+            break;
+          case TSDB_DATA_TYPE_DOUBLE:
+            printf(" %15lf |", *(double *)var);
+            tlen += taosEncodeBinary(&pDataBuf, var, sizeof(double));
+          case TSDB_DATA_TYPE_UBIGINT:
+            printf(" %15lu |", *(uint64_t *)var);
+            tlen += taosEncodeFixedU64(&pDataBuf, *(uint64_t *)var);
+            break;
+          case TSDB_DATA_TYPE_NCHAR: {
+            char tmpChar[100] = {0};
+            strncpy(tmpChar, varDataVal(var), varDataLen(var));
+            printf(" %s |", tmpChar);
+            tlen += taosEncodeBinary(&pDataBuf, varDataVal(var), varDataLen(var));
+            break;
+          }
+          case TSDB_DATA_TYPE_VARCHAR: {  // TSDB_DATA_TYPE_BINARY
+            char tmpChar[100] = {0};
+            strncpy(tmpChar, varDataVal(var), varDataLen(var));
+            printf(" %s |", tmpChar);
+            tlen += taosEncodeBinary(&pDataBuf, varDataVal(var), varDataLen(var));
+            break;
+          }
+          case TSDB_DATA_TYPE_VARBINARY:
+            // TODO: add binary/varbinary
+            TASSERT(0);
+          default:
+            printf("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
+            TASSERT(0);
+            break;
+        }
+      }
+      printf("\n");
+      // if ((tlen > 0) && (skey != TSKEY_INITIAL_VAL)) {
+      if (tlen > 0) {
+        int32_t fid = (int32_t)(TSDB_KEY_FID(skey, minutePerFile, pCfg->precision));
+
+        // Step 2: Set the DFile for storage of SMA index, and iterate/split the TSma data and store to B+Tree index
+        // file
+        //         - Set and open the DFile or the B+Tree file
+        // TODO: tsdbStartTSmaCommit();
+        if (fid != tSmaH.dFile.fid) {
+          if (tSmaH.dFile.fid != SMA_IVLD_FID) {
+            tdSmaEndCommit(pEnv);
+            smaCloseDBF(&tSmaH.dFile);
+          }
+          tdSetTSmaDataFile(&tSmaH, indexUid, fid);
+          smaDebug("@@@ vgId:%d write to DBF %s, days:%d, interval:%" PRIi64 ", storageLevel:%" PRIi32
+                   " queryKey:%" PRIi64,
+                   SMA_VID(pSma), tSmaH.dFile.path, minutePerFile, tSmaH.interval, storageLevel, testSkey);
+          if (smaOpenDBF(pEnv->dbEnv, &tSmaH.dFile) != 0) {
+            smaWarn("vgId:%d open DB file %s failed since %s", SMA_VID(pSma),
+                    tSmaH.dFile.path ? tSmaH.dFile.path : "path is NULL", tstrerror(terrno));
+            tdDestroyTSmaWriteH(&tSmaH);
+            tdUnRefSmaStat(pSma, pStat);
+            return TSDB_CODE_FAILED;
+          }
+          tdSmaBeginCommit(pEnv);
+        }
+
+        if (tdInsertTSmaBlocks(&tSmaH, &smaKey, SMA_KEY_LEN, dataBuf, tlen, &pEnv->txn) != 0) {
+          smaWarn("vgId:%d insert tsma data blocks fail for index %" PRIi64 ", skey %" PRIi64 ", groupId %" PRIi64
+                  " since %s",
+                  SMA_VID(pSma), indexUid, skey, groupId, tstrerror(terrno));
+          tdSmaEndCommit(pEnv);
+          tdDestroyTSmaWriteH(&tSmaH);
+          tdUnRefSmaStat(pSma, pStat);
+          return TSDB_CODE_FAILED;
+        }
+
+        smaDebug("vgId:%d insert tsma data blocks success for index %" PRIi64 ", skey %" PRIi64 ", groupId %" PRIi64,
+                 SMA_VID(pSma), indexUid, skey, groupId);
+        // TODO:tsdbEndTSmaCommit();
+
+        // Step 3: reset the SSmaStat
+        tdResetExpiredWindow(pSma, pStat, indexUid, skey);
+      } else {
+        smaWarn("vgId:%d invalid data skey:%" PRIi64 ", tlen %" PRIi32 " during insert tSma data for %" PRIi64,
+                SMA_VID(pSma), skey, tlen, indexUid);
+      }
+    }
+  }
+  tdSmaEndCommit(pEnv);  // TODO: not commit for every insert
+  tdDestroyTSmaWriteH(&tSmaH);
+  tdUnRefSmaStat(pSma, pStat);
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t tdDropTSmaData(SSma *pSma, int64_t indexUid) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if ((code = tdDropTSmaDataImpl(pSma, indexUid)) < 0) {
+    smaWarn("vgId:%d drop tSma data failed since %s", SMA_VID(pSma), tstrerror(terrno));
+  }
+  return code;
+}
+
+/**
+ * @brief Insert TSma data blocks to DB File build by B+Tree
+ *
+ * @param pSmaH
+ * @param smaKey  tableUid-colId-skeyOfWindow(8-2-8)
+ * @param keyLen
+ * @param pData
+ * @param dataLen
+ * @return int32_t
+ */
+static int32_t tdInsertTSmaBlocks(STSmaWriteH *pSmaH, void *smaKey, int32_t keyLen, void *pData, int32_t dataLen,
+                                  TXN *txn) {
+  SDBFile *pDBFile = &pSmaH->dFile;
+
+  // TODO: insert tsma data blocks into B+Tree(TTB)
+  if (smaSaveSmaToDB(pDBFile, smaKey, keyLen, pData, dataLen, txn) != 0) {
+    smaWarn("vgId:%d insert tsma data blocks into %s: smaKey %" PRIx64 "-%" PRIx64 ", dataLen %" PRIu32 " fail",
+            SMA_VID(pSmaH->pSma), pDBFile->path, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), dataLen);
+    return TSDB_CODE_FAILED;
+  }
+  smaDebug("vgId:%d insert tsma data blocks into %s: smaKey %" PRIx64 "-%" PRIx64 ", dataLen %" PRIu32 " succeed",
+           SMA_VID(pSmaH->pSma), pDBFile->path, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), dataLen);
+
+#ifdef _TEST_SMA_PRINT_DEBUG_LOG_
+  uint32_t valueSize = 0;
+  void    *data = tdGetSmaDataByKey(pDBFile, smaKey, keyLen, &valueSize);
+  ASSERT(data != NULL);
+  for (uint32_t v = 0; v < valueSize; v += 8) {
+    smaWarn("vgId:%d insert sma data val[%d] %" PRIi64, REPO_ID(pSmaH->pTsdb), v, *(int64_t *)POINTER_SHIFT(data, v));
+  }
+#endif
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief When sma data received from stream computing, make the relative expired window valid.
+ *
+ * @param pSma
+ * @param pStat
+ * @param indexUid
+ * @param skey
+ * @return int32_t
+ */
+static int32_t tdResetExpiredWindow(SSma *pSma, SSmaStat *pStat, int64_t indexUid, TSKEY skey) {
+  SSmaStatItem *pItem = NULL;
+
+  tdRefSmaStat(pSma, pStat);
+
+  if (pStat && SMA_STAT_ITEMS(pStat)) {
+    pItem = taosHashGet(SMA_STAT_ITEMS(pStat), &indexUid, sizeof(indexUid));
+  }
+  if ((pItem) && ((pItem = *(SSmaStatItem **)pItem))) {
+    // pItem resides in hash buffer all the time unless drop sma index
+    // TODO: multithread protect
+    if (taosHashRemove(pItem->expiredWindows, &skey, sizeof(TSKEY)) != 0) {
+      // error handling
+      tdUnRefSmaStat(pSma, pStat);
+      smaWarn("vgId:%d remove skey %" PRIi64 " from expired window for sma index %" PRIi64 " fail", SMA_VID(pSma), skey,
+              indexUid);
+      return TSDB_CODE_FAILED;
+    }
+    smaDebug("vgId:%d remove skey %" PRIi64 " from expired window for sma index %" PRIi64 " succeed", SMA_VID(pSma),
+             skey, indexUid);
+    // TODO: use a standalone interface to received state upate notification from stream computing module.
+    /**
+     * @brief state
+     *  - When SMA env init in TSDB, its status is TSDB_SMA_STAT_OK.
+     *  - In startup phase of stream computing module, it should notify the SMA env in TSDB to expired if needed(e.g.
+     * when batch data caculation not finised)
+     *  - When TSDB_SMA_STAT_OK, the stream computing module should also notify that to the SMA env in TSDB.
+     */
+    pItem->state = TSDB_SMA_STAT_OK;
+  } else {
+    // error handling
+    tdUnRefSmaStat(pSma, pStat);
+    smaWarn("vgId:%d expired window %" PRIi64 " not exists for sma index %" PRIi64, SMA_VID(pSma), skey, indexUid);
+    return TSDB_CODE_FAILED;
+  }
+
+  tdUnRefSmaStat(pSma, pStat);
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief Drop tSma data and local cache
+ *        - insert/query reference
+ * @param pSma
+ * @param msg
+ * @return int32_t
+ */
+static int32_t tdDropTSmaDataImpl(SSma *pSma, int64_t indexUid) {
+  SSmaEnv *pEnv = atomic_load_ptr(&SMA_TSMA_ENV(pSma));
+
+  // clear local cache
+  if (pEnv) {
+    smaDebug("vgId:%d drop tSma local cache for %" PRIi64, SMA_VID(pSma), indexUid);
+
+    SSmaStatItem *pItem = taosHashGet(SMA_ENV_STAT_ITEMS(pEnv), &indexUid, sizeof(indexUid));
+    if ((pItem) || ((pItem = *(SSmaStatItem **)pItem))) {
+      if (tdSmaStatIsDropped(pItem)) {
+        smaDebug("vgId:%d tSma stat is already dropped for %" PRIi64, SMA_VID(pSma), indexUid);
+        return TSDB_CODE_TDB_INVALID_ACTION;  // TODO: duplicate drop msg would be intercepted by mnode
+      }
+
+      tdWLockSmaEnv(pEnv);
+      if (tdSmaStatIsDropped(pItem)) {
+        tdUnLockSmaEnv(pEnv);
+        smaDebug("vgId:%d tSma stat is already dropped for %" PRIi64, SMA_VID(pSma), indexUid);
+        return TSDB_CODE_TDB_INVALID_ACTION;  // TODO: duplicate drop msg would be intercepted by mnode
+      }
+      tdSmaStatSetDropped(pItem);
+      tdUnLockSmaEnv(pEnv);
+
+      int32_t nSleep = 0;
+      int32_t refVal = INT32_MAX;
+      while (true) {
+        if ((refVal = T_REF_VAL_GET(SMA_ENV_STAT(pEnv))) <= 0) {
+          smaDebug("vgId:%d drop index %" PRIi64 " since refVal=%d", SMA_VID(pSma), indexUid, refVal);
+          break;
+        }
+        smaDebug("vgId:%d wait 1s to drop index %" PRIi64 " since refVal=%d", SMA_VID(pSma), indexUid, refVal);
+        taosSsleep(1);
+        if (++nSleep > SMA_DROP_EXPIRED_TIME) {
+          smaDebug("vgId:%d drop index %" PRIi64 " after wait %d (refVal=%d)", SMA_VID(pSma), indexUid, nSleep, refVal);
+          break;
+        };
+      }
+
+      tdFreeSmaStatItem(pItem);
+      smaDebug("vgId:%d getTSmaDataImpl failed since no index %" PRIi64 " in local cache", SMA_VID(pSma), indexUid);
+    }
+  }
+  // clear sma data files
+  // TODO:
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief
+ *
+ * @param pSma Return the data between queryWin and fill the pData.
+ * @param pData
+ * @param indexUid
+ * @param pQuerySKey
+ * @param nMaxResult The query invoker should control the nMaxResult need to return to avoid OOM.
+ * @return int32_t
+ */
+int32_t tdGetTSmaDataImpl(SSma *pSma, char *pData, int64_t indexUid, TSKEY querySKey, int32_t nMaxResult) {
+  SSmaEnv  *pEnv = atomic_load_ptr(&SMA_TSMA_ENV(pSma));
+  SSmaStat *pStat = NULL;
+
+  if (!pEnv) {
+    terrno = TSDB_CODE_INVALID_PTR;
+    smaWarn("vgId:%d getTSmaDataImpl failed since pTSmaEnv is NULL", SMA_VID(pSma));
+    return TSDB_CODE_FAILED;
+  }
+
+  pStat = SMA_ENV_STAT(pEnv);
+
+  tdRefSmaStat(pSma, pStat);
+  SSmaStatItem *pItem = taosHashGet(SMA_ENV_STAT_ITEMS(pEnv), &indexUid, sizeof(indexUid));
+  if (!pItem || !(pItem = *(SSmaStatItem **)pItem)) {
+    // Normally pItem should not be NULL, mark all windows as expired and notify query module to fetch raw TS data if
+    // it's NULL.
+    tdUnRefSmaStat(pSma, pStat);
+    terrno = TSDB_CODE_TDB_INVALID_ACTION;
+    smaDebug("vgId:%d getTSmaDataImpl failed since no index %" PRIi64, SMA_VID(pSma), indexUid);
+    return TSDB_CODE_FAILED;
+  }
+
+#if 0
+  int32_t nQueryWin = taosArrayGetSize(pQuerySKey);
+  for (int32_t n = 0; n < nQueryWin; ++n) {
+    TSKEY skey = taosArrayGet(pQuerySKey, n);
+    if (taosHashGet(pItem->expiredWindows, &skey, sizeof(TSKEY))) {
+      // TODO: mark this window as expired.
+    }
+  }
+#endif
+
+#if 1
+  int8_t smaStat = 0;
+  if (!tdSmaStatIsOK(pItem, &smaStat)) {  // TODO: multiple check for large scale sma query
+    tdUnRefSmaStat(pSma, pStat);
+    terrno = TSDB_CODE_TDB_INVALID_SMA_STAT;
+    smaWarn("vgId:%d getTSmaDataImpl failed from index %" PRIi64 " since %s %" PRIi8, SMA_VID(pSma), indexUid,
+            tstrerror(terrno), smaStat);
+    return TSDB_CODE_FAILED;
+  }
+
+  if (taosHashGet(pItem->expiredWindows, &querySKey, sizeof(TSKEY))) {
+    // TODO: mark this window as expired.
+    smaDebug("vgId:%d skey %" PRIi64 " of window exists in expired window for index %" PRIi64, SMA_VID(pSma), querySKey,
+             indexUid);
+  } else {
+    smaDebug("vgId:%d skey %" PRIi64 " of window not in expired window for index %" PRIi64, SMA_VID(pSma), querySKey,
+             indexUid);
+  }
+
+  STSma *pTSma = pItem->pTSma;
+#endif
+
+#if 1
+  STSmaReadH tReadH = {0};
+  tdInitTSmaReadH(&tReadH, pSma, pTSma->interval, pTSma->intervalUnit);
+  smaCloseDBF(&tReadH.dFile);
+
+  tdUnRefSmaStat(pSma, pStat);
+
+  tdInitTSmaFile(&tReadH, indexUid, querySKey);
+  smaDebug("### vgId:%d read from DBF %s  days:%d, interval:%" PRIi64 ", storageLevel:%" PRIi8 " queryKey:%" PRIi64,
+           SMA_VID(pSma), tReadH.dFile.path, tReadH.days, tReadH.interval, tReadH.storageLevel, querySKey);
+  if (smaOpenDBF(pEnv->dbEnv, &tReadH.dFile) != 0) {
+    smaWarn("vgId:%d open DBF %s failed since %s", SMA_VID(pSma), tReadH.dFile.path, tstrerror(terrno));
+    return TSDB_CODE_FAILED;
+  }
+
+  char    smaKey[SMA_KEY_LEN] = {0};
+  void   *pSmaKey = &smaKey;
+  int64_t queryGroupId = 0;
+  tdEncodeTSmaKey(queryGroupId, querySKey, (void **)&pSmaKey);
+
+  smaDebug("vgId:%d get sma data from %s: smaKey %" PRIx64 "-%" PRIx64 ", keyLen %d", SMA_VID(pSma), tReadH.dFile.path,
+           *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), SMA_KEY_LEN);
+
+  void   *result = NULL;
+  int32_t valueSize = 0;
+  if (!(result = smaGetSmaDataByKey(&tReadH.dFile, smaKey, SMA_KEY_LEN, &valueSize))) {
+    smaWarn("vgId:%d get sma data failed from smaIndex %" PRIi64 ", smaKey %" PRIx64 "-%" PRIx64 " since %s",
+            SMA_VID(pSma), indexUid, *(int64_t *)smaKey, *(int64_t *)POINTER_SHIFT(smaKey, 8), tstrerror(terrno));
+    smaCloseDBF(&tReadH.dFile);
+    return TSDB_CODE_FAILED;
+  }
+#endif
+
+#ifdef _TEST_SMA_PRINT_DEBUG_LOG_
+  for (uint32_t v = 0; v < valueSize; v += 8) {
+    smaWarn("vgId:%d get sma data v[%d]=%" PRIi64, SMA_VID(pSma), v, *(int64_t *)POINTER_SHIFT(result, v));
+  }
+#endif
+  taosMemoryFreeClear(result);  // TODO: fill the result to output
+
+#if 0
+  int32_t nResult = 0;
+  int64_t lastKey = 0;
+
+  while (true) {
+    if (nResult >= nMaxResult) {
+      break;
+    }
+
+    // set and open the file according to the STSma param
+    if (tdSetAndOpenTSmaFile(&tReadH, queryWin)) {
+      char bTree[100] = "\0";
+      while (strncmp(bTree, "has more nodes", 100) == 0) {
+        if (nResult >= nMaxResult) {
+          break;
+        }
+        // tdGetDataFromBTree(bTree, queryWin, lastKey)
+        // fill the pData
+        ++nResult;
+      }
+    }
+  }
+#endif
+  // read data from file and fill the result
+  smaCloseDBF(&tReadH.dFile);
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
+  SSmaCfg *pCfg = (SSmaCfg *)pMsg;
+
+  if (metaCreateTSma(SMA_META(pSma), version, pCfg) < 0) {
+    return -1;
+  }
+
+  tdTSmaAdd(pSma, 1);
+  return 0;
+}
+
+int32_t tdDropTSma(SSma *pSma, char *pMsg) {
+#if 0
+  SVDropTSmaReq vDropSmaReq = {0};
+  if (!tDeserializeSVDropTSmaReq(pMsg, &vDropSmaReq)) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  // TODO: send msg to stream computing to drop tSma
+  // if ((send msg to stream computing) < 0) {
+  //   tDestroyTSma(&vCreateSmaReq);
+  //   return -1;
+  // }
+  //
+
+  if (metaDropTSma(SMA_META(pSma), vDropSmaReq.indexUid) < 0) {
+    // TODO: handle error
+    return -1;
+  }
+
+  if (tdDropTSmaData(pSma, vDropSmaReq.indexUid) < 0) {
+    // TODO: handle error
+    return -1;
+  }
+
+  tdTSmaSub(pSma, 1);
+#endif
+
+  // TODO: return directly or go on follow steps?
+  return TSDB_CODE_SUCCESS;
+}
+
+static SSmaStatItem *tdNewSmaStatItem(int8_t state) {
+  SSmaStatItem *pItem = NULL;
+
+  pItem = (SSmaStatItem *)taosMemoryCalloc(1, sizeof(SSmaStatItem));
+  if (!pItem) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return NULL;
+  }
+
+  pItem->state = state;
+  pItem->expiredWindows = taosHashInit(SMA_STATE_ITEM_HASH_SLOT, taosGetDefaultHashFunction(TSDB_DATA_TYPE_TIMESTAMP),
+                                       true, HASH_ENTRY_LOCK);
+  if (!pItem->expiredWindows) {
+    taosMemoryFreeClear(pItem);
+    return NULL;
+  }
+
+  return pItem;
+}
+
+static int32_t tdSetExpiredWindow(SSma *pSma, SHashObj *pItemsHash, int64_t indexUid, int64_t winSKey,
+                                  int64_t version) {
+  SSmaStatItem *pItem = taosHashGet(pItemsHash, &indexUid, sizeof(indexUid));
+  if (!pItem) {
+    // TODO: use TSDB_SMA_STAT_EXPIRED and update by stream computing later
+    pItem = tdNewSmaStatItem(TSDB_SMA_STAT_OK);  // TODO use the real state
+    if (!pItem) {
+      // Response to stream computing: OOM
+      // For query, if the indexUid not found, the TSDB should tell query module to query raw TS data.
+      return TSDB_CODE_FAILED;
+    }
+
+    // cache smaMeta
+    STSma *pTSma = metaGetSmaInfoByIndex(SMA_META(pSma), indexUid);
+    if (!pTSma) {
+      terrno = TSDB_CODE_TDB_NO_SMA_INDEX_IN_META;
+      taosHashCleanup(pItem->expiredWindows);
+      taosMemoryFree(pItem);
+      smaWarn("vgId:%d set expire window, get tsma meta failed for smaIndex %" PRIi64 " since %s", SMA_VID(pSma),
+              indexUid, tstrerror(terrno));
+      return TSDB_CODE_FAILED;
+    }
+    pItem->pTSma = pTSma;
+
+    if (taosHashPut(pItemsHash, &indexUid, sizeof(indexUid), &pItem, sizeof(pItem)) != 0) {
+      // If error occurs during put smaStatItem, free the resources of pItem
+      taosHashCleanup(pItem->expiredWindows);
+      taosMemoryFree(pItem);
+      return TSDB_CODE_FAILED;
+    }
+  } else if (!(pItem = *(SSmaStatItem **)pItem)) {
+    terrno = TSDB_CODE_INVALID_PTR;
+    return TSDB_CODE_FAILED;
+  }
+
+  if (taosHashPut(pItem->expiredWindows, &winSKey, sizeof(TSKEY), &version, sizeof(version)) != 0) {
+    // If error occurs during taosHashPut expired windows, remove the smaIndex from pSma->pSmaStat, thus TSDB would
+    // tell query module to query raw TS data.
+    // N.B.
+    //  1) It is assumed to be extemely little probability event of fail to taosHashPut.
+    //  2) This would solve the inconsistency to some extent, but not completely, unless we record all expired
+    // windows failed to put into hash table.
+    taosHashCleanup(pItem->expiredWindows);
+    taosMemoryFreeClear(pItem->pTSma);
+    taosHashRemove(pItemsHash, &indexUid, sizeof(indexUid));
+    smaWarn("vgId:%d smaIndex %" PRIi64 ", put skey %" PRIi64 " to expire window fail", SMA_VID(pSma), indexUid,
+            winSKey);
+    return TSDB_CODE_FAILED;
+  }
+
+  smaDebug("vgId:%d smaIndex %" PRIi64 ", put skey %" PRIi64 " to expire window succeed", SMA_VID(pSma), indexUid,
+           winSKey);
+  return TSDB_CODE_SUCCESS;
+}
+
+/**
+ * @brief Update expired window according to msg from stream computing module.
+ *
+ * @param pSma
+ * @param msg SSubmitReq
+ * @return int32_t
+ */
+int32_t tdUpdateExpiredWindowImpl(SSma *pSma, SSubmitReq *pMsg, int64_t version) {
+  // no time-range-sma, just return success
+  if (atomic_load_16(&SMA_TSMA_NUM(pSma)) <= 0) {
+    smaTrace("vgId:%d not update expire window since no tSma", SMA_VID(pSma));
+    return TSDB_CODE_SUCCESS;
+  }
+
+  if (!SMA_META(pSma)) {
+    terrno = TSDB_CODE_INVALID_PTR;
+    smaError("vgId:%d update expire window failed since no meta ptr", SMA_VID(pSma));
+    return TSDB_CODE_FAILED;
+  }
+
+  if (tdCheckAndInitSmaEnv(pSma, TSDB_SMA_TYPE_TIME_RANGE) < 0) {
+    smaError("vgId:%d init sma env failed since %s", SMA_VID(pSma), terrstr(terrno));
+    terrno = TSDB_CODE_TDB_INIT_FAILED;
+    return TSDB_CODE_FAILED;
+  }
+
+  // Firstly, assume that tSma can only be created on super table/normal table.
+  // getActiveTimeWindow
+
+  SSmaEnv  *pEnv = SMA_TSMA_ENV(pSma);
+  SSmaStat *pStat = SMA_ENV_STAT(pEnv);
+  SHashObj *pItemsHash = SMA_ENV_STAT_ITEMS(pEnv);
+
+  TASSERT(pEnv && pStat && pItemsHash);
+
+  // basic procedure
+  // TODO: optimization
+  tdRefSmaStat(pSma, pStat);
+
+  SSubmitMsgIter msgIter = {0};
+  SSubmitBlk    *pBlock = NULL;
+  SInterval      interval = {0};
+  TSKEY          lastWinSKey = INT64_MIN;
+
+  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) {
+    return TSDB_CODE_FAILED;
+  }
+
+  while (true) {
+    tGetSubmitMsgNext(&msgIter, &pBlock);
+    if (!pBlock) break;
+
+    STSmaWrapper *pSW = NULL;
+    STSma        *pTSma = NULL;
+
+    SSubmitBlkIter blkIter = {0};
+    if (tInitSubmitBlkIter(&msgIter, pBlock, &blkIter) < 0) {
+      pSW = tFreeTSmaWrapper(pSW, false);
+      break;
+    }
+
+    while (true) {
+      STSRow *row = tGetSubmitBlkNext(&blkIter);
+      if (!row) {
+        pSW = tFreeTSmaWrapper(pSW, false);
+        break;
+      }
+      if (!pSW || (pTSma && (pTSma->tableUid != msgIter.suid))) {
+        if (pSW) {
+          pSW = tFreeTSmaWrapper(pSW, false);
+        }
+        if (!(pSW = metaGetSmaInfoByTable(SMA_META(pSma), msgIter.suid, false))) {
+          break;
+        }
+        if ((pSW->number) <= 0 || !pSW->tSma) {
+          pSW = tFreeTSmaWrapper(pSW, false);
+          break;
+        }
+
+        pTSma = pSW->tSma;
+
+        interval.interval = pTSma->interval;
+        interval.intervalUnit = pTSma->intervalUnit;
+        interval.offset = pTSma->offset;
+        interval.precision = SMA_TSDB_CFG(pSma)->precision;
+        interval.sliding = pTSma->sliding;
+        interval.slidingUnit = pTSma->slidingUnit;
+      }
+
+      // TODO: process multiple tsma for one table uid
+      TSKEY winSKey = taosTimeTruncate(TD_ROW_KEY(row), &interval, interval.precision);
+
+      if (lastWinSKey != winSKey) {
+        lastWinSKey = winSKey;
+        if (tdSetExpiredWindow(pSma, pItemsHash, pTSma->indexUid, winSKey, version) < 0) {
+          pSW = tFreeTSmaWrapper(pSW, false);
+          tdUnRefSmaStat(pSma, pStat);
+          return TSDB_CODE_FAILED;
+        }
+      } else {
+        smaDebug("vgId:%d smaIndex %" PRIi64 ", put skey %" PRIi64 " to expire window ignore as duplicated",
+                 SMA_VID(pSma), pTSma->indexUid, winSKey);
+      }
+    }
+  }
+
+  tdUnRefSmaStat(pSma, pStat);
+
+  return TSDB_CODE_SUCCESS;
+}
\ No newline at end of file

source/dnode/vnode/src/tq/tq.c

@@ -229,7 +229,7 @@ int tqPushMsg(STQ* pTq, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver)
     }
     memcpy(data, msg, msgLen);
 
-    tqProcessStreamTrigger(pTq, data);
+    // tqProcessStreamTrigger(pTq, data);
   }
 
   return 0;

source/dnode/vnode/src/tsdb/tsdbSma.c

@@ -515,7 +515,7 @@ static SSmaStatItem *tsdbNewSmaStatItem(int8_t state) {
 
 static void *tsdbFreeSmaStatItem(SSmaStatItem *pSmaStatItem) {
   if (pSmaStatItem) {
-    tdDestroyTSma(pSmaStatItem->pSma);
+    tDestroyTSma(pSmaStatItem->pSma);
     taosMemoryFreeClear(pSmaStatItem->pSma);
     taosHashCleanup(pSmaStatItem->expiredWindows);
     taosMemoryFreeClear(pSmaStatItem);
@@ -718,25 +718,25 @@ int32_t tsdbUpdateExpiredWindowImpl(STsdb *pTsdb, SSubmitReq *pMsg, int64_t vers
 
     SSubmitBlkIter blkIter = {0};
     if (tInitSubmitBlkIter(&msgIter, pBlock, &blkIter) != TSDB_CODE_SUCCESS) {
-      pSW = tdFreeTSmaWrapper(pSW);
+      pSW = tFreeTSmaWrapper(pSW);
       break;
     }
 
     while (true) {
       STSRow *row = tGetSubmitBlkNext(&blkIter);
       if (!row) {
-        tdFreeTSmaWrapper(pSW);
+        tFreeTSmaWrapper(pSW);
         break;
       }
       if (!pSW || (pTSma->tableUid != pBlock->suid)) {
         if (pSW) {
-          pSW = tdFreeTSmaWrapper(pSW);
+          pSW = tFreeTSmaWrapper(pSW);
         }
         if (!(pSW = metaGetSmaInfoByTable(REPO_META(pTsdb), pBlock->suid))) {
           break;
         }
         if ((pSW->number) <= 0 || !pSW->tSma) {
-          pSW = tdFreeTSmaWrapper(pSW);
+          pSW = tFreeTSmaWrapper(pSW);
           break;
         }
 
@@ -1649,13 +1649,13 @@ int32_t tsdbCreateTSma(STsdb *pTsdb, char *pMsg) {
     // TODO: handle error
     tsdbWarn("vgId:%d tsma %s:%" PRIi64 " create failed for table %" PRIi64 " since %s", REPO_ID(pTsdb),
              vCreateSmaReq.tSma.indexName, vCreateSmaReq.tSma.indexUid, vCreateSmaReq.tSma.tableUid, terrstr(terrno));
-    tdDestroyTSma(&vCreateSmaReq.tSma);
+    tDestroyTSma(&vCreateSmaReq.tSma);
     return -1;
   }
 
   tsdbTSmaAdd(pTsdb, 1);
 
-  tdDestroyTSma(&vCreateSmaReq.tSma);
+  tDestroyTSma(&vCreateSmaReq.tSma);
   // TODO: return directly or go on follow steps?
   return TSDB_CODE_SUCCESS;
 }
@@ -1669,7 +1669,7 @@ int32_t tsdbDropTSma(STsdb *pTsdb, char *pMsg) {
 
   // TODO: send msg to stream computing to drop tSma
   // if ((send msg to stream computing) < 0) {
-  //   tdDestroyTSma(&vCreateSmaReq);
+  //   tDestroyTSma(&vCreateSmaReq);
   //   return -1;
   // }
   //

source/dnode/vnode/test/tsdbSmaTest.cpp

@@ -147,8 +147,8 @@ TEST(testCase, tSma_Meta_Encode_Decode_Test) {
 
   // resource release
   taosMemoryFreeClear(pSW);
-  tdDestroyTSma(&tSma);
-  tdDestroyTSmaWrapper(&dstTSmaWrapper);
+  tDestroyTSma(&tSma);
+  tDestroyTSmaWrapper(&dstTSmaWrapper);
 }
 #endif
 
@@ -218,7 +218,7 @@ TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
   printf("tagsFilter1 = %s\n", qSmaCfg->tagsFilter != NULL ? qSmaCfg->tagsFilter : "");
   EXPECT_STRCASEEQ(qSmaCfg->indexName, smaIndexName1);
   EXPECT_EQ(qSmaCfg->tableUid, tSma.tableUid);
-  tdDestroyTSma(qSmaCfg);
+  tDestroyTSma(qSmaCfg);
   taosMemoryFreeClear(qSmaCfg);
 
   qSmaCfg = metaGetSmaInfoByIndex(pMeta, indexUid2, true);
@@ -229,7 +229,7 @@ TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
   printf("tagsFilter2 = %s\n", qSmaCfg->tagsFilter != NULL ? qSmaCfg->tagsFilter : "");
   EXPECT_STRCASEEQ(qSmaCfg->indexName, smaIndexName2);
   EXPECT_EQ(qSmaCfg->interval, tSma.interval);
-  tdDestroyTSma(qSmaCfg);
+  tDestroyTSma(qSmaCfg);
   taosMemoryFreeClear(qSmaCfg);
 
   // get index name by table uid
@@ -265,7 +265,7 @@ TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
   EXPECT_EQ((pSW->tSma + 1)->indexUid, indexUid2);
   EXPECT_EQ((pSW->tSma + 1)->tableUid, tbUid);
 
-  tdDestroyTSmaWrapper(pSW);
+  tDestroyTSmaWrapper(pSW);
   taosMemoryFreeClear(pSW);
 
   // get all sma table uids
@@ -282,7 +282,7 @@ TEST(testCase, tSma_metaDB_Put_Get_Del_Test) {
   metaRemoveSmaFromDb(pMeta, indexUid1);
   metaRemoveSmaFromDb(pMeta, indexUid2);
 
-  tdDestroyTSma(&tSma);
+  tDestroyTSma(&tSma);
   metaClose(pMeta);
 }
 #endif
@@ -576,7 +576,7 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
   taosArrayDestroy(pDataBlocks);
 
   // release meta
-  tdDestroyTSma(&tSma);
+  tDestroyTSma(&tSma);
   tfsClose(pTsdb->pTfs);
   tsdbClose(pTsdb);
   metaClose(pMeta);