Merge remote-tracking branch 'origin/3.0' into enh/3.0_planner_optimize

docs/zh/12-taos-sql/20-keywords.md

@@ -6,7 +6,8 @@ description: TDengine 保留关键字的详细列表
 
 ## 保留关键字
 
-目前 TDengine 有将近 200 个内部保留关键字，这些关键字无论大小写均不可以用作库名、表名、STable 名、数据列名及标签列名等。这些关键字列表如下：
+目前 TDengine 有将近 200 个内部保留关键字，这些关键字无论大小写如果需要用作库名、表名、STable 名、数据列名及标签列名等，需要使用符合``将关键字括起来使用，例如`ADD`。
+关键字列表如下：
 
 ### A
 
@@ -239,6 +240,7 @@ description: TDengine 保留关键字的详细列表
 - TOPICS
 - TRIGGER
 - TSERIES
+- TTL
 
 ### U
 

include/common/taosdef.h

@@ -65,13 +65,6 @@ typedef enum {
   TSDB_STATIS_NONE = 1,  // statis part not exist
 } ETsdbStatisStatus;
 
-typedef enum {
-  TSDB_SMA_STAT_UNKNOWN = -1,  // unknown
-  TSDB_SMA_STAT_OK = 0,        // ready to provide service
-  TSDB_SMA_STAT_EXPIRED = 1,   // not ready or expired
-  TSDB_SMA_STAT_DROPPED = 2,   // sma dropped
-} ETsdbSmaStat;                // bit operation
-
 typedef enum {
   TSDB_SMA_TYPE_BLOCK = 0,       // Block-wise SMA
   TSDB_SMA_TYPE_TIME_RANGE = 1,  // Time-range-wise SMA

include/common/tcommon.h

@@ -184,7 +184,6 @@ typedef struct SQueryTableDataCond {
   STimeWindow  twindows;
   int64_t      startVersion;
   int64_t      endVersion;
-  int64_t      schemaVersion;
 } SQueryTableDataCond;
 
 int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock);

include/common/tmsg.h

@@ -845,6 +845,8 @@ typedef struct {
   int64_t uid;
   int32_t vgVersion;
   int32_t vgNum;
+  int16_t hashPrefix;
+  int16_t hashSuffix;
   int8_t  hashMethod;
   SArray* pVgroupInfos;  // Array of SVgroupInfo
 } SUseDbRsp;
@@ -1070,6 +1072,7 @@ typedef struct {
 typedef struct {
   int32_t vgId;
   int32_t syncState;
+  int64_t cacheUsage;
   int64_t numOfTables;
   int64_t numOfTimeSeries;
   int64_t totalStorage;

include/libs/qcom/query.h

@@ -116,6 +116,8 @@ typedef struct STableMeta {
 
 typedef struct SDBVgInfo {
   int32_t   vgVersion;
+  int16_t   hashPrefix;
+  int16_t   hashSuffix;
   int8_t    hashMethod;
   int32_t   numOfTable;  // DB's table num, unit is TSDB_TABLE_NUM_UNIT
   SHashObj* vgHash;      // key:vgId, value:SVgroupInfo

include/util/taoserror.h

@@ -616,6 +616,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP        TAOS_DEF_ERROR_CODE(0, 0x3155)
 #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)
 
 //index
 #define TSDB_CODE_INDEX_REBUILDING               TAOS_DEF_ERROR_CODE(0, 0x3200)

include/util/trbtree.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_UTIL_RBTREE_H_
+#define _TD_UTIL_RBTREE_H_
+
+#include "os.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct SRBTree     SRBTree;
+typedef struct SRBTreeNode SRBTreeNode;
+typedef struct SRBTreeIter SRBTreeIter;
+
+typedef int32_t (*tRBTreeCmprFn)(const void *, const void *);
+
+// SRBTree =============================================
+#define tRBTreeMin(T) ((T)->min == ((T)->NIL) ? NULL : (T)->min)
+#define tRBTreeMax(T) ((T)->max == ((T)->NIL) ? NULL : (T)->max)
+
+void         tRBTreeCreate(SRBTree *pTree, tRBTreeCmprFn cmprFn);
+SRBTreeNode *tRBTreePut(SRBTree *pTree, SRBTreeNode *z);
+void         tRBTreeDrop(SRBTree *pTree, SRBTreeNode *z);
+SRBTreeNode *tRBTreeDropByKey(SRBTree *pTree, void *pKey);
+SRBTreeNode *tRBTreeGet(SRBTree *pTree, void *pKey);
+
+// SRBTreeIter =============================================
+#define tRBTreeIterCreate(tree, ascend) \
+  (SRBTreeIter) { .asc = (ascend), .pTree = (tree), .pNode = (ascend) ? (tree)->min : (tree)->max }
+
+SRBTreeNode *tRBTreeIterNext(SRBTreeIter *pIter);
+
+// STRUCT =============================================
+typedef enum { RED, BLACK } ECOLOR;
+struct SRBTreeNode {
+  ECOLOR       color;
+  SRBTreeNode *parent;
+  SRBTreeNode *left;
+  SRBTreeNode *right;
+};
+
+#define RBTREE_NODE_PAYLOAD(N) ((const void *)&(N)[1])
+
+struct SRBTree {
+  tRBTreeCmprFn cmprFn;
+  int64_t       n;
+  SRBTreeNode  *root;
+  SRBTreeNode  *min;
+  SRBTreeNode  *max;
+  SRBTreeNode  *NIL;
+  SRBTreeNode   NILNODE;
+};
+
+struct SRBTreeIter {
+  int8_t       asc;
+  SRBTree     *pTree;
+  SRBTreeNode *pNode;
+};
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_UTIL_RBTREE_H_*/
\ No newline at end of file

source/client/src/clientEnv.c

@@ -488,7 +488,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) {
  */
 uint64_t generateRequestId() {
   static uint64_t hashId = 0;
-  static int32_t  requestSerialId = 0;
+  static uint32_t requestSerialId = 0;
 
   if (hashId == 0) {
     char    uid[64] = {0};
@@ -507,7 +507,8 @@ uint64_t generateRequestId() {
   while (true) {
     int64_t  ts = taosGetTimestampMs();
     uint64_t pid = taosGetPId();
-    int32_t  val = atomic_add_fetch_32(&requestSerialId, 1);
+    uint32_t val = atomic_add_fetch_32(&requestSerialId, 1);
+    if (val >= 0xFFFF) atomic_store_32(&requestSerialId, 0);
 
     id = ((hashId & 0x0FFF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
     if (id) {

source/client/src/clientHb.c

@@ -73,6 +73,8 @@ static int32_t hbProcessDBInfoRsp(void *value, int32_t valueLen, struct SCatalog
 
       vgInfo->vgVersion = rsp->vgVersion;
       vgInfo->hashMethod = rsp->hashMethod;
+      vgInfo->hashPrefix = rsp->hashPrefix;
+      vgInfo->hashSuffix = rsp->hashSuffix;
       vgInfo->vgHash = taosHashInit(rsp->vgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
       if (NULL == vgInfo->vgHash) {
         taosMemoryFree(vgInfo);

source/client/src/clientImpl.c

@@ -1672,7 +1672,12 @@ static int32_t doConvertJson(SReqResultInfo* pResultInfo, int32_t numOfCols, int
       break;
     }
   }
-  if (!needConvert) return TSDB_CODE_SUCCESS;
+
+  if (!needConvert) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  tscDebug("start to convert form json format string");
 
   char*   p = (char*)pResultInfo->pData;
   int32_t dataLen = estimateJsonLen(pResultInfo, numOfCols, numOfRows);

source/client/src/clientRawBlockWrite.c

@@ -765,6 +765,7 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) {
     }
     taosArrayPush(pRequest->tableList, &pName);
 
+    pCreateReq->flags |= TD_CREATE_IF_NOT_EXISTS;
     // change tag cid to new cid
     if (pCreateReq->type == TSDB_CHILD_TABLE) {
       STableMeta* pTableMeta = NULL;

source/common/src/systable.c

@@ -207,6 +207,7 @@ static const SSysDbTableSchema vgroupsSchema[] = {
     {.name = "v3_dnode", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "v3_status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
     {.name = "status", .bytes = 12 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "cacheload", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "nfiles", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "file_size", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "tsma", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT, .sysInfo = true},

source/common/src/tdatablock.c

@@ -2120,6 +2120,7 @@ void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_
   int32_t* rows = (int32_t*)data;
   *rows = pBlock->info.rows;
   data += sizeof(int32_t);
+  ASSERT(*rows > 0);
 
   int32_t* cols = (int32_t*)data;
   *cols = numOfCols;
@@ -2183,6 +2184,8 @@ void blockEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_
 
   *actualLen = *dataLen;
   *groupId = pBlock->info.groupId;
+  ASSERT(*dataLen > 0);
+  uDebug("build data block, actualLen:%d, rows:%d, cols:%d", *dataLen, *rows, *cols);
 }
 
 const char* blockDecode(SSDataBlock* pBlock, const char* pData) {

source/common/src/tglobal.c

@@ -129,10 +129,6 @@ int32_t tsMinIntervalTime = 1;
 int32_t tsQueryBufferSize = -1;
 int64_t tsQueryBufferSizeBytes = -1;
 
-// tsdb config
-// For backward compatibility
-bool tsdbForceKeepFile = false;
-
 int32_t  tsDiskCfgNum = 0;
 SDiskCfg tsDiskCfg[TFS_MAX_DISKS] = {0};
 

source/common/src/tmsg.c

@@ -994,6 +994,7 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
     SVnodeLoad *pload = taosArrayGet(pReq->pVloads, i);
     if (tEncodeI32(&encoder, pload->vgId) < 0) return -1;
     if (tEncodeI32(&encoder, pload->syncState) < 0) return -1;
+    if (tEncodeI64(&encoder, pload->cacheUsage) < 0) return -1;
     if (tEncodeI64(&encoder, pload->numOfTables) < 0) return -1;
     if (tEncodeI64(&encoder, pload->numOfTimeSeries) < 0) return -1;
     if (tEncodeI64(&encoder, pload->totalStorage) < 0) return -1;
@@ -1063,6 +1064,7 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
     SVnodeLoad vload = {0};
     if (tDecodeI32(&decoder, &vload.vgId) < 0) return -1;
     if (tDecodeI32(&decoder, &vload.syncState) < 0) return -1;
+    if (tDecodeI64(&decoder, &vload.cacheUsage) < 0) return -1;
     if (tDecodeI64(&decoder, &vload.numOfTables) < 0) return -1;
     if (tDecodeI64(&decoder, &vload.numOfTimeSeries) < 0) return -1;
     if (tDecodeI64(&decoder, &vload.totalStorage) < 0) return -1;
@@ -2461,6 +2463,8 @@ int32_t tSerializeSUseDbRspImp(SEncoder *pEncoder, const SUseDbRsp *pRsp) {
   if (tEncodeI64(pEncoder, pRsp->uid) < 0) return -1;
   if (tEncodeI32(pEncoder, pRsp->vgVersion) < 0) return -1;
   if (tEncodeI32(pEncoder, pRsp->vgNum) < 0) return -1;
+  if (tEncodeI16(pEncoder, pRsp->hashPrefix) < 0) return -1;
+  if (tEncodeI16(pEncoder, pRsp->hashSuffix) < 0) return -1;
   if (tEncodeI8(pEncoder, pRsp->hashMethod) < 0) return -1;
 
   for (int32_t i = 0; i < pRsp->vgNum; ++i) {
@@ -2512,6 +2516,8 @@ int32_t tDeserializeSUseDbRspImp(SDecoder *pDecoder, SUseDbRsp *pRsp) {
   if (tDecodeI64(pDecoder, &pRsp->uid) < 0) return -1;
   if (tDecodeI32(pDecoder, &pRsp->vgVersion) < 0) return -1;
   if (tDecodeI32(pDecoder, &pRsp->vgNum) < 0) return -1;
+  if (tDecodeI16(pDecoder, &pRsp->hashPrefix) < 0) return -1;
+  if (tDecodeI16(pDecoder, &pRsp->hashSuffix) < 0) return -1;
   if (tDecodeI8(pDecoder, &pRsp->hashMethod) < 0) return -1;
 
   if (pRsp->vgNum <= 0) {

source/common/src/trow.c

@@ -538,12 +538,12 @@ bool tdSTSRowIterGetTpVal(STSRowIter *pIter, col_type_t colType, int32_t offset,
     } else {
       pVal->val = POINTER_SHIFT(TD_ROW_DATA(pRow), offset);
     }
-    return TSDB_CODE_SUCCESS;
+    return true;
   }
 
   if (tdGetBitmapValType(pIter->pBitmap, pIter->colIdx - 1, &pVal->valType, 0) != TSDB_CODE_SUCCESS) {
     pVal->valType = TD_VTYPE_NONE;
-    return terrno;
+    return true;
   }
 
   if (pVal->valType == TD_VTYPE_NORM) {

source/dnode/mnode/impl/inc/mndDef.h

@@ -343,6 +343,7 @@ typedef struct {
   uint32_t  hashEnd;
   char      dbName[TSDB_DB_FNAME_LEN];
   int64_t   dbUid;
+  int64_t   cacheUsage;
   int64_t   numOfTables;
   int64_t   numOfTimeSeries;
   int64_t   totalStorage;

source/dnode/mnode/impl/src/mndDb.c

@@ -1171,6 +1171,8 @@ int32_t mndExtractDbInfo(SMnode *pMnode, SDbObj *pDb, SUseDbRsp *pRsp, const SUs
   pRsp->vgVersion = pDb->vgVersion;
   pRsp->vgNum = taosArrayGetSize(pRsp->pVgroupInfos);
   pRsp->hashMethod = pDb->cfg.hashMethod;
+  pRsp->hashPrefix = pDb->cfg.hashPrefix;
+  pRsp->hashSuffix = pDb->cfg.hashSuffix;
   return 0;
 }
 
@@ -1303,6 +1305,8 @@ int32_t mndValidateDbInfo(SMnode *pMnode, SDbVgVersion *pDbs, int32_t numOfDbs,
     usedbRsp.vgVersion = pDb->vgVersion;
     usedbRsp.vgNum = (int32_t)taosArrayGetSize(usedbRsp.pVgroupInfos);
     usedbRsp.hashMethod = pDb->cfg.hashMethod;
+    usedbRsp.hashPrefix = pDb->cfg.hashPrefix;
+    usedbRsp.hashSuffix = pDb->cfg.hashSuffix;
 
     taosArrayPush(batchUseRsp.pArray, &usedbRsp);
     mndReleaseDb(pMnode, pDb);

source/dnode/mnode/impl/src/mndDnode.c

@@ -347,6 +347,7 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
     SVgObj *pVgroup = mndAcquireVgroup(pMnode, pVload->vgId);
     if (pVgroup != NULL) {
       if (pVload->syncState == TAOS_SYNC_STATE_LEADER) {
+        pVgroup->cacheUsage = pVload->cacheUsage;
         pVgroup->numOfTables = pVload->numOfTables;
         pVgroup->numOfTimeSeries = pVload->numOfTimeSeries;
         pVgroup->totalStorage = pVload->totalStorage;

source/dnode/mnode/impl/src/mndVgroup.c

@@ -696,6 +696,9 @@ static int32_t mndRetrieveVgroups(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *p
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppendNULL(pColInfo, numOfRows);
 
+    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
+    colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->cacheUsage, false);
+
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppendNULL(pColInfo, numOfRows);
 

source/dnode/vnode/CMakeLists.txt

@@ -29,6 +29,7 @@ target_sources(
     # sma
     "src/sma/smaEnv.c"
     "src/sma/smaUtil.c"
+    "src/sma/smaFS.c"
     "src/sma/smaOpen.c"
     "src/sma/smaCommit.c"
     "src/sma/smaRollup.c"
@@ -49,6 +50,10 @@ target_sources(
     "src/tsdb/tsdbSnapshot.c"
     "src/tsdb/tsdbCacheRead.c"
     "src/tsdb/tsdbRetention.c"
+    "src/tsdb/tsdbDiskData.c"
+    "src/tsdb/tsdbCompress.c"
+    "src/tsdb/tsdbCompact.c"
+    "src/tsdb/tsdbMergeTree.c"
 
     # tq
     "src/tq/tq.c"

source/dnode/vnode/inc/vnode.h

@@ -155,6 +155,7 @@ int32_t tsdbGetTableSchema(SVnode *pVnode, int64_t uid, STSchema **pSchema, int6
 
 void   tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity);
 size_t tsdbCacheGetCapacity(SVnode *pVnode);
+size_t tsdbCacheGetUsage(SVnode *pVnode);
 
 // tq
 typedef struct SMetaTableInfo {

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

@@ -38,9 +38,10 @@ typedef struct SSmaEnv       SSmaEnv;
 typedef struct SSmaStat      SSmaStat;
 typedef struct STSmaStat     STSmaStat;
 typedef struct SRSmaStat     SRSmaStat;
-typedef struct SSmaKey       SSmaKey;
+typedef struct SRSmaRef      SRSmaRef;
 typedef struct SRSmaInfo     SRSmaInfo;
 typedef struct SRSmaInfoItem SRSmaInfoItem;
+typedef struct SRSmaFS       SRSmaFS;
 typedef struct SQTaskFile    SQTaskFile;
 typedef struct SQTaskFReader SQTaskFReader;
 typedef struct SQTaskFWriter SQTaskFWriter;
@@ -54,10 +55,21 @@ struct SSmaEnv {
 
 #define SMA_ENV_FLG_CLOSE ((int8_t)0x1)
 
+struct SRSmaRef {
+  int64_t refId;  // for SRSmaStat
+  int64_t suid;
+};
+
 typedef struct {
   int8_t  inited;
   int32_t rsetId;
   void   *tmrHandle;  // shared by all fetch tasks
+  /**
+   * @brief key: void* of SRSmaInfoItem, value: SRSmaRef
+   *  N.B. Although there is a very small possibility that "void*" point to different objects while with the same
+   * address after release/renew, the functionality is not affected as it just used to fetch the rsma results.
+   */
+  SHashObj *refHash;  // shared by all vgroups
 } SSmaMgmt;
 
 #define SMA_ENV_LOCK(env)  (&(env)->lock)
@@ -73,22 +85,27 @@ struct STSmaStat {
 
 struct SQTaskFile {
   volatile int32_t nRef;
-  int64_t          commitID;
+  int32_t          padding;
+  int64_t          version;
   int64_t          size;
 };
 
 struct SQTaskFReader {
   SSma     *pSma;
-  SQTaskFile fTask;
+  int64_t   version;
   TdFilePtr pReadH;
 };
 struct SQTaskFWriter {
   SSma     *pSma;
-  SQTaskFile fTask;
+  int64_t   version;
   TdFilePtr pWriteH;
   char     *fname;
 };
 
+struct SRSmaFS {
+  SArray *aQTaskInf;  // array of SQTaskFile
+};
+
 struct SRSmaStat {
   SSma            *pSma;
   int64_t          commitAppliedVer;  // vnode applied version for async commit
@@ -98,7 +115,7 @@ struct SRSmaStat {
   volatile int32_t nFetchAll;         // active number of fetch all
   int8_t           triggerStat;       // shared by fetch tasks
   int8_t           commitStat;        // 0 not in committing, 1 in committing
-  SArray          *aTaskFile;         // qTaskFiles committed recently(for recovery/snapshot r/w)
+  SRSmaFS          fs;                // for recovery/snapshot r/w
   SHashObj        *infoHash;          // key: suid, value: SRSmaInfo
   tsem_t           notEmpty;          // has items in queue buffer
 };
@@ -118,6 +135,7 @@ struct SSmaStat {
 #define RSMA_TRIGGER_STAT(r) (&(r)->triggerStat)
 #define RSMA_COMMIT_STAT(r)  (&(r)->commitStat)
 #define RSMA_REF_ID(r)       ((r)->refId)
+#define RSMA_FS(r)           (&(r)->fs)
 #define RSMA_FS_LOCK(r)      (&(r)->lock)
 
 struct SRSmaInfoItem {
@@ -130,9 +148,9 @@ struct SRSmaInfoItem {
 };
 
 struct SRSmaInfo {
+  SSma     *pSma;
   STSchema *pTSchema;
   int64_t   suid;
-  int64_t   refId;     // refId of SRSmaStat
   int64_t   lastRecv;  // ms
   int8_t    assigned;  // 0 idle, 1 assgined for exec
   int8_t    delFlag;
@@ -163,14 +181,6 @@ enum {
   TASK_TRIGGER_STAT_DROPPED = 5,
 };
 
-enum {
-  RSMA_ROLE_CREATE = 0,
-  RSMA_ROLE_DROP = 1,
-  RSMA_ROLE_SUBMIT = 2,
-  RSMA_ROLE_FETCH = 3,
-  RSMA_ROLE_ITERATE = 4,
-};
-
 enum {
   RSMA_RESTORE_REBOOT = 1,
   RSMA_RESTORE_SYNC = 2,
@@ -182,88 +192,34 @@ typedef enum {
   RSMA_EXEC_COMMIT = 3,    // triggered by commit
 } ERsmaExecType;
 
+// sma
+int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
 void    tdDestroySmaEnv(SSmaEnv *pSmaEnv);
 void   *tdFreeSmaEnv(SSmaEnv *pSmaEnv);
-
-int32_t tdDropTSma(SSma *pSma, char *pMsg);
-int32_t tdDropTSmaData(SSma *pSma, int64_t indexUid);
-int32_t tdInsertRSmaData(SSma *pSma, char *msg);
-
 int32_t tdRefSmaStat(SSma *pSma, SSmaStat *pStat);
 int32_t tdUnRefSmaStat(SSma *pSma, SSmaStat *pStat);
-int32_t tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
-int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
-
-void   *tdAcquireSmaRef(int32_t rsetId, int64_t refId);
-int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId);
-
-int32_t tdCheckAndInitSmaEnv(SSma *pSma, int8_t smaType);
-
 int32_t tdLockSma(SSma *pSma);
 int32_t tdUnLockSma(SSma *pSma);
+void   *tdAcquireSmaRef(int32_t rsetId, int64_t refId);
+int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId);
 
-static FORCE_INLINE int8_t tdSmaStat(STSmaStat *pTStat) {
-  if (pTStat) {
-    return atomic_load_8(&pTStat->state);
-  }
-  return TSDB_SMA_STAT_UNKNOWN;
-}
-
-static FORCE_INLINE bool tdSmaStatIsOK(STSmaStat *pTStat, int8_t *state) {
-  if (!pTStat) {
-    return false;
-  }
-
-  if (state) {
-    *state = atomic_load_8(&pTStat->state);
-    return *state == TSDB_SMA_STAT_OK;
-  }
-  return atomic_load_8(&pTStat->state) == TSDB_SMA_STAT_OK;
-}
-
-static FORCE_INLINE bool tdSmaStatIsExpired(STSmaStat *pTStat) {
-  return pTStat ? (atomic_load_8(&pTStat->state) & TSDB_SMA_STAT_EXPIRED) : true;
-}
-
-static FORCE_INLINE bool tdSmaStatIsDropped(STSmaStat *pTStat) {
-  return pTStat ? (atomic_load_8(&pTStat->state) & TSDB_SMA_STAT_DROPPED) : true;
-}
-
-static FORCE_INLINE void tdSmaStatSetOK(STSmaStat *pTStat) {
-  if (pTStat) {
-    atomic_store_8(&pTStat->state, TSDB_SMA_STAT_OK);
-  }
-}
-
-static FORCE_INLINE void tdSmaStatSetExpired(STSmaStat *pTStat) {
-  if (pTStat) {
-    atomic_or_fetch_8(&pTStat->state, TSDB_SMA_STAT_EXPIRED);
-  }
-}
-
-static FORCE_INLINE void tdSmaStatSetDropped(STSmaStat *pTStat) {
-  if (pTStat) {
-    atomic_or_fetch_8(&pTStat->state, TSDB_SMA_STAT_DROPPED);
-  }
-}
-
-void           tdRSmaQTaskInfoGetFileName(int32_t vid, int64_t version, char *outputName);
-void           tdRSmaQTaskInfoGetFullName(int32_t vid, int64_t version, const char *path, char *outputName);
-int32_t        tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pInfo);
-void           tdFreeQTaskInfo(qTaskInfo_t *taskHandle, int32_t vgId, int32_t level);
-static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
-void          *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
+// rsma
+int32_t tdRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
+int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo);
 void   *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree);
-int32_t        tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat, SHashObj *pInfoHash);
+int32_t tdRSmaFSOpen(SSma *pSma, int64_t version);
+void    tdRSmaFSClose(SRSmaFS *fs);
+int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t version);
+void    tdRSmaFSUnRef(SSma *pSma, SRSmaStat *pStat, int64_t version);
+int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile);
+int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer);
+int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);
 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);
 
-int32_t tdProcessRSmaCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, const char *tbName);
-int32_t tdProcessRSmaRestoreImpl(SSma *pSma, int8_t type, int64_t qtaskFileVer);
-int32_t tdRsmaRestore(SSma *pSma, int8_t type, int64_t committedVer);
-
-int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
-int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg);
-int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
+void tdRSmaQTaskInfoGetFileName(int32_t vid, int64_t version, char *outputName);
+void tdRSmaQTaskInfoGetFullName(int32_t vid, int64_t version, const char *path, char *outputName);
 
 // smaFileUtil ================
 

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

@@ -42,15 +42,15 @@ typedef struct SMemTable     SMemTable;
 typedef struct STbDataIter   STbDataIter;
 typedef struct SMapData      SMapData;
 typedef struct SBlockIdx     SBlockIdx;
-typedef struct SBlock        SBlock;
-typedef struct SBlockL       SBlockL;
+typedef struct SDataBlk      SDataBlk;
+typedef struct SSttBlk       SSttBlk;
 typedef struct SColData      SColData;
 typedef struct SDiskDataHdr  SDiskDataHdr;
 typedef struct SBlockData    SBlockData;
 typedef struct SDelFile      SDelFile;
 typedef struct SHeadFile     SHeadFile;
 typedef struct SDataFile     SDataFile;
-typedef struct SLastFile     SLastFile;
+typedef struct SSttFile      SSttFile;
 typedef struct SSmaFile      SSmaFile;
 typedef struct SDFileSet     SDFileSet;
 typedef struct SDataFWriter  SDataFWriter;
@@ -64,10 +64,15 @@ typedef struct STsdbReadSnap STsdbReadSnap;
 typedef struct SBlockInfo    SBlockInfo;
 typedef struct SSmaInfo      SSmaInfo;
 typedef struct SBlockCol     SBlockCol;
+typedef struct SVersionRange SVersionRange;
+typedef struct SLDataIter    SLDataIter;
 
 #define TSDB_FILE_DLMT         ((uint32_t)0xF00AFA0F)
 #define TSDB_MAX_SUBBLOCKS     8
+#define TSDB_MAX_STT_FILE      16
+#define TSDB_DEFAULT_STT_FILE  8
 #define TSDB_FHDR_SIZE         512
+#define TSDB_DEFAULT_PAGE_SIZE 4096
 
 #define HAS_NONE  ((int8_t)0x1)
 #define HAS_NULL  ((int8_t)0x2)
@@ -79,6 +84,14 @@ typedef struct SBlockCol     SBlockCol;
 #define TSDBKEY_MIN ((TSDBKEY){.ts = TSKEY_MIN, .version = VERSION_MIN})
 #define TSDBKEY_MAX ((TSDBKEY){.ts = TSKEY_MAX, .version = VERSION_MAX})
 
+#define PAGE_CONTENT_SIZE(PAGE) ((PAGE) - sizeof(TSCKSUM))
+#define LOGIC_TO_FILE_OFFSET(LOFFSET, PAGE) \
+  ((LOFFSET) / PAGE_CONTENT_SIZE(PAGE) * (PAGE) + (LOFFSET) % PAGE_CONTENT_SIZE(PAGE))
+#define FILE_TO_LOGIC_OFFSET(OFFSET, PAGE) ((OFFSET) / (PAGE)*PAGE_CONTENT_SIZE(PAGE) + (OFFSET) % (PAGE))
+#define PAGE_OFFSET(PGNO, PAGE)            (((PGNO)-1) * (PAGE))
+#define OFFSET_PGNO(OFFSET, PAGE)          ((OFFSET) / (PAGE) + 1)
+#define LOGIC_TO_FILE_SIZE(LSIZE, PAGE)    OFFSET_PGNO(LOGIC_TO_FILE_OFFSET(LSIZE, PAGE), PAGE) * (PAGE)
+
 // tsdbUtil.c ==============================================================================================
 // TSDBROW
 #define TSDBROW_TS(ROW)                       (((ROW)->type == 0) ? (ROW)->pTSRow->ts : (ROW)->pBlockData->aTSKEY[(ROW)->iRow])
@@ -111,15 +124,15 @@ int32_t tTABLEIDCmprFn(const void *p1, const void *p2);
 int32_t tPutBlockCol(uint8_t *p, void *ph);
 int32_t tGetBlockCol(uint8_t *p, void *ph);
 int32_t tBlockColCmprFn(const void *p1, const void *p2);
-// SBlock
-void    tBlockReset(SBlock *pBlock);
-int32_t tPutBlock(uint8_t *p, void *ph);
-int32_t tGetBlock(uint8_t *p, void *ph);
-int32_t tBlockCmprFn(const void *p1, const void *p2);
-bool    tBlockHasSma(SBlock *pBlock);
-// SBlockL
-int32_t tPutBlockL(uint8_t *p, void *ph);
-int32_t tGetBlockL(uint8_t *p, void *ph);
+// SDataBlk
+void    tDataBlkReset(SDataBlk *pBlock);
+int32_t tPutDataBlk(uint8_t *p, void *ph);
+int32_t tGetDataBlk(uint8_t *p, void *ph);
+int32_t tDataBlkCmprFn(const void *p1, const void *p2);
+bool    tDataBlkHasSma(SDataBlk *pDataBlk);
+// SSttBlk
+int32_t tPutSttBlk(uint8_t *p, void *ph);
+int32_t tGetSttBlk(uint8_t *p, void *ph);
 // SBlockIdx
 int32_t tPutBlockIdx(uint8_t *p, void *ph);
 int32_t tGetBlockIdx(uint8_t *p, void *ph);
@@ -170,6 +183,7 @@ int32_t tGetDelData(uint8_t *p, void *ph);
 void    tMapDataReset(SMapData *pMapData);
 void    tMapDataClear(SMapData *pMapData);
 int32_t tMapDataPutItem(SMapData *pMapData, void *pItem, int32_t (*tPutItemFn)(uint8_t *, void *));
+int32_t tMapDataCopy(SMapData *pFrom, SMapData *pTo);
 void    tMapDataGetItemByIdx(SMapData *pMapData, int32_t idx, void *pItem, int32_t (*tGetItemFn)(uint8_t *, void *));
 int32_t tMapDataSearch(SMapData *pMapData, void *pSearchItem, int32_t (*tGetItemFn)(uint8_t *, void *),
                        int32_t (*tItemCmprFn)(const void *, const void *), void *pItem);
@@ -191,7 +205,6 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol
                         uint8_t **ppBuf);
 int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, int32_t nVal, SColData *pColData,
                           uint8_t **ppBuf);
-int32_t tsdbReadAndCheck(TdFilePtr pFD, int64_t offset, uint8_t **ppOut, int32_t size, int8_t toCheck);
 // tsdbMemTable ==============================================================================================
 // SMemTable
 int32_t  tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable);
@@ -215,7 +228,7 @@ bool    tsdbDelFileIsSame(SDelFile *pDelFile1, SDelFile *pDelFile2);
 int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype);
 int32_t tPutHeadFile(uint8_t *p, SHeadFile *pHeadFile);
 int32_t tPutDataFile(uint8_t *p, SDataFile *pDataFile);
-int32_t tPutLastFile(uint8_t *p, SLastFile *pLastFile);
+int32_t tPutSttFile(uint8_t *p, SSttFile *pSttFile);
 int32_t tPutSmaFile(uint8_t *p, SSmaFile *pSmaFile);
 int32_t tPutDelFile(uint8_t *p, SDelFile *pDelFile);
 int32_t tGetDelFile(uint8_t *p, SDelFile *pDelFile);
@@ -224,7 +237,7 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet);
 
 void tsdbHeadFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SHeadFile *pHeadF, char fname[]);
 void tsdbDataFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SDataFile *pDataF, char fname[]);
-void tsdbLastFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SLastFile *pLastF, char fname[]);
+void tsdbSttFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSttFile *pSttF, char fname[]);
 void tsdbSmaFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSmaFile *pSmaF, char fname[]);
 // SDelFile
 void tsdbDelFileName(STsdb *pTsdb, SDelFile *pFile, char fname[]);
@@ -250,7 +263,7 @@ int32_t tsdbDataFWriterClose(SDataFWriter **ppWriter, int8_t sync);
 int32_t tsdbUpdateDFileSetHeader(SDataFWriter *pWriter);
 int32_t tsdbWriteBlockIdx(SDataFWriter *pWriter, SArray *aBlockIdx);
 int32_t tsdbWriteBlock(SDataFWriter *pWriter, SMapData *pMapData, SBlockIdx *pBlockIdx);
-int32_t tsdbWriteBlockL(SDataFWriter *pWriter, SArray *aBlockL);
+int32_t tsdbWriteSttBlk(SDataFWriter *pWriter, SArray *aSttBlk);
 int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlockInfo *pBlkInfo, SSmaInfo *pSmaInfo,
                            int8_t cmprAlg, int8_t toLast);
 
@@ -260,10 +273,10 @@ int32_t tsdbDataFReaderOpen(SDataFReader **ppReader, STsdb *pTsdb, SDFileSet *pS
 int32_t tsdbDataFReaderClose(SDataFReader **ppReader);
 int32_t tsdbReadBlockIdx(SDataFReader *pReader, SArray *aBlockIdx);
 int32_t tsdbReadBlock(SDataFReader *pReader, SBlockIdx *pBlockIdx, SMapData *pMapData);
-int32_t tsdbReadBlockL(SDataFReader *pReader, SArray *aBlockL);
-int32_t tsdbReadBlockSma(SDataFReader *pReader, SBlock *pBlock, SArray *aColumnDataAgg);
-int32_t tsdbReadDataBlock(SDataFReader *pReader, SBlock *pBlock, SBlockData *pBlockData);
-int32_t tsdbReadLastBlock(SDataFReader *pReader, SBlockL *pBlockL, SBlockData *pBlockData);
+int32_t tsdbReadSttBlk(SDataFReader *pReader, int32_t iStt, SArray *aSttBlk);
+int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pBlock, SArray *aColumnDataAgg);
+int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pBlock, SBlockData *pBlockData);
+int32_t tsdbReadSttBlock(SDataFReader *pReader, int32_t iStt, SSttBlk *pSttBlk, SBlockData *pBlockData);
 // SDelFWriter
 int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb);
 int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync);
@@ -278,6 +291,8 @@ int32_t tsdbReadDelIdx(SDelFReader *pReader, SArray *aDelIdx);
 // tsdbRead.c ==============================================================================================
 int32_t tsdbTakeReadSnap(STsdb *pTsdb, STsdbReadSnap **ppSnap);
 void    tsdbUntakeReadSnap(STsdb *pTsdb, STsdbReadSnap *pSnap);
+// tsdbMerge.c ==============================================================================================
+int32_t tsdbMerge(STsdb *pTsdb);
 
 #define TSDB_CACHE_NO(c)       ((c).cacheLast == 0)
 #define TSDB_CACHE_LAST_ROW(c) (((c).cacheLast & 1) > 0)
@@ -324,6 +339,11 @@ struct TSDBKEY {
   TSKEY   ts;
 };
 
+struct SVersionRange {
+  uint64_t minVer;
+  uint64_t maxVer;
+};
+
 typedef struct SMemSkipListNode SMemSkipListNode;
 struct SMemSkipListNode {
   int8_t            level;
@@ -416,7 +436,7 @@ struct SSmaInfo {
   int32_t size;
 };
 
-struct SBlock {
+struct SDataBlk {
   TSDBKEY    minKey;
   TSDBKEY    maxKey;
   int64_t    minVer;
@@ -428,7 +448,7 @@ struct SBlock {
   SSmaInfo   smaInfo;
 };
 
-struct SBlockL {
+struct SSttBlk {
   int64_t    suid;
   int64_t    minUid;
   int64_t    maxUid;
@@ -467,12 +487,6 @@ struct SBlockData {
   SArray  *aColData;  // SArray<SColData>
 };
 
-// ================== TSDB global config
-extern bool tsdbForceKeepFile;
-
-#define TSDB_FS_ITER_FORWARD  TSDB_ORDER_ASC
-#define TSDB_FS_ITER_BACKWARD TSDB_ORDER_DESC
-
 struct TABLEID {
   tb_uid_t suid;
   tb_uid_t uid;
@@ -536,7 +550,7 @@ struct SDataFile {
   int64_t size;
 };
 
-struct SLastFile {
+struct SSttFile {
   volatile int32_t nRef;
 
   int64_t commitID;
@@ -556,8 +570,9 @@ struct SDFileSet {
   int32_t    fid;
   SHeadFile *pHeadF;
   SDataFile *pDataF;
-  SLastFile *pLastF;
   SSmaFile  *pSmaF;
+  uint8_t    nSttF;
+  SSttFile  *aSttF[TSDB_MAX_STT_FILE];
 };
 
 struct SRowIter {
@@ -572,37 +587,76 @@ struct SRowMerger {
   SArray   *pArray;  // SArray<SColVal>
 };
 
+typedef struct {
+  char     *path;
+  int32_t   szPage;
+  int32_t   flag;
+  TdFilePtr pFD;
+  int64_t   pgno;
+  uint8_t  *pBuf;
+  int64_t   szFile;
+} STsdbFD;
+
 struct SDelFWriter {
   STsdb   *pTsdb;
   SDelFile fDel;
-  TdFilePtr pWriteH;
-
+  STsdbFD *pWriteH;
   uint8_t *aBuf[1];
 };
 
+struct STsdbReadSnap {
+  SMemTable *pMem;
+  SMemTable *pIMem;
+  STsdbFS    fs;
+};
+
 struct SDataFWriter {
   STsdb    *pTsdb;
   SDFileSet wSet;
 
-  TdFilePtr pHeadFD;
-  TdFilePtr pDataFD;
-  TdFilePtr pLastFD;
-  TdFilePtr pSmaFD;
+  STsdbFD *pHeadFD;
+  STsdbFD *pDataFD;
+  STsdbFD *pSmaFD;
+  STsdbFD *pSttFD;
 
   SHeadFile fHead;
   SDataFile fData;
-  SLastFile fLast;
   SSmaFile  fSma;
+  SSttFile  fStt[TSDB_MAX_STT_FILE];
 
   uint8_t *aBuf[4];
 };
 
-struct STsdbReadSnap {
-  SMemTable *pMem;
-  SMemTable *pIMem;
-  STsdbFS    fs;
+struct SDataFReader {
+  STsdb     *pTsdb;
+  SDFileSet *pSet;
+  STsdbFD   *pHeadFD;
+  STsdbFD   *pDataFD;
+  STsdbFD   *pSmaFD;
+  STsdbFD   *aSttFD[TSDB_MAX_STT_FILE];
+  uint8_t   *aBuf[3];
 };
 
+typedef struct {
+  int64_t suid;
+  int64_t uid;
+  TSDBROW row;
+} SRowInfo;
+
+typedef struct SMergeTree {
+  int8_t      backward;
+  SRBTree     rbt;
+  SArray     *pIterList;
+  SLDataIter *pIter;
+} SMergeTree;
+
+int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t suid, uint64_t uid,
+                       STimeWindow *pTimeWindow, SVersionRange *pVerRange);
+void    tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
+bool    tMergeTreeNext(SMergeTree *pMTree);
+TSDBROW tMergeTreeGetRow(SMergeTree *pMTree);
+void    tMergeTreeClose(SMergeTree *pMTree);
+
 // ========== inline functions ==========
 static FORCE_INLINE int32_t tsdbKeyCmprFn(const void *p1, const void *p2) {
   TSDBKEY *pKey1 = (TSDBKEY *)p1;

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

@@ -36,6 +36,7 @@
 #include "tlosertree.h"
 #include "tlrucache.h"
 #include "tmsgcb.h"
+#include "trbtree.h"
 #include "tref.h"
 #include "tskiplist.h"
 #include "tstream.h"
@@ -70,8 +71,8 @@ typedef struct SStreamTaskReader  SStreamTaskReader;
 typedef struct SStreamTaskWriter  SStreamTaskWriter;
 typedef struct SStreamStateReader SStreamStateReader;
 typedef struct SStreamStateWriter SStreamStateWriter;
-typedef struct SRsmaSnapReader    SRsmaSnapReader;
-typedef struct SRsmaSnapWriter    SRsmaSnapWriter;
+typedef struct SRSmaSnapReader    SRSmaSnapReader;
+typedef struct SRSmaSnapWriter    SRSmaSnapWriter;
 typedef struct SSnapDataHdr       SSnapDataHdr;
 
 #define VNODE_META_DIR  "meta"
@@ -103,7 +104,7 @@ int             metaCreateSTable(SMeta* pMeta, int64_t version, SVCreateStbReq*
 int             metaAlterSTable(SMeta* pMeta, int64_t version, SVCreateStbReq* pReq);
 int             metaDropSTable(SMeta* pMeta, int64_t verison, SVDropStbReq* pReq, SArray* tbUidList);
 int             metaCreateTable(SMeta* pMeta, int64_t version, SVCreateTbReq* pReq, STableMetaRsp** pMetaRsp);
-int             metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq, SArray* tbUids);
+int             metaDropTable(SMeta* pMeta, int64_t version, SVDropTbReq* pReq, SArray* tbUids, int64_t *tbUid);
 int             metaTtlDropTable(SMeta* pMeta, int64_t ttl, SArray* tbUids);
 int             metaAlterTable(SMeta* pMeta, int64_t version, SVAlterTbReq* pReq, STableMetaRsp* pMetaRsp);
 SSchemaWrapper* metaGetTableSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver, bool isinline);
@@ -207,7 +208,7 @@ int32_t tdProcessRSmaCreate(SSma* pSma, SVCreateStbReq* pReq);
 int32_t tdProcessRSmaSubmit(SSma* pSma, void* pMsg, int32_t inputType);
 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);
+int32_t tdUpdateTbUidList(SSma* pSma, STbUidStore* pUidStore, bool isAdd);
 void    tdUidStoreDestory(STbUidStore* pStore);
 void*   tdUidStoreFree(STbUidStore* pStore);
 
@@ -247,14 +248,14 @@ int32_t tqOffsetSnapWrite(STqOffsetWriter* pWriter, uint8_t* pData, uint32_t nDa
 // SStreamTaskReader ======================================
 // SStreamStateWriter =====================================
 // SStreamStateReader =====================================
-// SRsmaSnapReader ========================================
-int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapReader** ppReader);
-int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader);
-int32_t rsmaSnapRead(SRsmaSnapReader* pReader, uint8_t** ppData);
-// SRsmaSnapWriter ========================================
-int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapWriter** ppWriter);
-int32_t rsmaSnapWrite(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
-int32_t rsmaSnapWriterClose(SRsmaSnapWriter** ppWriter, int8_t rollback);
+// SRSmaSnapReader ========================================
+int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapReader** ppReader);
+int32_t rsmaSnapReaderClose(SRSmaSnapReader** ppReader);
+int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData);
+// SRSmaSnapWriter ========================================
+int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter);
+int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
+int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback);
 
 typedef struct {
   int8_t  streamType;  // sma or other

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

@@ -291,6 +291,38 @@ _query:
       tDecoderClear(&dc);
       goto _exit;
     }
+    { // Traverse to find the previous qualified data
+      TBC      *pCur;
+      tdbTbcOpen(pMeta->pTbDb, &pCur, NULL);
+      STbDbKey key = {.version = sver, .uid = INT64_MAX};
+      int c = 0;
+      tdbTbcMoveTo(pCur, &key, sizeof(key), &c);
+      if(c < 0){
+        tdbTbcMoveToPrev(pCur);
+      }
+
+      void *pKey = NULL;
+      void *pVal = NULL;
+      int   vLen = 0, kLen = 0;
+      while(1){
+        int32_t ret = tdbTbcPrev(pCur, &pKey, &kLen, &pVal, &vLen);
+        if (ret < 0) break;
+
+        STbDbKey *tmp = (STbDbKey*)pKey;
+        if(tmp->uid != uid){
+          continue;
+        }
+        SDecoder   dcNew = {0};
+        SMetaEntry meNew = {0};
+        tDecoderInit(&dcNew, pVal, vLen);
+        metaDecodeEntry(&dcNew, &meNew);
+        pSchema = tCloneSSchemaWrapper(&meNew.stbEntry.schemaRow);
+        tDecoderClear(&dcNew);
+        tdbTbcClose(pCur);
+        goto _exit;
+      }
+      tdbTbcClose(pCur);
+    }
   } else if (me.type == TSDB_CHILD_TABLE) {
     uid = me.ctbEntry.suid;
     tDecoderClear(&dc);

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

@@ -474,7 +474,7 @@ _err:
   return -1;
 }
 
-int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids) {
+int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUids, tb_uid_t *tbUid) {
   void    *pData = NULL;
   int      nData = 0;
   int      rc = 0;
@@ -496,6 +496,10 @@ int metaDropTable(SMeta *pMeta, int64_t version, SVDropTbReq *pReq, SArray *tbUi
     taosArrayPush(tbUids, &uid);
   }
 
+  if ((type == TSDB_CHILD_TABLE) && tbUid) {
+    *tbUid = uid;
+  }
+
   tdbFree(pData);
   return 0;
 }

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

@@ -15,13 +15,15 @@
 
 #include "sma.h"
 
+extern SSmaMgmt smaMgmt;
+
 static int32_t tdProcessRSmaSyncPreCommitImpl(SSma *pSma);
 static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma);
 static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma);
 static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma);
 static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma);
 static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma);
-static int32_t tdCleanupQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat);
+static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat);
 
 /**
  * @brief Only applicable to Rollup SMA
@@ -166,114 +168,51 @@ static int32_t tdProcessRSmaSyncCommitImpl(SSma *pSma) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t tdCleanupQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat) {
-  SVnode       *pVnode = pSma->pVnode;
-  int64_t       committed = pRSmaStat->commitAppliedVer;
-  TdDirPtr      pDir = NULL;
-  TdDirEntryPtr pDirEntry = NULL;
-  char          dir[TSDB_FILENAME_LEN];
-  const char   *pattern = "v[0-9]+qinf\\.v([0-9]+)?$";
-  regex_t       regex;
-  int           code = 0;
-
-  tdGetVndDirName(TD_VID(pVnode), tfsGetPrimaryPath(pVnode->pTfs), VNODE_RSMA_DIR, true, dir);
-
-  // Resource allocation and init
-  if ((code = regcomp(&regex, pattern, REG_EXTENDED)) != 0) {
-    char errbuf[128];
-    regerror(code, &regex, errbuf, sizeof(errbuf));
-    smaWarn("vgId:%d, rsma post commit, regcomp for %s failed since %s", TD_VID(pVnode), dir, errbuf);
-    return TSDB_CODE_FAILED;
-  }
-
-  if ((pDir = taosOpenDir(dir)) == NULL) {
-    regfree(&regex);
-    terrno = TAOS_SYSTEM_ERROR(errno);
-    smaDebug("vgId:%d, rsma post commit, open dir %s failed since %s", TD_VID(pVnode), dir, terrstr());
-    return TSDB_CODE_FAILED;
-  }
-
-  int32_t    dirLen = strlen(dir);
-  char      *dirEnd = POINTER_SHIFT(dir, dirLen);
-  regmatch_t regMatch[2];
-  while ((pDirEntry = taosReadDir(pDir)) != NULL) {
-    char *entryName = taosGetDirEntryName(pDirEntry);
-    if (!entryName) {
-      continue;
-    }
-
-    code = regexec(&regex, entryName, 2, regMatch, 0);
-
-    if (code == 0) {
-      // match
-      int64_t version = -1;
-      sscanf((const char *)POINTER_SHIFT(entryName, regMatch[1].rm_so), "%" PRIi64, &version);
-      if ((version < committed) && (version > -1)) {
-        strncpy(dirEnd, entryName, TSDB_FILENAME_LEN - dirLen);
-        if (taosRemoveFile(dir) != 0) {
-          terrno = TAOS_SYSTEM_ERROR(errno);
-          smaWarn("vgId:%d, committed version:%" PRIi64 ", failed to remove %s since %s", TD_VID(pVnode), committed,
-                  dir, terrstr());
-        } else {
-          smaDebug("vgId:%d, committed version:%" PRIi64 ", success to remove %s", TD_VID(pVnode), committed, dir);
-        }
-      }
-    } else if (code == REG_NOMATCH) {
-      // not match
-      smaTrace("vgId:%d, rsma post commit, not match %s", TD_VID(pVnode), entryName);
-      continue;
-    } else {
-      // has other error
-      char errbuf[128];
-      regerror(code, &regex, errbuf, sizeof(errbuf));
-      smaWarn("vgId:%d, rsma post commit, regexec failed since %s", TD_VID(pVnode), errbuf);
-
-      taosCloseDir(&pDir);
-      regfree(&regex);
-      return TSDB_CODE_FAILED;
-    }
-  }
-
-  taosCloseDir(&pDir);
-  regfree(&regex);
-
-  return TSDB_CODE_SUCCESS;
-}
-
 // SQTaskFile ======================================================
-// int32_t tCmprQTaskFile(void const *lhs, void const *rhs) {
-//   int64_t    *lCommitted = *(int64_t *)lhs;
-//   SQTaskFile *rQTaskF = (SQTaskFile *)rhs;
 
-//   if (lCommitted < rQTaskF->commitID) {
-//     return -1;
-//   } else if (lCommitted > rQTaskF->commitID) {
-//     return 1;
-//   }
-
-//   return 0;
-// }
-
-#if 0
 /**
  * @brief At most time, there is only one qtaskinfo file committed latest in aTaskFile. Sometimes, there would be
  * multiple qtaskinfo files supporting snapshot replication.
  *
  * @param pSma
- * @param pRSmaStat
+ * @param pStat
  * @return int32_t
  */
-static int32_t tdCleanupQTaskInfoFiles(SSma *pSma, SRSmaStat *pRSmaStat) {
+static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) {
   SVnode  *pVnode = pSma->pVnode;
-  int64_t committed = pRSmaStat->commitAppliedVer;
-  SArray *aTaskFile = pRSmaStat->aTaskFile;
-
-  void *qTaskFile = taosArraySearch(aTaskFile, committed, tCmprQTaskFile, TD_LE);
+  SRSmaFS *pFS = RSMA_FS(pStat);
+  int64_t  committed = pStat->commitAppliedVer;
+  char     qTaskInfoFullName[TSDB_FILENAME_LEN];
+
+  taosWLockLatch(RSMA_FS_LOCK(pStat));
+
+  for (int32_t i = 0; i < taosArrayGetSize(pFS->aQTaskInf);) {
+    SQTaskFile *pTaskF = taosArrayGet(pFS->aQTaskInf, i);
+    int32_t     oldVal = atomic_fetch_sub_32(&pTaskF->nRef, 1);
+    if ((oldVal <= 1) && (pTaskF->version < committed)) {
+      tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->version, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
+      if (taosRemoveFile(qTaskInfoFullName) < 0) {
+        smaWarn("vgId:%d, cleanup qinf, committed %" PRIi64 ", failed to remove %s since %s", TD_VID(pVnode), committed,
+                qTaskInfoFullName, tstrerror(TAOS_SYSTEM_ERROR(errno)));
+      } else {
+        smaDebug("vgId:%d, cleanup qinf, committed %" PRIi64 ", success to remove %s", TD_VID(pVnode), committed,
+                 qTaskInfoFullName);
+      }
+      taosArrayRemove(pFS->aQTaskInf, i);
+      continue;
+    }
+    ++i;
+  }
 
+  SQTaskFile qFile = {.nRef = 1, .padding = 0, .version = committed, .size = 0};
+  if (tdRSmaFSUpsertQTaskFile(pFS, &qFile) < 0) {
+    taosWUnLockLatch(RSMA_FS_LOCK(pStat));
+    return TSDB_CODE_FAILED;
+  }
 
+  taosWUnLockLatch(RSMA_FS_LOCK(pStat));
   return TSDB_CODE_SUCCESS;
 }
-#endif
 
 /**
  * @brief post-commit for rollup sma
@@ -290,8 +229,7 @@ static int32_t tdProcessRSmaSyncPostCommitImpl(SSma *pSma) {
 
   SRSmaStat *pRSmaStat = SMA_RSMA_STAT(pSma);
 
-  // cleanup outdated qtaskinfo files
-  tdCleanupQTaskInfoFiles(pSma, pRSmaStat);
+  tdUpdateQTaskInfoFiles(pSma, pRSmaStat);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -428,7 +366,6 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
   }
 
   SRSmaStat     *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
-  SArray    *rsmaDeleted = NULL;
 
   // step 1: merge qTaskInfo and iQTaskInfo
   // lock
@@ -441,11 +378,7 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
     if (RSMA_INFO_IS_DEL(pRSmaInfo)) {
       int32_t refVal = T_REF_VAL_GET(pRSmaInfo);
       if (refVal == 0) {
-        if (!rsmaDeleted) {
-          if ((rsmaDeleted = taosArrayInit(1, sizeof(tb_uid_t)))) {
-            taosArrayPush(rsmaDeleted, pSuid);
-          }
-        }
+        taosHashRemove(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(*pSuid));
       } else {
         smaDebug(
             "vgId:%d, rsma async post commit, not free rsma info since ref is %d although already deleted for "
@@ -471,25 +404,10 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
 #endif
   }
 
-  for (int32_t i = 0; i < taosArrayGetSize(rsmaDeleted); ++i) {
-    tb_uid_t *pSuid = taosArrayGet(rsmaDeleted, i);
-    void     *pRSmaInfo = taosHashGet(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t));
-    if ((pRSmaInfo = *(SRSmaInfo **)pRSmaInfo)) {
-      tdFreeRSmaInfo(pSma, pRSmaInfo, true);
-      smaDebug(
-          "vgId:%d, rsma async post commit, free rsma info since already deleted and ref is 0 for "
-          "table:%" PRIi64,
-          SMA_VID(pSma), *pSuid);
-    }
-    taosHashRemove(RSMA_INFO_HASH(pRSmaStat), pSuid, sizeof(tb_uid_t));
-  }
-  taosArrayDestroy(rsmaDeleted);
-
   // unlock
   // taosWUnLockLatch(SMA_ENV_LOCK(pEnv));
 
-  // step 2: cleanup outdated qtaskinfo files
-  tdCleanupQTaskInfoFiles(pSma, pRSmaStat);
+  tdUpdateQTaskInfoFiles(pSma, pRSmaStat);
 
   atomic_store_8(RSMA_COMMIT_STAT(pRSmaStat), 0);
 

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

@@ -28,6 +28,8 @@ static int32_t tdInitSmaEnv(SSma *pSma, int8_t smaType, SSmaEnv **ppEnv);
 static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pSma);
 static int32_t tdRsmaStartExecutor(const SSma *pSma);
 static int32_t tdRsmaStopExecutor(const SSma *pSma);
+static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType);
+static void   *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType);
 static void   *tdFreeTSmaStat(STSmaStat *pStat);
 static void    tdDestroyRSmaStat(void *pRSmaStat);
 
@@ -59,12 +61,23 @@ int32_t smaInit() {
       return TSDB_CODE_FAILED;
     }
 
+    int32_t type = (8 == POINTER_BYTES) ? TSDB_DATA_TYPE_UBIGINT : TSDB_DATA_TYPE_UINT;
+    smaMgmt.refHash = taosHashInit(64, taosGetDefaultHashFunction(type), true, HASH_ENTRY_LOCK);
+    if (!smaMgmt.refHash) {
+      taosCloseRef(smaMgmt.rsetId);
+      atomic_store_8(&smaMgmt.inited, 0);
+      smaError("failed to init sma tmr hanle since %s", terrstr());
+      return TSDB_CODE_FAILED;
+    }
+
     // init fetch timer handle
     smaMgmt.tmrHandle = taosTmrInit(10000, 100, 10000, "RSMA");
     if (!smaMgmt.tmrHandle) {
       taosCloseRef(smaMgmt.rsetId);
+      taosHashCleanup(smaMgmt.refHash);
+      smaMgmt.refHash = NULL;
       atomic_store_8(&smaMgmt.inited, 0);
-      smaError("failed to init sma tmr hanle since %s", terrstr());
+      smaError("failed to init sma tmr handle since %s", terrstr());
       return TSDB_CODE_FAILED;
     }
 
@@ -93,6 +106,8 @@ void smaCleanUp() {
 
   if (old == 1) {
     taosCloseRef(smaMgmt.rsetId);
+    taosHashCleanup(smaMgmt.refHash);
+    smaMgmt.refHash = NULL;
     taosTmrCleanUp(smaMgmt.tmrHandle);
     smaInfo("sma mgmt env is cleaned up, rsetId:%d, tmrHandle:%p", smaMgmt.rsetId, smaMgmt.tmrHandle);
     atomic_store_8(&smaMgmt.inited, 0);
@@ -195,6 +210,21 @@ int32_t tdUnRefRSmaInfo(SSma *pSma, SRSmaInfo *pRSmaInfo) {
   return 0;
 }
 
+static void tRSmaInfoHashFreeNode(void *data) {
+  SRSmaInfo     *pRSmaInfo = NULL;
+  SRSmaInfoItem *pItem = NULL;
+
+  if ((pRSmaInfo = *(SRSmaInfo **)data)) {
+    if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 0)) && pItem->level) {
+      taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
+    }
+    if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 1)) && pItem->level) {
+      taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
+    }
+    tdFreeRSmaInfo(pRSmaInfo->pSma, pRSmaInfo, true);
+  }
+}
+
 static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pSma) {
   ASSERT(pSmaStat != NULL);
 
@@ -240,10 +270,16 @@ static int32_t tdInitSmaStat(SSmaStat **pSmaStat, int8_t smaType, const SSma *pS
       if (!RSMA_INFO_HASH(pRSmaStat)) {
         return TSDB_CODE_FAILED;
       }
+      taosHashSetFreeFp(RSMA_INFO_HASH(pRSmaStat), tRSmaInfoHashFreeNode);
 
       if (tdRsmaStartExecutor(pSma) < 0) {
         return TSDB_CODE_FAILED;
       }
+
+      if (!(RSMA_FS(pRSmaStat)->aQTaskInf = taosArrayInit(1, sizeof(SQTaskFile)))) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        return TSDB_CODE_FAILED;
+      }
     } else if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
       // TODO
     } else {
@@ -278,14 +314,6 @@ static void tdDestroyRSmaStat(void *pRSmaStat) {
     tsem_destroy(&(pStat->notEmpty));
 
     // step 2: destroy the rsma info and associated fetch tasks
-    if (taosHashGetSize(RSMA_INFO_HASH(pStat)) > 0) {
-      void *infoHash = taosHashIterate(RSMA_INFO_HASH(pStat), NULL);
-      while (infoHash) {
-        SRSmaInfo *pSmaInfo = *(SRSmaInfo **)infoHash;
-        tdFreeRSmaInfo(pSma, pSmaInfo, true);
-        infoHash = taosHashIterate(RSMA_INFO_HASH(pStat), infoHash);
-      }
-    }
     taosHashCleanup(RSMA_INFO_HASH(pStat));
 
     // step 3: wait for all triggered fetch tasks to finish
@@ -307,12 +335,15 @@ static void tdDestroyRSmaStat(void *pRSmaStat) {
     // step 4:
     tdRsmaStopExecutor(pSma);
 
-    // step 5: free pStat
+    // step 5:
+    tdRSmaFSClose(RSMA_FS(pStat));
+
+    // step 6: free pStat
     taosMemoryFreeClear(pStat);
   }
 }
 
-void *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType) {
+static void *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType) {
   tdDestroySmaState(pSmaStat, smaType);
   if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
     taosMemoryFreeClear(pSmaStat);
@@ -329,7 +360,7 @@ void *tdFreeSmaState(SSmaStat *pSmaStat, int8_t smaType) {
  * @return int32_t
  */
 
-int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
+static int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
   if (pSmaStat) {
     if (smaType == TSDB_SMA_TYPE_TIME_RANGE) {
       tdDestroyTSmaStat(SMA_STAT_TSMA(pSmaStat));
@@ -337,7 +368,7 @@ int32_t tdDestroySmaState(SSmaStat *pSmaStat, int8_t smaType) {
       SRSmaStat *pRSmaStat = &pSmaStat->rsmaStat;
       int32_t    vid = SMA_VID(pRSmaStat->pSma);
       int64_t    refId = RSMA_REF_ID(pRSmaStat);
-      if (taosRemoveRef(smaMgmt.rsetId, RSMA_REF_ID(pRSmaStat)) < 0) {
+      if (taosRemoveRef(smaMgmt.rsetId, refId) < 0) {
         smaError("vgId:%d, remove refId:%" PRIi64 " from rsmaRef:%" PRIi32 " failed since %s", vid, refId,
                  smaMgmt.rsetId, terrstr());
       } else {

source/dnode/vnode/src/sma/smaFS.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"
+
+// =================================================================================================
+
+static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **output);
+static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2);
+static int32_t tdQTaskInfCmprFn2(const void *p1, const void *p2);
+/**
+ * @brief Open RSma FS from qTaskInfo files
+ *
+ * @param pSma
+ * @param version
+ * @return int32_t
+ */
+int32_t tdRSmaFSOpen(SSma *pSma, int64_t version) {
+  SVnode    *pVnode = pSma->pVnode;
+  int64_t    commitID = pVnode->state.commitID;
+  SSmaEnv   *pEnv = SMA_RSMA_ENV(pSma);
+  SRSmaStat *pStat = NULL;
+  SArray    *output = NULL;
+  
+  terrno = TSDB_CODE_SUCCESS;
+
+  if (!pEnv) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  if (tdFetchQTaskInfoFiles(pSma, version, &output) < 0) {
+    goto _end;
+  }
+
+  pStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
+
+  for (int32_t i = 0; i < taosArrayGetSize(output); ++i) {
+    int32_t vid = 0;
+    int64_t version = -1;
+    sscanf((const char *)taosArrayGetP(output, i), "v%dqinfo.v%" PRIi64, &vid, &version);
+    SQTaskFile qTaskFile = {.version = version, .nRef = 1};
+    if ((terrno = tdRSmaFSUpsertQTaskFile(RSMA_FS(pStat), &qTaskFile)) < 0) {
+      goto _end;
+    }
+    smaInfo("vgId:%d, open fs, version:%" PRIi64 ", ref:%" PRIi64, TD_VID(pVnode), qTaskFile.version, qTaskFile.nRef);
+  }
+
+_end:
+  for (int32_t i = 0; i < taosArrayGetSize(output); ++i) {
+    void *ptr = taosArrayGetP(output, i);
+    taosMemoryFreeClear(ptr);
+  }
+  taosArrayDestroy(output);
+
+  if (terrno != TSDB_CODE_SUCCESS) {
+    smaError("vgId:%d, open rsma fs failed since %s", TD_VID(pVnode), terrstr());
+    return TSDB_CODE_FAILED;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+void tdRSmaFSClose(SRSmaFS *fs) { taosArrayDestroy(fs->aQTaskInf); }
+
+static int32_t tdQTaskInfCmprFn1(const void *p1, const void *p2) {
+  if (*(int64_t *)p1 < ((SQTaskFile *)p2)->version) {
+    return -1;
+  } else if (*(int64_t *)p1 > ((SQTaskFile *)p2)->version) {
+    return 1;
+  }
+  return 0;
+}
+
+int32_t tdRSmaFSRef(SSma *pSma, SRSmaStat *pStat, int64_t version) {
+  SArray     *aQTaskInf = RSMA_FS(pStat)->aQTaskInf;
+  SQTaskFile *pTaskF = NULL;
+  int32_t     oldVal = 0;
+
+  taosRLockLatch(RSMA_FS_LOCK(pStat));
+  if ((pTaskF = taosArraySearch(aQTaskInf, &version, tdQTaskInfCmprFn1, TD_EQ))) {
+    oldVal = atomic_fetch_add_32(&pTaskF->nRef, 1);
+    ASSERT(oldVal > 0);
+  }
+  taosRUnLockLatch(RSMA_FS_LOCK(pStat));
+  return oldVal;
+}
+
+void tdRSmaFSUnRef(SSma *pSma, SRSmaStat *pStat, int64_t version) {
+  SVnode     *pVnode = pSma->pVnode;
+  SArray     *aQTaskInf = RSMA_FS(pStat)->aQTaskInf;
+  char        qTaskFullName[TSDB_FILENAME_LEN];
+  SQTaskFile *pTaskF = NULL;
+  int32_t     idx = -1;
+
+  taosWLockLatch(RSMA_FS_LOCK(pStat));
+  if ((idx = taosArraySearchIdx(aQTaskInf, &version, tdQTaskInfCmprFn1, TD_EQ)) >= 0) {
+    ASSERT(idx < taosArrayGetSize(aQTaskInf));
+    pTaskF = taosArrayGet(aQTaskInf, idx);
+    if (atomic_sub_fetch_32(&pTaskF->nRef, 1) <= 0) {
+      tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->version, tfsGetPrimaryPath(pVnode->pTfs), qTaskFullName);
+      if (taosRemoveFile(qTaskFullName) < 0) {
+        smaWarn("vgId:%d, failed to remove %s since %s", TD_VID(pVnode), qTaskFullName,
+                tstrerror(TAOS_SYSTEM_ERROR(errno)));
+      } else {
+        smaDebug("vgId:%d, success to remove %s", TD_VID(pVnode), qTaskFullName);
+      }
+      taosArrayRemove(aQTaskInf, idx);
+    }
+  }
+  taosWUnLockLatch(RSMA_FS_LOCK(pStat));
+}
+
+/**
+ * @brief Fetch qtaskfiles LE than version
+ *
+ * @param pSma
+ * @param version
+ * @param output
+ * @return int32_t
+ */
+static int32_t tdFetchQTaskInfoFiles(SSma *pSma, int64_t version, SArray **output) {
+  SVnode       *pVnode = pSma->pVnode;
+  TdDirPtr      pDir = NULL;
+  TdDirEntryPtr pDirEntry = NULL;
+  char          dir[TSDB_FILENAME_LEN];
+  const char   *pattern = "v[0-9]+qinf\\.v([0-9]+)?$";
+  regex_t       regex;
+  int           code = 0;
+
+  tdGetVndDirName(TD_VID(pVnode), tfsGetPrimaryPath(pVnode->pTfs), VNODE_RSMA_DIR, true, dir);
+
+  if (!taosCheckExistFile(dir)) {
+    smaDebug("vgId:%d, fetch qtask files, no need as dir %s not exist", TD_VID(pVnode), dir);
+    return TSDB_CODE_SUCCESS;
+  }
+
+  // Resource allocation and init
+  if ((code = regcomp(&regex, pattern, REG_EXTENDED)) != 0) {
+    terrno = TSDB_CODE_RSMA_REGEX_MATCH;
+    char errbuf[128];
+    regerror(code, &regex, errbuf, sizeof(errbuf));
+    smaWarn("vgId:%d, fetch qtask files, regcomp for %s failed since %s", TD_VID(pVnode), dir, errbuf);
+    return TSDB_CODE_FAILED;
+  }
+
+  if (!(pDir = taosOpenDir(dir))) {
+    regfree(&regex);
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    smaError("vgId:%d, fetch qtask files, open dir %s failed since %s", TD_VID(pVnode), dir, terrstr());
+    return TSDB_CODE_FAILED;
+  }
+
+  int32_t    dirLen = strlen(dir);
+  char      *dirEnd = POINTER_SHIFT(dir, dirLen);
+  regmatch_t regMatch[2];
+  while ((pDirEntry = taosReadDir(pDir))) {
+    char *entryName = taosGetDirEntryName(pDirEntry);
+    if (!entryName) {
+      continue;
+    }
+
+    code = regexec(&regex, entryName, 2, regMatch, 0);
+
+    if (code == 0) {
+      // match
+      smaInfo("vgId:%d, fetch qtask files, max ver:%" PRIi64 ", %s found", TD_VID(pVnode), version, entryName);
+
+      int64_t ver = -1;
+      sscanf((const char *)POINTER_SHIFT(entryName, regMatch[1].rm_so), "%" PRIi64, &ver);
+      if ((ver <= version) && (ver > -1)) {
+        if (!(*output)) {
+          if (!(*output = taosArrayInit(1, POINTER_BYTES))) {
+            terrno = TSDB_CODE_OUT_OF_MEMORY;
+            goto _end;
+          }
+        }
+        char *entryDup = strdup(entryName);
+        if (!entryDup) {
+          terrno = TSDB_CODE_OUT_OF_MEMORY;
+          goto _end;
+        }
+        if (!taosArrayPush(*output, &entryDup)) {
+          terrno = TSDB_CODE_OUT_OF_MEMORY;
+          goto _end;
+        }
+      } else {
+      }
+    } else if (code == REG_NOMATCH) {
+      // not match
+      smaTrace("vgId:%d, fetch qtask files, not match %s", TD_VID(pVnode), entryName);
+      continue;
+    } else {
+      // has other error
+      char errbuf[128];
+      regerror(code, &regex, errbuf, sizeof(errbuf));
+      smaWarn("vgId:%d, fetch qtask files, regexec failed since %s", TD_VID(pVnode), errbuf);
+      terrno = TSDB_CODE_RSMA_REGEX_MATCH;
+      goto _end;
+    }
+  }
+_end:
+  taosCloseDir(&pDir);
+  regfree(&regex);
+  return terrno == 0 ? TSDB_CODE_SUCCESS : TSDB_CODE_FAILED;
+}
+
+static int32_t tdQTaskFileCmprFn2(const void *p1, const void *p2) {
+  if (((SQTaskFile *)p1)->version < ((SQTaskFile *)p2)->version) {
+    return -1;
+  } else if (((SQTaskFile *)p1)->version > ((SQTaskFile *)p2)->version) {
+    return 1;
+  }
+
+  return 0;
+}
+
+int32_t tdRSmaFSUpsertQTaskFile(SRSmaFS *pFS, SQTaskFile *qTaskFile) {
+  int32_t code = 0;
+  int32_t idx = taosArraySearchIdx(pFS->aQTaskInf, qTaskFile, tdQTaskFileCmprFn2, TD_GE);
+
+  if (idx < 0) {
+    idx = taosArrayGetSize(pFS->aQTaskInf);
+  } else {
+    SQTaskFile *pTaskF = (SQTaskFile *)taosArrayGet(pFS->aQTaskInf, idx);
+    int32_t     c = tdQTaskFileCmprFn2(pTaskF, qTaskFile);
+    if (c == 0) {
+      pTaskF->nRef = qTaskFile->nRef;
+      pTaskF->version = qTaskFile->version;
+      pTaskF->size = qTaskFile->size;
+      goto _exit;
+    }
+  }
+
+  if (taosArrayInsert(pFS->aQTaskInf, idx, qTaskFile) == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
+
+_exit:
+  return code;
+}
\ No newline at end of file

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

@@ -150,7 +150,7 @@ int32_t smaOpen(SVnode *pVnode) {
     }
 
     // restore the rsma
-    if (tdRsmaRestore(pSma, RSMA_RESTORE_REBOOT, pVnode->state.committed) < 0) {
+    if (tdRSmaRestore(pSma, RSMA_RESTORE_REBOOT, pVnode->state.committed) < 0) {
       goto _err;
     }
   }
@@ -181,8 +181,8 @@ int32_t smaClose(SSma *pSma) {
  * @param committedVer
  * @return int32_t
  */
-int32_t tdRsmaRestore(SSma *pSma, int8_t type, int64_t committedVer) {
+int32_t tdRSmaRestore(SSma *pSma, int8_t type, int64_t committedVer) {
   ASSERT(VND_IS_RSMA(pSma->pVnode));
 
-  return tdProcessRSmaRestoreImpl(pSma, type, committedVer);
+  return tdRSmaProcessRestoreImpl(pSma, type, committedVer);
 }
\ No newline at end of file

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

@@ -15,11 +15,13 @@
 
 #include "sma.h"
 
-static int32_t rsmaSnapReadQTaskInfo(SRsmaSnapReader* pReader, uint8_t** ppData);
-static int32_t rsmaSnapWriteQTaskInfo(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
+static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppData);
+static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
+static int32_t rsmaQTaskInfSnapReaderOpen(SRSmaSnapReader* pReader, int64_t version);
+static int32_t rsmaQTaskInfSnapReaderClose(SQTaskFReader** ppReader);
 
-// SRsmaSnapReader ========================================
-struct SRsmaSnapReader {
+// SRSmaSnapReader ========================================
+struct SRSmaSnapReader {
   SSma*   pSma;
   int64_t sver;
   int64_t ever;
@@ -33,13 +35,13 @@ struct SRsmaSnapReader {
   SQTaskFReader* pQTaskFReader;
 };
 
-int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapReader** ppReader) {
+int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapReader** ppReader) {
   int32_t          code = 0;
   SVnode*          pVnode = pSma->pVnode;
-  SRsmaSnapReader* pReader = NULL;
+  SRSmaSnapReader* pReader = NULL;
 
   // alloc
-  pReader = (SRsmaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));
+  pReader = (SRSmaSnapReader*)taosMemoryCalloc(1, sizeof(*pReader));
   if (pReader == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
     goto _err;
@@ -48,7 +50,7 @@ int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapRead
   pReader->sver = sver;
   pReader->ever = ever;
 
-  // rsma1/rsma2
+  // open rsma1/rsma2
   for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
     if (pSma->pRSmaTsdb[i]) {
       code = tsdbSnapReaderOpen(pSma->pRSmaTsdb[i], sver, ever, i == 0 ? SNAP_DATA_RSMA1 : SNAP_DATA_RSMA2,
@@ -59,51 +61,112 @@ int32_t rsmaSnapReaderOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapRead
     }
   }
 
-  // qtaskinfo
-  // 1. add ref to qtaskinfo.v${ever} if exists and then start to replicate
+  // open qtaskinfo
+  if ((code = rsmaQTaskInfSnapReaderOpen(pReader, ever)) < 0) {
+    goto _err;
+  }
+
+  *ppReader = pReader;
+
+  return TSDB_CODE_SUCCESS;
+_err:
+  smaError("vgId:%d, vnode snapshot rsma reader open failed since %s", TD_VID(pVnode), tstrerror(code));
+  return TSDB_CODE_FAILED;
+}
+
+static int32_t rsmaQTaskInfSnapReaderOpen(SRSmaSnapReader* pReader, int64_t version) {
+  int32_t    code = 0;
+  SSma*      pSma = pReader->pSma;
+  SVnode*    pVnode = pSma->pVnode;
+  SSmaEnv*   pEnv = NULL;
+  SRSmaStat* pStat = NULL;
+
+  if (!(pEnv = SMA_RSMA_ENV(pSma))) {
+    smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as env is NULL",
+            TD_VID(pVnode), version);
+    return TSDB_CODE_SUCCESS;
+  }
+
+  pStat = (SRSmaStat*)SMA_ENV_STAT(pEnv);
+
+  int32_t ref = tdRSmaFSRef(pReader->pSma, pStat, version);
+  if (ref < 1) {
+    smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as ref is %d",
+            TD_VID(pVnode), version, ref);
+    return TSDB_CODE_SUCCESS;
+  }
+
   char qTaskInfoFullName[TSDB_FILENAME_LEN];
-  tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), ever, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
+  tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), version, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
 
   if (!taosCheckExistFile(qTaskInfoFullName)) {
-    smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo not need as %s not exists", TD_VID(pVnode),
-            qTaskInfoFullName);
-  } else {
+    tdRSmaFSUnRef(pSma, pStat, version);
+    smaInfo("vgId:%d, vnode snapshot rsma reader for qtaskinfo version %" PRIi64 " not need as %s not exists",
+            TD_VID(pVnode), qTaskInfoFullName);
+    return TSDB_CODE_SUCCESS;
+  }
+
   pReader->pQTaskFReader = taosMemoryCalloc(1, sizeof(SQTaskFReader));
   if (!pReader->pQTaskFReader) {
     code = TSDB_CODE_OUT_OF_MEMORY;
-      goto _err;
+    goto _end;
   }
 
-    TdFilePtr qTaskF = taosOpenFile(qTaskInfoFullName, TD_FILE_READ);
-    if (!qTaskF) {
+  TdFilePtr fp = taosOpenFile(qTaskInfoFullName, TD_FILE_READ);
+  if (!fp) {
     code = TAOS_SYSTEM_ERROR(errno);
-      goto _err;
+    taosMemoryFreeClear(pReader->pQTaskFReader);
+    goto _end;
   }
-    pReader->pQTaskFReader->pReadH = qTaskF;
-#if 0
-    SQTaskFile* pQTaskF = &pReader->pQTaskFReader->fTask;
-    pQTaskF->nRef = 1;
-#endif
+
+  pReader->pQTaskFReader->pReadH = fp;
+  pReader->pQTaskFReader->pSma = pSma;
+  pReader->pQTaskFReader->version = pReader->ever;
+
+_end:
+  if (code < 0) {
+    tdRSmaFSUnRef(pSma, pStat, version);
+    smaError("vgId:%d, vnode snapshot rsma reader open %s succeed", TD_VID(pVnode), qTaskInfoFullName);
+    return TSDB_CODE_FAILED;
   }
 
-  *ppReader = pReader;
-  smaInfo("vgId:%d, vnode snapshot rsma reader opened %s succeed", TD_VID(pVnode), qTaskInfoFullName);
+  smaInfo("vgId:%d, vnode snapshot rsma reader open %s succeed", TD_VID(pVnode), qTaskInfoFullName);
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t rsmaQTaskInfSnapReaderClose(SQTaskFReader** ppReader) {
+  if (!(*ppReader)) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SSma*      pSma = (*ppReader)->pSma;
+  SRSmaStat* pStat = SMA_RSMA_STAT(pSma);
+  int64_t    version = (*ppReader)->version;
+
+  taosCloseFile(&(*ppReader)->pReadH);
+  tdRSmaFSUnRef(pSma, pStat, version);
+  taosMemoryFreeClear(*ppReader);
+  smaInfo("vgId:%d, vnode snapshot rsma reader closed for qTaskInfo version %" PRIi64, SMA_VID(pSma), version);
+
   return TSDB_CODE_SUCCESS;
-_err:
-  smaError("vgId:%d, vnode snapshot rsma reader opened failed since %s", TD_VID(pVnode), tstrerror(code));
-  return TSDB_CODE_FAILED;
 }
 
-static int32_t rsmaSnapReadQTaskInfo(SRsmaSnapReader* pReader, uint8_t** ppBuf) {
+static int32_t rsmaSnapReadQTaskInfo(SRSmaSnapReader* pReader, uint8_t** ppBuf) {
   int32_t        code = 0;
   SSma*          pSma = pReader->pSma;
   int64_t        n = 0;
   uint8_t*       pBuf = NULL;
   SQTaskFReader* qReader = pReader->pQTaskFReader;
 
+  if (!qReader) {
+    *ppBuf = NULL;
+    smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, qTaskReader is NULL", SMA_VID(pSma));
+    return 0;
+  }
+
   if (!qReader->pReadH) {
     *ppBuf = NULL;
-    smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, readh is empty", SMA_VID(pSma));
+    smaInfo("vgId:%d, vnode snapshot rsma reader qtaskinfo, readh is NULL", SMA_VID(pSma));
     return 0;
   }
 
@@ -153,7 +216,7 @@ _err:
   return code;
 }
 
-int32_t rsmaSnapRead(SRsmaSnapReader* pReader, uint8_t** ppData) {
+int32_t rsmaSnapRead(SRSmaSnapReader* pReader, uint8_t** ppData) {
   int32_t code = 0;
 
   *ppData = NULL;
@@ -205,9 +268,9 @@ _err:
   return code;
 }
 
-int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader) {
+int32_t rsmaSnapReaderClose(SRSmaSnapReader** ppReader) {
   int32_t          code = 0;
-  SRsmaSnapReader* pReader = *ppReader;
+  SRSmaSnapReader* pReader = *ppReader;
 
   for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
     if (pReader->pDataReader[i]) {
@@ -215,11 +278,7 @@ int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader) {
     }
   }
 
-  if (pReader->pQTaskFReader) {
-    taosCloseFile(&pReader->pQTaskFReader->pReadH);
-    taosMemoryFreeClear(pReader->pQTaskFReader);
-    smaInfo("vgId:%d, vnode snapshot rsma reader closed for qTaskInfo", SMA_VID(pReader->pSma));
-  }
+  rsmaQTaskInfSnapReaderClose(&pReader->pQTaskFReader);
 
   smaInfo("vgId:%d, vnode snapshot rsma reader closed", SMA_VID(pReader->pSma));
 
@@ -227,8 +286,8 @@ int32_t rsmaSnapReaderClose(SRsmaSnapReader** ppReader) {
   return code;
 }
 
-// SRsmaSnapWriter ========================================
-struct SRsmaSnapWriter {
+// SRSmaSnapWriter ========================================
+struct SRSmaSnapWriter {
   SSma*   pSma;
   int64_t sver;
   int64_t ever;
@@ -244,13 +303,13 @@ struct SRsmaSnapWriter {
   SQTaskFWriter* pQTaskFWriter;
 };
 
-int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRsmaSnapWriter** ppWriter) {
+int32_t rsmaSnapWriterOpen(SSma* pSma, int64_t sver, int64_t ever, SRSmaSnapWriter** ppWriter) {
   int32_t          code = 0;
-  SRsmaSnapWriter* pWriter = NULL;
+  SRSmaSnapWriter* pWriter = NULL;
   SVnode*          pVnode = pSma->pVnode;
 
   // alloc
-  pWriter = (SRsmaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
+  pWriter = (SRSmaSnapWriter*)taosMemoryCalloc(1, sizeof(*pWriter));
   if (pWriter == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
     goto _err;
@@ -301,9 +360,9 @@ _err:
   return code;
 }
 
-int32_t rsmaSnapWriterClose(SRsmaSnapWriter** ppWriter, int8_t rollback) {
+int32_t rsmaSnapWriterClose(SRSmaSnapWriter** ppWriter, int8_t rollback) {
   int32_t          code = 0;
-  SRsmaSnapWriter* pWriter = *ppWriter;
+  SRSmaSnapWriter* pWriter = *ppWriter;
   SVnode*          pVnode = pWriter->pSma->pVnode;
 
   if (rollback) {
@@ -332,7 +391,7 @@ int32_t rsmaSnapWriterClose(SRsmaSnapWriter** ppWriter, int8_t rollback) {
               pWriter->pQTaskFWriter->fname, qTaskInfoFullName);
 
       // rsma restore
-      if ((code = tdRsmaRestore(pWriter->pSma, RSMA_RESTORE_SYNC, pWriter->ever)) < 0) {
+      if ((code = tdRSmaRestore(pWriter->pSma, RSMA_RESTORE_SYNC, pWriter->ever)) < 0) {
         goto _err;
       }
       smaInfo("vgId:%d, vnode snapshot rsma writer restore from %s succeed", SMA_VID(pWriter->pSma), qTaskInfoFullName);
@@ -349,7 +408,7 @@ _err:
   return code;
 }
 
-int32_t rsmaSnapWrite(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
+int32_t rsmaSnapWrite(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
   int32_t       code = 0;
   SSnapDataHdr* pHdr = (SSnapDataHdr*)pData;
 
@@ -377,7 +436,7 @@ _err:
   return code;
 }
 
-static int32_t rsmaSnapWriteQTaskInfo(SRsmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
+static int32_t rsmaSnapWriteQTaskInfo(SRSmaSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
   int32_t        code = 0;
   SQTaskFWriter* qWriter = pWriter->pQTaskFWriter;
 

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

@@ -20,6 +20,10 @@
 #define SMA_STORAGE_MINUTES_DAY  1440
 #define SMA_STORAGE_SPLIT_FACTOR 14400  // least records in tsma file
 
+static int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg);
+static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg);
+static int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
+
 // TODO: Who is responsible for resource allocate and release?
 int32_t tdProcessTSmaInsert(SSma *pSma, int64_t indexUid, const char *msg) {
   int32_t code = TSDB_CODE_SUCCESS;
@@ -59,7 +63,7 @@ int32_t smaGetTSmaDays(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *
  * @param days unit is minute
  * @return int32_t
  */
-int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days) {
+static int32_t tdProcessTSmaGetDaysImpl(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days) {
   SDecoder coder = {0};
   tDecoderInit(&coder, pCont, contLen);
 
@@ -106,7 +110,7 @@ _err:
  * @param pMsg
  * @return int32_t
  */
-int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
+static int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
   SSmaCfg *pCfg = (SSmaCfg *)pMsg;
 
   if (TD_VID(pSma->pVnode) == pCfg->dstVgId) {
@@ -145,7 +149,7 @@ int32_t tdProcessTSmaCreateImpl(SSma *pSma, int64_t version, const char *pMsg) {
  * @param msg
  * @return int32_t
  */
-int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
+static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char *msg) {
   const SArray *pDataBlocks = (const SArray *)msg;
   // TODO: destroy SSDataBlocks(msg)
   if (!pDataBlocks) {

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

@@ -306,92 +306,3 @@ int32_t tdReleaseSmaRef(int32_t rsetId, int64_t refId) {
 
   return TSDB_CODE_SUCCESS;
 }
-
-static int32_t tdCloneQTaskInfo(SSma *pSma, qTaskInfo_t dstTaskInfo, qTaskInfo_t srcTaskInfo, SRSmaParam *param,
-                                tb_uid_t suid, int8_t idx) {
-  SVnode *pVnode = pSma->pVnode;
-  char   *pOutput = NULL;
-  int32_t len = 0;
-
-  if ((terrno = qSerializeTaskStatus(srcTaskInfo, &pOutput, &len)) < 0) {
-    smaError("vgId:%d, rsma clone, table %" PRIi64 " serialize qTaskInfo failed since %s", TD_VID(pVnode), suid,
-             terrstr());
-    goto _err;
-  }
-
-  SReadHandle handle = {
-      .meta = pVnode->pMeta,
-      .vnode = pVnode,
-      .initTqReader = 1,
-  };
-  ASSERT(!dstTaskInfo);
-  dstTaskInfo = qCreateStreamExecTaskInfo(param->qmsg[idx], &handle);
-  if (!dstTaskInfo) {
-    terrno = TSDB_CODE_RSMA_QTASKINFO_CREATE;
-    goto _err;
-  }
-
-  if (qDeserializeTaskStatus(dstTaskInfo, pOutput, len) < 0) {
-    smaError("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " failed since %s", TD_VID(pVnode), suid,
-             terrstr());
-    goto _err;
-  }
-
-  smaDebug("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " succeed", TD_VID(pVnode), suid);
-
-  taosMemoryFreeClear(pOutput);
-  return TSDB_CODE_SUCCESS;
-_err:
-  taosMemoryFreeClear(pOutput);
-  tdFreeQTaskInfo(dstTaskInfo, TD_VID(pVnode), idx + 1);
-  smaError("vgId:%d, rsma clone, restore rsma task for table:%" PRIi64 " failed since %s", TD_VID(pVnode), suid,
-           terrstr());
-  return TSDB_CODE_FAILED;
-}
-
-/**
- * @brief Clone qTaskInfo of SRSmaInfo
- *
- * @param pSma
- * @param pInfo
- * @return int32_t
- */
-int32_t tdCloneRSmaInfo(SSma *pSma, SRSmaInfo *pInfo) {
-  SRSmaParam *param = NULL;
-  if (!pInfo) {
-    return TSDB_CODE_SUCCESS;
-  }
-
-  SMetaReader mr = {0};
-  metaReaderInit(&mr, SMA_META(pSma), 0);
-  smaDebug("vgId:%d, rsma clone qTaskInfo for suid:%" PRIi64, SMA_VID(pSma), pInfo->suid);
-  if (metaGetTableEntryByUid(&mr, pInfo->suid) < 0) {
-    smaError("vgId:%d, rsma clone, failed to get table meta for %" PRIi64 " since %s", SMA_VID(pSma), pInfo->suid,
-             terrstr());
-    goto _err;
-  }
-  ASSERT(mr.me.type == TSDB_SUPER_TABLE);
-  ASSERT(mr.me.uid == pInfo->suid);
-  if (TABLE_IS_ROLLUP(mr.me.flags)) {
-    param = &mr.me.stbEntry.rsmaParam;
-    for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
-      if (!pInfo->iTaskInfo[i]) {
-        continue;
-      }
-      if (tdCloneQTaskInfo(pSma, pInfo->taskInfo[i], pInfo->iTaskInfo[i], param, pInfo->suid, i) < 0) {
-        goto _err;
-      }
-    }
-    smaDebug("vgId:%d, rsma clone env success for %" PRIi64, SMA_VID(pSma), pInfo->suid);
-  } else {
-    terrno = TSDB_CODE_RSMA_INVALID_SCHEMA;
-    goto _err;
-  }
-
-  metaReaderClear(&mr);
-  return TSDB_CODE_SUCCESS;
-_err:
-  metaReaderClear(&mr);
-  smaError("vgId:%d, rsma clone env failed for %" PRIi64 " since %s", SMA_VID(pSma), pInfo->suid, terrstr());
-  return TSDB_CODE_FAILED;
-}
\ No newline at end of file

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

@@ -420,29 +420,15 @@ typedef enum {
 typedef struct {
   SFSLASTNEXTROWSTATES state;  // [input]
   STsdb               *pTsdb;  // [input]
-  SBlockIdx           *pBlockIdxExp;  // [input]
-  STSchema            *pTSchema;      // [input]
   tb_uid_t             suid;
   tb_uid_t             uid;
   int32_t              nFileSet;
   int32_t              iFileSet;
   SArray              *aDFileSet;
   SDataFReader        *pDataFReader;
-  SArray              *aBlockL;
-  SBlockL             *pBlockL;
-  SBlockData          *pBlockDataL;
-  SBlockData           blockDataL;
-  int32_t              nRow;
-  int32_t              iRow;
   TSDBROW              row;
-  /*
-  SArray    *aBlockIdx;
-  SBlockIdx *pBlockIdx;
-  SMapData   blockMap;
-  int32_t    nBlock;
-  int32_t    iBlock;
-  SBlock     block;
-  */
+
+  SMergeTree mergeTree;
 } SFSLastNextRowIter;
 
 static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
@@ -451,22 +437,16 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
 
   switch (state->state) {
     case SFSLASTNEXTROW_FS:
-      // state->aDFileSet = state->pTsdb->pFS->cState->aDFileSet;
       state->nFileSet = taosArrayGetSize(state->aDFileSet);
       state->iFileSet = state->nFileSet;
 
-      state->pBlockDataL = NULL;
-
     case SFSLASTNEXTROW_FILESET: {
       SDFileSet *pFileSet = NULL;
     _next_fileset:
       if (--state->iFileSet >= 0) {
         pFileSet = (SDFileSet *)taosArrayGet(state->aDFileSet, state->iFileSet);
       } else {
-        if (state->pBlockDataL) {
-          tBlockDataDestroy(state->pBlockDataL, 1);
-          state->pBlockDataL = NULL;
-        }
+        // tMergeTreeClose(&state->mergeTree);
 
         *ppRow = NULL;
         return code;
@@ -475,68 +455,24 @@ static int32_t getNextRowFromFSLast(void *iter, TSDBROW **ppRow) {
       code = tsdbDataFReaderOpen(&state->pDataFReader, state->pTsdb, pFileSet);
       if (code) goto _err;
 
-      if (!state->aBlockL) {
-        state->aBlockL = taosArrayInit(0, sizeof(SBlockL));
-      } else {
-        taosArrayClear(state->aBlockL);
-      }
-
-      code = tsdbReadBlockL(state->pDataFReader, state->aBlockL);
-      if (code) goto _err;
-
-      // SBlockL *pBlockL = (SBlockL *)taosArrayGet(state->aBlockL, state->iBlockL);
-
-      state->pBlockL = taosArraySearch(state->aBlockL, state->pBlockIdxExp, tCmprBlockL, TD_EQ);
-      if (!state->pBlockL) {
+      tMergeTreeOpen(&state->mergeTree, 1, state->pDataFReader, state->suid, state->uid,
+                     &(STimeWindow){.skey = TSKEY_MIN, .ekey = TSKEY_MAX},
+                     &(SVersionRange){.minVer = 0, .maxVer = UINT64_MAX});
+      bool hasVal = tMergeTreeNext(&state->mergeTree);
+      if (!hasVal) {
+        state->state = SFSLASTNEXTROW_FILESET;
+        // tMergeTreeClose(&state->mergeTree);
         goto _next_fileset;
       }
-
-      int64_t suid = state->pBlockL->suid;
-      int64_t uid = state->pBlockL->maxUid;
-
-      if (!state->pBlockDataL) {
-        state->pBlockDataL = &state->blockDataL;
-
-        tBlockDataCreate(state->pBlockDataL);
-      }
-      code = tBlockDataInit(state->pBlockDataL, suid, suid ? 0 : uid, state->pTSchema);
-      if (code) goto _err;
-    }
-    case SFSLASTNEXTROW_BLOCKDATA:
-      code = tsdbReadLastBlock(state->pDataFReader, state->pBlockL, state->pBlockDataL);
-      if (code) goto _err;
-
-      state->nRow = state->blockDataL.nRow;
-      state->iRow = state->nRow - 1;
-
-      if (!state->pBlockDataL->uid) {
-        while (state->pBlockIdxExp->uid != state->pBlockDataL->aUid[state->iRow]) {
-          --state->iRow;
-        }
-      }
-
       state->state = SFSLASTNEXTROW_BLOCKROW;
+    }
     case SFSLASTNEXTROW_BLOCKROW:
-      if (state->pBlockDataL->uid) {
-        if (state->iRow >= 0) {
-          state->row = tsdbRowFromBlockData(state->pBlockDataL, state->iRow);
+      state->row = tMergeTreeGetRow(&state->mergeTree);
       *ppRow = &state->row;
-
-          if (--state->iRow < 0) {
+      bool hasVal = tMergeTreeNext(&state->mergeTree);
+      if (!hasVal) {
         state->state = SFSLASTNEXTROW_FILESET;
       }
-        }
-      } else {
-        if (state->iRow >= 0 && state->pBlockIdxExp->uid == state->pBlockDataL->aUid[state->iRow]) {
-          state->row = tsdbRowFromBlockData(state->pBlockDataL, state->iRow);
-          *ppRow = &state->row;
-
-          if (--state->iRow < 0 || state->pBlockIdxExp->uid != state->pBlockDataL->aUid[state->iRow]) {
-            state->state = SFSLASTNEXTROW_FILESET;
-          }
-        }
-      }
-
       return code;
     default:
       ASSERT(0);
@@ -548,15 +484,6 @@ _err:
     tsdbDataFReaderClose(&state->pDataFReader);
     state->pDataFReader = NULL;
   }
-  if (state->aBlockL) {
-    taosArrayDestroy(state->aBlockL);
-    state->aBlockL = NULL;
-  }
-  if (state->pBlockDataL) {
-    tBlockDataDestroy(state->pBlockDataL, 1);
-    state->pBlockDataL = NULL;
-  }
-
   *ppRow = NULL;
 
   return code;
@@ -574,14 +501,6 @@ int32_t clearNextRowFromFSLast(void *iter) {
     tsdbDataFReaderClose(&state->pDataFReader);
     state->pDataFReader = NULL;
   }
-  if (state->aBlockL) {
-    taosArrayDestroy(state->aBlockL);
-    state->aBlockL = NULL;
-  }
-  if (state->pBlockDataL) {
-    tBlockDataDestroy(state->pBlockDataL, 1);
-    state->pBlockDataL = NULL;
-  }
 
   return code;
 }
@@ -609,7 +528,7 @@ typedef struct SFSNextRowIter {
   SMapData         blockMap;
   int32_t          nBlock;
   int32_t          iBlock;
-  SBlock           block;
+  SDataBlk         block;
   SBlockData       blockData;
   SBlockData      *pBlockData;
   int32_t          nRow;
@@ -684,13 +603,13 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow) {
     }
     case SFSNEXTROW_BLOCKDATA:
       if (state->iBlock >= 0) {
-        SBlock block = {0};
+        SDataBlk block = {0};
 
-        tBlockReset(&block);
+        tDataBlkReset(&block);
         // tBlockDataReset(&state->blockData);
         tBlockDataReset(state->pBlockData);
 
-        tMapDataGetItemByIdx(&state->blockMap, state->iBlock, &block, tGetBlock);
+        tMapDataGetItemByIdx(&state->blockMap, state->iBlock, &block, tGetDataBlk);
         /* code = tsdbReadBlockData(state->pDataFReader, &state->blockIdx, &block, &state->blockData, NULL, NULL); */
         tBlockDataReset(state->pBlockData);
         code = tBlockDataInit(state->pBlockData, state->suid, state->uid, state->pTSchema);
@@ -878,7 +797,7 @@ static bool tsdbKeyDeleted(TSDBKEY *key, SArray *pSkyline, int64_t *iSkyline) {
     if (key->ts > pItemBack->ts) {
       return false;
     } else if (key->ts >= pItemFront->ts && key->ts <= pItemBack->ts) {
-      if ((key->version <= pItemFront->version || key->ts == pItemBack->ts && key->version <= pItemBack->version)) {
+      if (key->version <= pItemFront->version || (key->ts == pItemBack->ts && key->version <= pItemBack->version)) {
         return true;
       } else {
         return false;
@@ -972,8 +891,6 @@ static int32_t nextRowIterOpen(CacheNextRowIter *pIter, tb_uid_t uid, STsdb *pTs
   pIter->fsLastState.state = (SFSLASTNEXTROWSTATES)SFSNEXTROW_FS;
   pIter->fsLastState.pTsdb = pTsdb;
   pIter->fsLastState.aDFileSet = pIter->pReadSnap->fs.aDFileSet;
-  pIter->fsLastState.pBlockIdxExp = &pIter->idx;
-  pIter->fsLastState.pTSchema = pTSchema;
   pIter->fsLastState.suid = suid;
   pIter->fsLastState.uid = uid;
 
@@ -1372,8 +1289,11 @@ int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUHand
   // getTableCacheKeyS(uid, "l", key, &keyLen);
   getTableCacheKey(uid, 1, key, &keyLen);
   LRUHandle *h = taosLRUCacheLookup(pCache, key, keyLen);
-  if (h) {
-  } else {
+  if (!h) {
+    taosThreadMutexLock(&pTsdb->lruMutex);
+
+    h = taosLRUCacheLookup(pCache, key, keyLen);
+    if (!h) {
       SArray *pLastArray = NULL;
       code = mergeLast(uid, pTsdb, &pLastArray);
       // if table's empty or error, return code of -1
@@ -1384,13 +1304,18 @@ int32_t tsdbCacheGetLastH(SLRUCache *pCache, tb_uid_t uid, STsdb *pTsdb, LRUHand
       }
 
       _taos_lru_deleter_t deleter = deleteTableCacheLast;
-    LRUStatus           status =
-        taosLRUCacheInsert(pCache, key, keyLen, pLastArray, pLastArray->capacity, deleter, NULL, TAOS_LRU_PRIORITY_LOW);
+      LRUStatus status = taosLRUCacheInsert(pCache, key, keyLen, pLastArray, pLastArray->capacity, deleter, NULL,
+                                            TAOS_LRU_PRIORITY_LOW);
       if (status != TAOS_LRU_STATUS_OK) {
         code = -1;
       }
 
+      taosThreadMutexUnlock(&pTsdb->lruMutex);
+
       h = taosLRUCacheLookup(pCache, key, keyLen);
+    } else {
+      taosThreadMutexUnlock(&pTsdb->lruMutex);
+    }
   }
 
   *handle = h;
@@ -1411,3 +1336,5 @@ void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) {
 }
 
 size_t tsdbCacheGetCapacity(SVnode *pVnode) { return taosLRUCacheGetCapacity(pVnode->pTsdb->lruCache); }
+
+size_t tsdbCacheGetUsage(SVnode *pVnode) { return taosLRUCacheGetUsage(pVnode->pTsdb->lruCache); }

source/dnode/vnode/src/tsdb/tsdbCompact.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 "tsdb.h"
+
+typedef struct {
+  STsdb  *pTsdb;
+  STsdbFS fs;
+} STsdbCompactor;
+
+int32_t tsdbCompact(STsdb *pTsdb) {
+  int32_t code = 0;
+  // TODO
+  return code;
+}

source/dnode/vnode/src/tsdb/tsdbCompress.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 "tsdb.h"
+
+// Integer =====================================================
+typedef struct {
+  int8_t   rawCopy;
+  int64_t  prevVal;
+  int32_t  nVal;
+  int32_t  nBuf;
+  uint8_t *pBuf;
+} SIntCompressor;
+
+#define I64_SAFE_ADD(a, b) (((a) >= 0 && (b) <= INT64_MAX - (b)) || ((a) < 0 && (b) >= INT64_MIN - (a)))
+#define SIMPLE8B_MAX       ((uint64_t)1152921504606846974LL)
+
+static int32_t tsdbCmprI64(SIntCompressor *pCompressor, int64_t val) {
+  int32_t code = 0;
+
+  // raw copy
+  if (pCompressor->rawCopy) {
+    memcpy(pCompressor->pBuf + pCompressor->nBuf, &val, sizeof(val));
+    pCompressor->nBuf += sizeof(val);
+    pCompressor->nVal++;
+    goto _exit;
+  }
+
+  if (!I64_SAFE_ADD(val, pCompressor->prevVal)) {
+    pCompressor->rawCopy = 1;
+    // TODO: decompress and copy
+    pCompressor->nVal++;
+    goto _exit;
+  }
+
+  int64_t diff = val - pCompressor->prevVal;
+  uint8_t zigzag = ZIGZAGE(int64_t, diff);
+
+  if (zigzag >= SIMPLE8B_MAX) {
+    pCompressor->rawCopy = 1;
+    // TODO: decompress and copy
+    pCompressor->nVal++;
+    goto _exit;
+  }
+
+_exit:
+  return code;
+}
+
+// Timestamp =====================================================
+
+// Float =====================================================
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbDiskData.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 "tsdb.h"
+
+typedef struct SDiskColBuilder SDiskColBuilder;
+struct SDiskColBuilder {
+  uint8_t  flags;
+  uint8_t *pBitMap;
+  int32_t *aOffset;
+  int32_t  nData;
+  uint8_t *pData;
+};
+
+int32_t tDiskColAddVal(SDiskColBuilder *pBuilder, SColVal *pColVal) {
+  int32_t code = 0;
+  // TODO
+  return code;
+}
+
+// ================================================================
+typedef struct SDiskDataBuilder SDiskDataBuilder;
+struct SDiskDataBuilder {
+  SDiskDataHdr hdr;
+  SArray      *aBlockCol;  // SArray<SBlockCol>
+};
+
+int32_t tDiskDataBuilderCreate(SDiskDataBuilder **ppBuilder) {
+  int32_t code = 0;
+  // TODO
+  return code;
+}
+
+void tDiskDataBuilderDestroy(SDiskDataBuilder *pBuilder) {
+  // TODO
+}
+
+void tDiskDataBuilderInit(SDiskDataBuilder *pBuilder, int64_t suid, int64_t uid, STSchema *pTSchema, int8_t cmprAlg) {
+  pBuilder->hdr = (SDiskDataHdr){.delimiter = TSDB_FILE_DLMT,  //
+                                 .fmtVer = 0,
+                                 .suid = suid,
+                                 .uid = uid,
+                                 .cmprAlg = cmprAlg};
+}
+
+void tDiskDataBuilderReset(SDiskDataBuilder *pBuilder) {
+  // TODO
+}
+
+int32_t tDiskDataBuilderAddRow(SDiskDataBuilder *pBuilder, TSDBROW *pRow, STSchema *pTSchema, int64_t uid) {
+  int32_t code = 0;
+
+  // uid (todo)
+
+  // version (todo)
+
+  // TSKEY (todo)
+
+  SRowIter iter = {0};
+  tRowIterInit(&iter, pRow, pTSchema);
+
+  for (int32_t iDiskCol = 0; iDiskCol < 0; iDiskCol++) {
+  }
+
+  return code;
+}
+
+int32_t tDiskDataBuilderGet(SDiskDataBuilder *pBuilder, uint8_t **ppData) {
+  int32_t code = 0;
+  // TODO
+  return code;
+}
\ No newline at end of file

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

@@ -53,22 +53,22 @@ static int32_t tGetDataFile(uint8_t *p, SDataFile *pDataFile) {
   return n;
 }
 
-int32_t tPutLastFile(uint8_t *p, SLastFile *pLastFile) {
+int32_t tPutSttFile(uint8_t *p, SSttFile *pSttFile) {
   int32_t n = 0;
 
-  n += tPutI64v(p ? p + n : p, pLastFile->commitID);
-  n += tPutI64v(p ? p + n : p, pLastFile->size);
-  n += tPutI64v(p ? p + n : p, pLastFile->offset);
+  n += tPutI64v(p ? p + n : p, pSttFile->commitID);
+  n += tPutI64v(p ? p + n : p, pSttFile->size);
+  n += tPutI64v(p ? p + n : p, pSttFile->offset);
 
   return n;
 }
 
-static int32_t tGetLastFile(uint8_t *p, SLastFile *pLastFile) {
+static int32_t tGetSttFile(uint8_t *p, SSttFile *pSttFile) {
   int32_t n = 0;
 
-  n += tGetI64v(p + n, &pLastFile->commitID);
-  n += tGetI64v(p + n, &pLastFile->size);
-  n += tGetI64v(p + n, &pLastFile->offset);
+  n += tGetI64v(p + n, &pSttFile->commitID);
+  n += tGetI64v(p + n, &pSttFile->size);
+  n += tGetI64v(p + n, &pSttFile->offset);
 
   return n;
 }
@@ -102,9 +102,9 @@ void tsdbDataFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SDataFile *pDataF,
            TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), fid, pDataF->commitID, ".data");
 }
 
-void tsdbLastFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SLastFile *pLastF, char fname[]) {
+void tsdbSttFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSttFile *pSttF, char fname[]) {
   snprintf(fname, TSDB_FILENAME_LEN - 1, "%s%s%s%sv%df%dver%" PRId64 "%s", tfsGetDiskPath(pTsdb->pVnode->pTfs, did),
-           TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), fid, pLastF->commitID, ".last");
+           TD_DIRSEP, pTsdb->path, TD_DIRSEP, TD_VID(pTsdb->pVnode), fid, pSttF->commitID, ".stt");
 }
 
 void tsdbSmaFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSmaFile *pSmaF, char fname[]) {
@@ -148,7 +148,7 @@ int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype) {
   }
 
   // ftruncate
-  if (taosFtruncateFile(pFD, size) < 0) {
+  if (taosFtruncateFile(pFD, LOGIC_TO_FILE_SIZE(size, TSDB_DEFAULT_PAGE_SIZE)) < 0) {
     code = TAOS_SYSTEM_ERROR(errno);
     goto _err;
   }
@@ -194,9 +194,11 @@ int32_t tPutDFileSet(uint8_t *p, SDFileSet *pSet) {
   n += tPutDataFile(p ? p + n : p, pSet->pDataF);
   n += tPutSmaFile(p ? p + n : p, pSet->pSmaF);
 
-  // last
-  n += tPutU8(p ? p + n : p, 1);  // for future compatibility
-  n += tPutLastFile(p ? p + n : p, pSet->pLastF);
+  // stt
+  n += tPutU8(p ? p + n : p, pSet->nSttF);
+  for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
+    n += tPutSttFile(p ? p + n : p, pSet->aSttF[iStt]);
+  }
 
   return n;
 }
@@ -208,15 +210,40 @@ int32_t tGetDFileSet(uint8_t *p, SDFileSet *pSet) {
   n += tGetI32v(p + n, &pSet->diskId.id);
   n += tGetI32v(p + n, &pSet->fid);
 
-  // data
+  // head
+  pSet->pHeadF = (SHeadFile *)taosMemoryCalloc(1, sizeof(SHeadFile));
+  if (pSet->pHeadF == NULL) {
+    return -1;
+  }
+  pSet->pHeadF->nRef = 1;
   n += tGetHeadFile(p + n, pSet->pHeadF);
+
+  // data
+  pSet->pDataF = (SDataFile *)taosMemoryCalloc(1, sizeof(SDataFile));
+  if (pSet->pDataF == NULL) {
+    return -1;
+  }
+  pSet->pDataF->nRef = 1;
   n += tGetDataFile(p + n, pSet->pDataF);
+
+  // sma
+  pSet->pSmaF = (SSmaFile *)taosMemoryCalloc(1, sizeof(SSmaFile));
+  if (pSet->pSmaF == NULL) {
+    return -1;
+  }
+  pSet->pSmaF->nRef = 1;
   n += tGetSmaFile(p + n, pSet->pSmaF);
 
-  // last
-  uint8_t nLast;
-  n += tGetU8(p + n, &nLast);
-  n += tGetLastFile(p + n, pSet->pLastF);
+  // stt
+  n += tGetU8(p + n, &pSet->nSttF);
+  for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
+    pSet->aSttF[iStt] = (SSttFile *)taosMemoryCalloc(1, sizeof(SSttFile));
+    if (pSet->aSttF[iStt] == NULL) {
+      return -1;
+    }
+    pSet->aSttF[iStt]->nRef = 1;
+    n += tGetSttFile(p + n, pSet->aSttF[iStt]);
+  }
 
   return n;
 }

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

@@ -60,7 +60,7 @@ int32_t tsdbDoRetention(STsdb *pTsdb, int64_t now) {
     if (expLevel < 0) {
       taosMemoryFree(pSet->pHeadF);
       taosMemoryFree(pSet->pDataF);
-      taosMemoryFree(pSet->pLastF);
+      taosMemoryFree(pSet->aSttF[0]);
       taosMemoryFree(pSet->pSmaF);
       taosArrayRemove(fs.aDFileSet, iSet);
       iSet--;

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

@@ -39,7 +39,7 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *
     SSubmitBlkRsp r = {0};
     tGetSubmitMsgNext(&msgIter, &pBlock);
     if (pBlock == NULL) break;
-    if (tsdbInsertTableData(pTsdb, version, &msgIter, pBlock, &r) < 0) {
+    if ((terrno = tsdbInsertTableData(pTsdb, version, &msgIter, pBlock, &r)) < 0) {
       return -1;
     }
 

source/dnode/vnode/src/vnd/vnodeQuery.c

@@ -368,6 +368,7 @@ _exit:
 int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
   pLoad->vgId = TD_VID(pVnode);
   pLoad->syncState = syncGetMyRole(pVnode->sync);
+  pLoad->cacheUsage = tsdbCacheGetUsage(pVnode);
   pLoad->numOfTables = metaGetTbNum(pVnode->pMeta);
   pLoad->numOfTimeSeries = metaGetTimeSeriesNum(pVnode->pMeta);
   pLoad->totalStorage = (int64_t)3 * 1073741824;

source/dnode/vnode/src/vnd/vnodeSnapshot.c

@@ -39,7 +39,7 @@ struct SVSnapReader {
   SStreamStateReader *pStreamStateReader;
   // rsma
   int8_t           rsmaDone;
-  SRsmaSnapReader *pRsmaReader;
+  SRSmaSnapReader *pRsmaReader;
 };
 
 int32_t vnodeSnapReaderOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapReader **ppReader) {
@@ -241,7 +241,7 @@ struct SVSnapWriter {
   SStreamTaskWriter  *pStreamTaskWriter;
   SStreamStateWriter *pStreamStateWriter;
   // rsma
-  SRsmaSnapWriter *pRsmaSnapWriter;
+  SRSmaSnapWriter *pRsmaSnapWriter;
 };
 
 int32_t vnodeSnapWriterOpen(SVnode *pVnode, int64_t sver, int64_t ever, SVSnapWriter **ppWriter) {

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -534,7 +534,7 @@ static int32_t vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pR
   }
 
   tqUpdateTbUidList(pVnode->pTq, tbUids, true);
-  if (tdUpdateTbUidList(pVnode->pSma, pStore) < 0) {
+  if (tdUpdateTbUidList(pVnode->pSma, pStore, true) < 0) {
     goto _exit;
   }
   tdUidStoreFree(pStore);
@@ -692,6 +692,7 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq
   SEncoder         encoder = {0};
   int32_t          ret;
   SArray          *tbUids = NULL;
+  STbUidStore     *pStore = NULL;
 
   pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
   pRsp->pCont = NULL;
@@ -715,9 +716,10 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq
   for (int32_t iReq = 0; iReq < req.nReqs; iReq++) {
     SVDropTbReq *pDropTbReq = req.pReqs + iReq;
     SVDropTbRsp  dropTbRsp = {0};
+    tb_uid_t     tbUid = 0;
 
     /* code */
-    ret = metaDropTable(pVnode->pMeta, version, pDropTbReq, tbUids);
+    ret = metaDropTable(pVnode->pMeta, version, pDropTbReq, tbUids, &tbUid);
     if (ret < 0) {
       if (pDropTbReq->igNotExists && terrno == TSDB_CODE_VND_TABLE_NOT_EXIST) {
         dropTbRsp.code = TSDB_CODE_SUCCESS;
@@ -726,15 +728,18 @@ static int32_t vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq
       }
     } else {
       dropTbRsp.code = TSDB_CODE_SUCCESS;
+      if (tbUid > 0) tdFetchTbUidList(pVnode->pSma, &pStore, pDropTbReq->suid, tbUid);
     }
 
     taosArrayPush(rsp.pArray, &dropTbRsp);
   }
 
   tqUpdateTbUidList(pVnode->pTq, tbUids, false);
+  tdUpdateTbUidList(pVnode->pSma, pStore, false);
 
 _exit:
   taosArrayDestroy(tbUids);
+  tdUidStoreFree(pStore);
   tDecoderClear(&decoder);
   tEncodeSize(tEncodeSVDropTbBatchRsp, &rsp, pRsp->contLen, ret);
   pRsp->pCont = rpcMallocCont(pRsp->contLen);

source/libs/catalog/src/ctgDbg.c

@@ -367,18 +367,22 @@ void ctgdShowDBCache(SCatalog* pCtg, SHashObj *dbHash) {
     int32_t stbNum = dbCache->stbCache ? taosHashGetSize(dbCache->stbCache) : 0;
     int32_t vgVersion = CTG_DEFAULT_INVALID_VERSION;
     int32_t hashMethod = -1;
+    int16_t hashPrefix = 0;
+    int16_t hashSuffix = 0;
     int32_t vgNum = 0;
 
     if (dbCache->vgCache.vgInfo) {
       vgVersion = dbCache->vgCache.vgInfo->vgVersion;
       hashMethod = dbCache->vgCache.vgInfo->hashMethod;
+      hashPrefix = dbCache->vgCache.vgInfo->hashPrefix;
+      hashSuffix = dbCache->vgCache.vgInfo->hashSuffix;
       if (dbCache->vgCache.vgInfo->vgHash) {
         vgNum = taosHashGetSize(dbCache->vgCache.vgInfo->vgHash);
       }
     }
     
-    ctgDebug("[%d] db [%.*s][0x%"PRIx64"] %s: metaNum:%d, stbNum:%d, vgVersion:%d, hashMethod:%d, vgNum:%d",
-      i, (int32_t)len, dbFName, dbCache->dbId, dbCache->deleted?"deleted":"", metaNum, stbNum, vgVersion, hashMethod, vgNum);
+    ctgDebug("[%d] db [%.*s][0x%"PRIx64"] %s: metaNum:%d, stbNum:%d, vgVersion:%d, hashMethod:%d, prefix:%d, suffix:%d, vgNum:%d",
+      i, (int32_t)len, dbFName, dbCache->dbId, dbCache->deleted?"deleted":"", metaNum, stbNum, vgVersion, hashMethod, hashPrefix, hashSuffix, vgNum);
 
     pIter = taosHashIterate(dbHash, pIter);
   }

source/libs/catalog/src/ctgUtil.c

@@ -848,15 +848,11 @@ int32_t ctgGetVgInfoFromHashValue(SCatalog *pCtg, SDBVgInfo *dbInfo, const SName
     CTG_ERR_RET(TSDB_CODE_TSC_DB_NOT_SELECTED);
   }
 
-  tableNameHashFp fp = NULL;
   SVgroupInfo *vgInfo = NULL;
-
-  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
-
   char tbFullName[TSDB_TABLE_FNAME_LEN];
   tNameExtractFullName(pTableName, tbFullName);
 
-  uint32_t hashValue = (*fp)(tbFullName, (uint32_t)strlen(tbFullName));
+  uint32_t hashValue = taosGetTbHashVal(tbFullName, (uint32_t)strlen(tbFullName), dbInfo->hashMethod, dbInfo->hashPrefix, dbInfo->hashSuffix);
 
   void *pIter = taosHashIterate(dbInfo->vgHash, NULL);
   while (pIter) {
@@ -919,11 +915,7 @@ int32_t ctgGetVgInfosFromHashValue(SCatalog *pCtg, SCtgTaskReq* tReq, SDBVgInfo
     CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
   }
 
-  tableNameHashFp fp = NULL;
   SVgroupInfo *vgInfo = NULL;
-
-  CTG_ERR_RET(ctgGetHashFunction(dbInfo->hashMethod, &fp));
-
   int32_t tbNum = taosArrayGetSize(pNames);
 
   if (1 == vgNum) {
@@ -975,7 +967,7 @@ int32_t ctgGetVgInfosFromHashValue(SCatalog *pCtg, SCtgTaskReq* tReq, SDBVgInfo
     tbNameLen = offset + strlen(pName->tname);
     strcpy(tbFullName + offset, pName->tname);
 
-    uint32_t hashValue = (*fp)(tbFullName, (uint32_t)tbNameLen);
+    uint32_t hashValue = taosGetTbHashVal(tbFullName, (uint32_t)strlen(tbFullName), dbInfo->hashMethod, dbInfo->hashPrefix, dbInfo->hashSuffix);
 
     SVgroupInfo **p = taosArraySearch(pVgList, &hashValue, ctgHashValueComp, TD_EQ);
 

source/libs/catalog/test/catalogTests.cpp

@@ -218,6 +218,8 @@ void ctgTestBuildDBVgroup(SDBVgInfo **pdbVgroup) {
   ctgTestCurrentVgVersion = dbVgroup->vgVersion;
 
   dbVgroup->hashMethod = 0;
+  dbVgroup->hashPrefix = 0;
+  dbVgroup->hashSuffix = 0;
   dbVgroup->vgHash = taosHashInit(ctgTestVgNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
 
   vgNum = ctgTestGetVgNumFromVgVersion(dbVgroup->vgVersion);

source/libs/executor/inc/executil.h

@@ -87,8 +87,6 @@ struct SqlFunctionCtx;
 
 size_t getResultRowSize(struct SqlFunctionCtx* pCtx, int32_t numOfOutput);
 void   initResultRowInfo(SResultRowInfo* pResultRowInfo);
-
-void initResultRow(SResultRow* pResultRow);
 void   closeResultRow(SResultRow* pResultRow);
 
 struct SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset);

source/libs/executor/src/dataDeleter.c

@@ -168,7 +168,9 @@ static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, bool* pQueryE
   taosReadQitem(pDeleter->pDataBlocks, (void**)&pBuf);
   memcpy(&pDeleter->nextOutput, pBuf, sizeof(SDataDeleterBuf));
   taosFreeQitem(pBuf);
-  *pLen = ((SDataCacheEntry*)(pDeleter->nextOutput.pData))->dataLen;
+
+  SDataCacheEntry* pEntry = (SDataCacheEntry*)pDeleter->nextOutput.pData;
+  *pLen = pEntry->dataLen;
   *pQueryEnd = pDeleter->queryEnd;
   qDebug("got data len %" PRId64 ", row num %d in sink", *pLen, ((SDataCacheEntry*)(pDeleter->nextOutput.pData))->numOfRows);
 }

source/libs/executor/src/executil.c

@@ -1302,7 +1302,6 @@ int32_t initQueryTableDataCond(SQueryTableDataCond* pCond, const STableScanPhysi
   pCond->type = TIMEWINDOW_RANGE_CONTAINED;
   pCond->startVersion = -1;
   pCond->endVersion = -1;
-  pCond->schemaVersion = -1;
   //  pCond->type = pTableScanNode->scanFlag;
 
   int32_t j = 0;

source/libs/executor/src/scanoperator.c

@@ -695,6 +695,7 @@ static void destroyTableScanOperatorInfo(void* param) {
   cleanupQueryTableDataCond(&pTableScanInfo->cond);
 
   tsdbReaderClose(pTableScanInfo->dataReader);
+  pTableScanInfo->dataReader = NULL;
 
   if (pTableScanInfo->pColMatchInfo != NULL) {
     taosArrayDestroy(pTableScanInfo->pColMatchInfo);
@@ -2648,7 +2649,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
       }
 
       char* pStart = pRsp->data;
-      extractDataBlockFromFetchRsp(pInfo->pRes, pRsp->data, pOperator->exprSupp.numOfExprs, pInfo->scanCols, &pStart);
+      extractDataBlockFromFetchRsp(pInfo->pRes, pRsp->data, pInfo->scanCols, &pStart);
       updateLoadRemoteInfo(&pInfo->loadInfo, pRsp->numOfRows, pRsp->compLen, startTs, pOperator);
 
       // todo log the filter info

source/util/src/terror.c

@@ -618,6 +618,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_REMOVE_EXISTS,             "Rsma remove exists"
 TAOS_DEFINE_ERROR(TSDB_CODE_RSMA_FETCH_MSG_MSSED_UP,        "Rsma fetch msg is messed up")
 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")
 
 //index
 TAOS_DEFINE_ERROR(TSDB_CODE_INDEX_REBUILDING,               "Index is rebuilding")