Merge pull request #16600 from taosdata/refact/tsdb_optimize

refactor: refactor the last file format, along with the correlated write/read processing.

include/common/tmsg.h

@@ -1070,6 +1070,7 @@ typedef struct {
 typedef struct {
   int32_t vgId;
   int32_t syncState;
+  int64_t cacheUsage;
   int64_t numOfTables;
   int64_t numOfTimeSeries;
   int64_t totalStorage;

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/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/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;

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/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

@@ -49,6 +49,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/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 SSstBlk       SSstBlk;
 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 SSstFile      SSstFile;
 typedef struct SSmaFile      SSmaFile;
 typedef struct SDFileSet     SDFileSet;
 typedef struct SDataFWriter  SDataFWriter;
@@ -64,9 +64,12 @@ typedef struct STsdbReadSnap STsdbReadSnap;
 typedef struct SBlockInfo    SBlockInfo;
 typedef struct SSmaInfo      SSmaInfo;
 typedef struct SBlockCol     SBlockCol;
+typedef struct SVersionRange SVersionRange;
 
 #define TSDB_FILE_DLMT         ((uint32_t)0xF00AFA0F)
 #define TSDB_MAX_SUBBLOCKS     8
+#define TSDB_MAX_LAST_FILE     16
+#define TSDB_DEFAULT_LAST_FILE 8
 #define TSDB_FHDR_SIZE         512
 
 #define HAS_NONE  ((int8_t)0x1)
@@ -111,15 +114,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);
+// SSstBlk
+int32_t tPutSstBlk(uint8_t *p, void *ph);
+int32_t tGetSstBlk(uint8_t *p, void *ph);
 // SBlockIdx
 int32_t tPutBlockIdx(uint8_t *p, void *ph);
 int32_t tGetBlockIdx(uint8_t *p, void *ph);
@@ -170,6 +173,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);
@@ -215,7 +219,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 tPutSstFile(uint8_t *p, SSstFile *pSstFile);
 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 +228,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 tsdbSstFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSstFile *pSstF, 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 +254,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 tsdbWriteSstBlk(SDataFWriter *pWriter, SArray *aSstBlk);
 int32_t tsdbWriteBlockData(SDataFWriter *pWriter, SBlockData *pBlockData, SBlockInfo *pBlkInfo, SSmaInfo *pSmaInfo,
                            int8_t cmprAlg, int8_t toLast);
 
@@ -260,10 +264,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 tsdbReadSstBlk(SDataFReader *pReader, int32_t iSst, SArray *aSstBlk);
+int32_t tsdbReadBlockSma(SDataFReader *pReader, SDataBlk *pBlock, SArray *aColumnDataAgg);
+int32_t tsdbReadDataBlock(SDataFReader *pReader, SDataBlk *pBlock, SBlockData *pBlockData);
+int32_t tsdbReadSstBlock(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData);
 // SDelFWriter
 int32_t tsdbDelFWriterOpen(SDelFWriter **ppWriter, SDelFile *pFile, STsdb *pTsdb);
 int32_t tsdbDelFWriterClose(SDelFWriter **ppWriter, int8_t sync);
@@ -278,6 +282,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 +330,11 @@ struct TSDBKEY {
   TSKEY   ts;
 };
 
+struct SVersionRange {
+  uint64_t minVer;
+  uint64_t maxVer;
+};
+
 typedef struct SMemSkipListNode SMemSkipListNode;
 struct SMemSkipListNode {
   int8_t            level;
@@ -416,7 +427,7 @@ struct SSmaInfo {
   int32_t size;
 };
 
-struct SBlock {
+struct SDataBlk {
   TSDBKEY    minKey;
   TSDBKEY    maxKey;
   int64_t    minVer;
@@ -428,7 +439,7 @@ struct SBlock {
   SSmaInfo   smaInfo;
 };
 
-struct SBlockL {
+struct SSstBlk {
   int64_t    suid;
   int64_t    minUid;
   int64_t    maxUid;
@@ -467,12 +478,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 +541,7 @@ struct SDataFile {
   int64_t size;
 };
 
-struct SLastFile {
+struct SSstFile {
   volatile int32_t nRef;
 
   int64_t commitID;
@@ -556,8 +561,9 @@ struct SDFileSet {
   int32_t    fid;
   SHeadFile *pHeadF;
   SDataFile *pDataF;
-  SLastFile *pLastF;
   SSmaFile  *pSmaF;
+  uint8_t    nSstF;
+  SSstFile  *aSstF[TSDB_MAX_LAST_FILE];
 };
 
 struct SRowIter {
@@ -586,13 +592,13 @@ struct SDataFWriter {
 
   TdFilePtr pHeadFD;
   TdFilePtr pDataFD;
-  TdFilePtr pLastFD;
   TdFilePtr pSmaFD;
+  TdFilePtr pLastFD;
 
   SHeadFile fHead;
   SDataFile fData;
-  SLastFile fLast;
   SSmaFile  fSma;
+  SSstFile  fSst[TSDB_MAX_LAST_FILE];
 
   uint8_t *aBuf[4];
 };
@@ -603,6 +609,36 @@ struct STsdbReadSnap {
   STsdbFS    fs;
 };
 
+struct SDataFReader {
+  STsdb     *pTsdb;
+  SDFileSet *pSet;
+  TdFilePtr  pHeadFD;
+  TdFilePtr  pDataFD;
+  TdFilePtr  pSmaFD;
+  TdFilePtr  aLastFD[TSDB_MAX_LAST_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;
+  struct SLDataIter *pIter;
+} SMergeTree;
+
+int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader* pFReader, uint64_t uid, STimeWindow* pTimeWindow, SVersionRange* pVerRange);
+void tMergeTreeAddIter(SMergeTree *pMTree, struct 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"

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

@@ -420,29 +420,14 @@ 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 +436,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 +454,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->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);
-          *ppRow = &state->row;
-
-          if (--state->iRow < 0) {
-            state->state = SFSLASTNEXTROW_FILESET;
     }
-        }
-      } else {
-        if (state->iRow >= 0 && state->pBlockIdxExp->uid == state->pBlockDataL->aUid[state->iRow]) {
-          state->row = tsdbRowFromBlockData(state->pBlockDataL, state->iRow);
+    case SFSLASTNEXTROW_BLOCKROW:
+      state->row = tMergeTreeGetRow(&state->mergeTree);
       *ppRow = &state->row;
-
-          if (--state->iRow < 0 || state->pBlockIdxExp->uid != state->pBlockDataL->aUid[state->iRow]) {
+      bool hasVal = tMergeTreeNext(&state->mergeTree);
+      if (!hasVal) {
         state->state = SFSLASTNEXTROW_FILESET;
       }
-        }
-      }
-
       return code;
     default:
       ASSERT(0);
@@ -548,15 +483,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 +500,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 +527,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 +602,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);
@@ -972,9 +890,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;
 
   pIter->fsState.state = SFSNEXTROW_FS;
@@ -1372,8 +1287,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 +1302,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 +1334,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/tsdbFS.c

@@ -110,7 +110,7 @@ _err:
 //       taosRemoveFile(fname);
 //     }
 
-//     // last
+//     // sst
 //     if (isSameDisk && pFrom->pLastF->commitID == pTo->pLastF->commitID) {
 //       if (pFrom->pLastF->size > pTo->pLastF->size) {
 //         code = tsdbDFileRollback(pFS->pTsdb, pTo, TSDB_LAST_FILE);
@@ -140,7 +140,7 @@ _err:
 //     tsdbDataFileName(pFS->pTsdb, pFrom->diskId, pFrom->fid, pFrom->pDataF, fname);
 //     taosRemoveFile(fname);
 
-//     // last
+//     // sst
 //     tsdbLastFileName(pFS->pTsdb, pFrom->diskId, pFrom->fid, pFrom->pLastF, fname);
 //     taosRemoveFile(fname);
 
@@ -254,8 +254,10 @@ void tsdbFSDestroy(STsdbFS *pFS) {
     SDFileSet *pSet = (SDFileSet *)taosArrayGet(pFS->aDFileSet, iSet);
     taosMemoryFree(pSet->pHeadF);
     taosMemoryFree(pSet->pDataF);
-    taosMemoryFree(pSet->pLastF);
     taosMemoryFree(pSet->pSmaF);
+    for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+      taosMemoryFree(pSet->aSstF[iSst]);
+    }
   }
 
   taosArrayDestroy(pFS->aDFileSet);
@@ -309,29 +311,31 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
       if (code) goto _err;
     }
 
-    // last ===========
-    tsdbLastFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pLastF, fname);
+    // sma =============
+    tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
     if (taosStatFile(fname, &size, NULL)) {
       code = TAOS_SYSTEM_ERROR(errno);
       goto _err;
     }
-    if (size != pSet->pLastF->size) {
+    if (size < pSet->pSmaF->size) {
       code = TSDB_CODE_FILE_CORRUPTED;
       goto _err;
+    } else if (size > pSet->pSmaF->size) {
+      code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE);
+      if (code) goto _err;
     }
 
-    // sma =============
-    tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
+    // sst ===========
+    for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+      tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iSst], fname);
       if (taosStatFile(fname, &size, NULL)) {
         code = TAOS_SYSTEM_ERROR(errno);
         goto _err;
       }
-    if (size < pSet->pSmaF->size) {
+      if (size != pSet->aSstF[iSst]->size) {
         code = TSDB_CODE_FILE_CORRUPTED;
         goto _err;
-    } else if (size > pSet->pSmaF->size) {
-      code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE);
-      if (code) goto _err;
+      }
     }
   }
 
@@ -382,41 +386,15 @@ static int32_t tsdbRecoverFS(STsdb *pTsdb, uint8_t *pData, int64_t nData) {
   taosArrayClear(pTsdb->fs.aDFileSet);
   n += tGetU32v(pData + n, &nSet);
   for (uint32_t iSet = 0; iSet < nSet; iSet++) {
-    SDFileSet fSet;
+    SDFileSet fSet = {0};
 
-    // head
-    fSet.pHeadF = (SHeadFile *)taosMemoryCalloc(1, sizeof(SHeadFile));
-    if (fSet.pHeadF == NULL) {
+    int32_t nt = tGetDFileSet(pData + n, &fSet);
+    if (nt < 0) {
       code = TSDB_CODE_OUT_OF_MEMORY;
       goto _err;
     }
-    fSet.pHeadF->nRef = 1;
 
-    // data
-    fSet.pDataF = (SDataFile *)taosMemoryCalloc(1, sizeof(SDataFile));
-    if (fSet.pDataF == NULL) {
-      code = TSDB_CODE_OUT_OF_MEMORY;
-      goto _err;
-    }
-    fSet.pDataF->nRef = 1;
-
-    // last
-    fSet.pLastF = (SLastFile *)taosMemoryCalloc(1, sizeof(SLastFile));
-    if (fSet.pLastF == NULL) {
-      code = TSDB_CODE_OUT_OF_MEMORY;
-      goto _err;
-    }
-    fSet.pLastF->nRef = 1;
-
-    // sma
-    fSet.pSmaF = (SSmaFile *)taosMemoryCalloc(1, sizeof(SSmaFile));
-    if (fSet.pSmaF == NULL) {
-      code = TSDB_CODE_OUT_OF_MEMORY;
-      goto _err;
-    }
-    fSet.pSmaF->nRef = 1;
-
-    n += tGetDFileSet(pData + n, &fSet);
+    n += nt;
 
     if (taosArrayPush(pTsdb->fs.aDFileSet, &fSet) == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
@@ -532,13 +510,15 @@ int32_t tsdbFSClose(STsdb *pTsdb) {
     ASSERT(pSet->pDataF->nRef == 1);
     taosMemoryFree(pSet->pDataF);
 
-    // last
-    ASSERT(pSet->pLastF->nRef == 1);
-    taosMemoryFree(pSet->pLastF);
-
     // sma
     ASSERT(pSet->pSmaF->nRef == 1);
     taosMemoryFree(pSet->pSmaF);
+
+    // sst
+    for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+      ASSERT(pSet->aSstF[iSst]->nRef == 1);
+      taosMemoryFree(pSet->aSstF[iSst]);
+    }
   }
 
   taosArrayDestroy(pTsdb->fs.aDFileSet);
@@ -586,21 +566,23 @@ int32_t tsdbFSCopy(STsdb *pTsdb, STsdbFS *pFS) {
     }
     *fSet.pDataF = *pSet->pDataF;
 
-    // data
-    fSet.pLastF = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
-    if (fSet.pLastF == NULL) {
+    // sma
+    fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
+    if (fSet.pSmaF == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
       goto _exit;
     }
-    *fSet.pLastF = *pSet->pLastF;
+    *fSet.pSmaF = *pSet->pSmaF;
 
-    // last
-    fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
-    if (fSet.pSmaF == NULL) {
+    // sst
+    for (fSet.nSstF = 0; fSet.nSstF < pSet->nSstF; fSet.nSstF++) {
+      fSet.aSstF[fSet.nSstF] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+      if (fSet.aSstF[fSet.nSstF] == NULL) {
         code = TSDB_CODE_OUT_OF_MEMORY;
         goto _exit;
       }
-    *fSet.pSmaF = *pSet->pSmaF;
+      *fSet.aSstF[fSet.nSstF] = *pSet->aSstF[fSet.nSstF];
+    }
 
     if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
@@ -651,14 +633,38 @@ int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) {
     if (c == 0) {
       *pDFileSet->pHeadF = *pSet->pHeadF;
       *pDFileSet->pDataF = *pSet->pDataF;
-      *pDFileSet->pLastF = *pSet->pLastF;
       *pDFileSet->pSmaF = *pSet->pSmaF;
+      // sst
+      if (pSet->nSstF > pDFileSet->nSstF) {
+        ASSERT(pSet->nSstF == pDFileSet->nSstF + 1);
 
+        pDFileSet->aSstF[pDFileSet->nSstF] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+        if (pDFileSet->aSstF[pDFileSet->nSstF] == NULL) {
+          code = TSDB_CODE_OUT_OF_MEMORY;
           goto _exit;
         }
+        *pDFileSet->aSstF[pDFileSet->nSstF] = *pSet->aSstF[pSet->nSstF - 1];
+        pDFileSet->nSstF++;
+      } else if (pSet->nSstF < pDFileSet->nSstF) {
+        ASSERT(pSet->nSstF == 1);
+        for (int32_t iSst = 1; iSst < pDFileSet->nSstF; iSst++) {
+          taosMemoryFree(pDFileSet->aSstF[iSst]);
         }
 
-  SDFileSet fSet = {.diskId = pSet->diskId, .fid = pSet->fid};
+        *pDFileSet->aSstF[0] = *pSet->aSstF[0];
+        pDFileSet->nSstF = 1;
+      } else {
+        for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+          *pDFileSet->aSstF[iSst] = *pSet->aSstF[iSst];
+        }
+      }
+
+      goto _exit;
+    }
+  }
+
+  ASSERT(pSet->nSstF == 1);
+  SDFileSet fSet = {.diskId = pSet->diskId, .fid = pSet->fid, .nSstF = 1};
 
   // head
   fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
@@ -676,21 +682,21 @@ int32_t tsdbFSUpsertFSet(STsdbFS *pFS, SDFileSet *pSet) {
   }
   *fSet.pDataF = *pSet->pDataF;
 
-  // data
-  fSet.pLastF = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
-  if (fSet.pLastF == NULL) {
+  // sma
+  fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
+  if (fSet.pSmaF == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
     goto _exit;
   }
-  *fSet.pLastF = *pSet->pLastF;
+  *fSet.pSmaF = *pSet->pSmaF;
 
-  // last
-  fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
-  if (fSet.pSmaF == NULL) {
+  // sst
+  fSet.aSstF[0] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+  if (fSet.aSstF[0] == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
     goto _exit;
   }
-  *fSet.pSmaF = *pSet->pSmaF;
+  *fSet.aSstF[0] = *pSet->aSstF[0];
 
   if (taosArrayInsert(pFS->aDFileSet, idx, &fSet) == NULL) {
     code = TSDB_CODE_OUT_OF_MEMORY;
@@ -836,27 +842,6 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
       pSetOld->pDataF->size = pSetNew->pDataF->size;
     }
 
-    // last
-    fSet.pLastF = pSetOld->pLastF;
-    if ((!sameDisk) || (pSetOld->pLastF->commitID != pSetNew->pLastF->commitID)) {
-      pSetOld->pLastF = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
-      if (pSetOld->pLastF == NULL) {
-        code = TSDB_CODE_OUT_OF_MEMORY;
-        goto _err;
-      }
-      *pSetOld->pLastF = *pSetNew->pLastF;
-      pSetOld->pLastF->nRef = 1;
-
-      nRef = atomic_sub_fetch_32(&fSet.pLastF->nRef, 1);
-      if (nRef == 0) {
-        tsdbLastFileName(pTsdb, pSetOld->diskId, pSetOld->fid, fSet.pLastF, fname);
-        taosRemoveFile(fname);
-        taosMemoryFree(fSet.pLastF);
-      }
-    } else {
-      ASSERT(pSetOld->pLastF->size == pSetNew->pLastF->size);
-    }
-
     // sma
     fSet.pSmaF = pSetOld->pSmaF;
     if ((!sameDisk) || (pSetOld->pSmaF->commitID != pSetNew->pSmaF->commitID)) {
@@ -879,6 +864,84 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
       pSetOld->pSmaF->size = pSetNew->pSmaF->size;
     }
 
+    // sst
+    if (sameDisk) {
+      if (pSetNew->nSstF > pSetOld->nSstF) {
+        ASSERT(pSetNew->nSstF = pSetOld->nSstF + 1);
+        pSetOld->aSstF[pSetOld->nSstF] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+        if (pSetOld->aSstF[pSetOld->nSstF] == NULL) {
+          code = TSDB_CODE_OUT_OF_MEMORY;
+          goto _err;
+        }
+        *pSetOld->aSstF[pSetOld->nSstF] = *pSetNew->aSstF[pSetOld->nSstF];
+        pSetOld->aSstF[pSetOld->nSstF]->nRef = 1;
+        pSetOld->nSstF++;
+      } else if (pSetNew->nSstF < pSetOld->nSstF) {
+        ASSERT(pSetNew->nSstF == 1);
+        for (int32_t iSst = 0; iSst < pSetOld->nSstF; iSst++) {
+          SSstFile *pSstFile = pSetOld->aSstF[iSst];
+          nRef = atomic_sub_fetch_32(&pSstFile->nRef, 1);
+          if (nRef == 0) {
+            tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSstFile, fname);
+            taosRemoveFile(fname);
+            taosMemoryFree(pSstFile);
+          }
+          pSetOld->aSstF[iSst] = NULL;
+        }
+
+        pSetOld->nSstF = 1;
+        pSetOld->aSstF[0] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+        if (pSetOld->aSstF[0] == NULL) {
+          code = TSDB_CODE_OUT_OF_MEMORY;
+          goto _err;
+        }
+        *pSetOld->aSstF[0] = *pSetNew->aSstF[0];
+        pSetOld->aSstF[0]->nRef = 1;
+      } else {
+        for (int32_t iSst = 0; iSst < pSetOld->nSstF; iSst++) {
+          if (pSetOld->aSstF[iSst]->commitID != pSetNew->aSstF[iSst]->commitID) {
+            SSstFile *pSstFile = pSetOld->aSstF[iSst];
+            nRef = atomic_sub_fetch_32(&pSstFile->nRef, 1);
+            if (nRef == 0) {
+              tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSstFile, fname);
+              taosRemoveFile(fname);
+              taosMemoryFree(pSstFile);
+            }
+
+            pSetOld->aSstF[iSst] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+            if (pSetOld->aSstF[iSst] == NULL) {
+              code = TSDB_CODE_OUT_OF_MEMORY;
+              goto _err;
+            }
+            *pSetOld->aSstF[iSst] = *pSetNew->aSstF[iSst];
+            pSetOld->aSstF[iSst]->nRef = 1;
+          } else {
+            ASSERT(pSetOld->aSstF[iSst]->size == pSetOld->aSstF[iSst]->size);
+            ASSERT(pSetOld->aSstF[iSst]->offset == pSetOld->aSstF[iSst]->offset);
+          }
+        }
+      }
+    } else {
+      ASSERT(pSetOld->nSstF == pSetNew->nSstF);
+      for (int32_t iSst = 0; iSst < pSetOld->nSstF; iSst++) {
+        SSstFile *pSstFile = pSetOld->aSstF[iSst];
+        nRef = atomic_sub_fetch_32(&pSstFile->nRef, 1);
+        if (nRef == 0) {
+          tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSstFile, fname);
+          taosRemoveFile(fname);
+          taosMemoryFree(pSstFile);
+        }
+
+        pSetOld->aSstF[iSst] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+        if (pSetOld->aSstF[iSst] == NULL) {
+          code = TSDB_CODE_OUT_OF_MEMORY;
+          goto _err;
+        }
+        *pSetOld->aSstF[iSst] = *pSetNew->aSstF[iSst];
+        pSetOld->aSstF[iSst]->nRef = 1;
+      }
+    }
+
     if (!sameDisk) {
       pSetOld->diskId = pSetNew->diskId;
     }
@@ -902,26 +965,27 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
       taosMemoryFree(pSetOld->pDataF);
     }
 
-    nRef = atomic_sub_fetch_32(&pSetOld->pLastF->nRef, 1);
+    nRef = atomic_sub_fetch_32(&pSetOld->pSmaF->nRef, 1);
     if (nRef == 0) {
-      tsdbLastFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSetOld->pLastF, fname);
+      tsdbSmaFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSetOld->pSmaF, fname);
       taosRemoveFile(fname);
-      taosMemoryFree(pSetOld->pLastF);
+      taosMemoryFree(pSetOld->pSmaF);
     }
 
-    nRef = atomic_sub_fetch_32(&pSetOld->pSmaF->nRef, 1);
+    for (int8_t iSst = 0; iSst < pSetOld->nSstF; iSst++) {
+      nRef = atomic_sub_fetch_32(&pSetOld->aSstF[iSst]->nRef, 1);
       if (nRef == 0) {
-      tsdbSmaFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSetOld->pSmaF, fname);
+        tsdbSstFileName(pTsdb, pSetOld->diskId, pSetOld->fid, pSetOld->aSstF[iSst], fname);
         taosRemoveFile(fname);
-      taosMemoryFree(pSetOld->pSmaF);
+        taosMemoryFree(pSetOld->aSstF[iSst]);
+      }
     }
 
     taosArrayRemove(pTsdb->fs.aDFileSet, iOld);
     continue;
 
   _add_new:
-    fSet.diskId = pSetNew->diskId;
-    fSet.fid = pSetNew->fid;
+    fSet = (SDFileSet){.diskId = pSetNew->diskId, .fid = pSetNew->fid, .nSstF = 1};
 
     // head
     fSet.pHeadF = (SHeadFile *)taosMemoryMalloc(sizeof(SHeadFile));
@@ -941,15 +1005,6 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
     *fSet.pDataF = *pSetNew->pDataF;
     fSet.pDataF->nRef = 1;
 
-    // last
-    fSet.pLastF = (SLastFile *)taosMemoryMalloc(sizeof(SLastFile));
-    if (fSet.pLastF == NULL) {
-      code = TSDB_CODE_OUT_OF_MEMORY;
-      goto _err;
-    }
-    *fSet.pLastF = *pSetNew->pLastF;
-    fSet.pLastF->nRef = 1;
-
     // sma
     fSet.pSmaF = (SSmaFile *)taosMemoryMalloc(sizeof(SSmaFile));
     if (fSet.pSmaF == NULL) {
@@ -959,6 +1014,16 @@ int32_t tsdbFSCommit2(STsdb *pTsdb, STsdbFS *pFSNew) {
     *fSet.pSmaF = *pSetNew->pSmaF;
     fSet.pSmaF->nRef = 1;
 
+    // sst
+    ASSERT(pSetNew->nSstF == 1);
+    fSet.aSstF[0] = (SSstFile *)taosMemoryMalloc(sizeof(SSstFile));
+    if (fSet.aSstF[0] == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+      goto _err;
+    }
+    *fSet.aSstF[0] = *pSetNew->aSstF[0];
+    fSet.aSstF[0]->nRef = 1;
+
     if (taosArrayInsert(pTsdb->fs.aDFileSet, iOld, &fSet) == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
       goto _err;
@@ -1002,11 +1067,13 @@ int32_t tsdbFSRef(STsdb *pTsdb, STsdbFS *pFS) {
     nRef = atomic_fetch_add_32(&pSet->pDataF->nRef, 1);
     ASSERT(nRef > 0);
 
-    nRef = atomic_fetch_add_32(&pSet->pLastF->nRef, 1);
+    nRef = atomic_fetch_add_32(&pSet->pSmaF->nRef, 1);
     ASSERT(nRef > 0);
 
-    nRef = atomic_fetch_add_32(&pSet->pSmaF->nRef, 1);
+    for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+      nRef = atomic_fetch_add_32(&pSet->aSstF[iSst]->nRef, 1);
       ASSERT(nRef > 0);
+    }
 
     if (taosArrayPush(pFS->aDFileSet, &fSet) == NULL) {
       code = TSDB_CODE_OUT_OF_MEMORY;
@@ -1053,22 +1120,25 @@ void tsdbFSUnref(STsdb *pTsdb, STsdbFS *pFS) {
       taosMemoryFree(pSet->pDataF);
     }
 
-    // last
-    nRef = atomic_sub_fetch_32(&pSet->pLastF->nRef, 1);
+    // sma
+    nRef = atomic_sub_fetch_32(&pSet->pSmaF->nRef, 1);
     ASSERT(nRef >= 0);
     if (nRef == 0) {
-      tsdbLastFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pLastF, fname);
+      tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
       taosRemoveFile(fname);
-      taosMemoryFree(pSet->pLastF);
+      taosMemoryFree(pSet->pSmaF);
     }
 
-    // sma
-    nRef = atomic_sub_fetch_32(&pSet->pSmaF->nRef, 1);
+    // sst
+    for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+      nRef = atomic_sub_fetch_32(&pSet->aSstF[iSst]->nRef, 1);
       ASSERT(nRef >= 0);
       if (nRef == 0) {
-      tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
+        tsdbSstFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSstF[iSst], fname);
         taosRemoveFile(fname);
-      taosMemoryFree(pSet->pSmaF);
+        taosMemoryFree(pSet->aSstF[iSst]);
+        /* code */
+      }
     }
   }
 

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 tPutSstFile(uint8_t *p, SSstFile *pSstFile) {
   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, pSstFile->commitID);
+  n += tPutI64v(p ? p + n : p, pSstFile->size);
+  n += tPutI64v(p ? p + n : p, pSstFile->offset);
 
   return n;
 }
 
-static int32_t tGetLastFile(uint8_t *p, SLastFile *pLastFile) {
+static int32_t tGetSstFile(uint8_t *p, SSstFile *pSstFile) {
   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, &pSstFile->commitID);
+  n += tGetI64v(p + n, &pSstFile->size);
+  n += tGetI64v(p + n, &pSstFile->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 tsdbSstFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSstFile *pSstF, 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, pSstF->commitID, ".sst");
 }
 
 void tsdbSmaFileName(STsdb *pTsdb, SDiskID did, int32_t fid, SSmaFile *pSmaF, char fname[]) {
@@ -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);
+  // sst
+  n += tPutU8(p ? p + n : p, pSet->nSstF);
+  for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+    n += tPutSstFile(p ? p + n : p, pSet->aSstF[iSst]);
+  }
 
   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);
+  // sst
+  n += tGetU8(p + n, &pSet->nSstF);
+  for (int32_t iSst = 0; iSst < pSet->nSstF; iSst++) {
+    pSet->aSstF[iSst] = (SSstFile *)taosMemoryCalloc(1, sizeof(SSstFile));
+    if (pSet->aSstF[iSst] == NULL) {
+      return -1;
+    }
+    pSet->aSstF[iSst]->nRef = 1;
+    n += tGetSstFile(p + n, pSet->aSstF[iSst]);
+  }
 
   return n;
 }

source/dnode/vnode/src/tsdb/tsdbMergeTree.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"
+
+// SLDataIter =================================================
+typedef struct SLDataIter {
+  SRBTreeNode   node;
+  SSstBlk      *pSstBlk;
+  SDataFReader *pReader;
+  int32_t       iSst;
+  int8_t        backward;
+  SArray       *aSstBlk;
+  int32_t       iSstBlk;
+  SBlockData    bData[2];
+  int32_t       loadIndex;
+  int32_t       iRow;
+  SRowInfo      rInfo;
+  uint64_t      uid;
+  STimeWindow   timeWindow;
+  SVersionRange verRange;
+} SLDataIter;
+
+static SBlockData* getCurrentBlock(SLDataIter* pIter) {
+  return &pIter->bData[pIter->loadIndex];
+}
+
+static SBlockData* getNextBlock(SLDataIter* pIter) {
+  pIter->loadIndex ^= 1;
+  return getCurrentBlock(pIter);
+}
+
+int32_t tLDataIterOpen(struct SLDataIter **pIter, SDataFReader *pReader, int32_t iSst, int8_t backward, uint64_t uid,
+                       STimeWindow *pTimeWindow, SVersionRange *pRange) {
+  int32_t code = 0;
+  *pIter = taosMemoryCalloc(1, sizeof(SLDataIter));
+  if (*pIter == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
+
+  (*pIter)->uid = uid;
+  (*pIter)->timeWindow = *pTimeWindow;
+  (*pIter)->verRange = *pRange;
+  (*pIter)->pReader = pReader;
+  (*pIter)->iSst = iSst;
+  (*pIter)->backward = backward;
+  (*pIter)->aSstBlk = taosArrayInit(0, sizeof(SSstBlk));
+  if ((*pIter)->aSstBlk == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
+
+  code = tBlockDataCreate(&(*pIter)->bData[0]);
+  if (code) {
+    goto _exit;
+  }
+
+  code = tBlockDataCreate(&(*pIter)->bData[1]);
+  if (code) {
+    goto _exit;
+  }
+
+  code = tsdbReadSstBlk(pReader, iSst, (*pIter)->aSstBlk);
+  if (code) {
+    goto _exit;
+  }
+
+  size_t size = taosArrayGetSize((*pIter)->aSstBlk);
+
+  // find the start block
+  int32_t index = -1;
+  if (!backward) {  // asc
+    for (int32_t i = 0; i < size; ++i) {
+      SSstBlk *p = taosArrayGet((*pIter)->aSstBlk, i);
+      if (p->minUid <= uid && p->maxUid >= uid) {
+        index = i;
+        break;
+      }
+    }
+  } else {  // desc
+    for (int32_t i = size - 1; i >= 0; --i) {
+      SSstBlk *p = taosArrayGet((*pIter)->aSstBlk, i);
+      if (p->minUid <= uid && p->maxUid >= uid) {
+        index = i;
+        break;
+      }
+    }
+  }
+
+  (*pIter)->iSstBlk = index;
+  if (index != -1) {
+    (*pIter)->pSstBlk = taosArrayGet((*pIter)->aSstBlk, (*pIter)->iSstBlk);
+  }
+
+_exit:
+  return code;
+}
+
+void tLDataIterClose(SLDataIter *pIter) {
+  tBlockDataDestroy(&pIter->bData[0], 1);
+  tBlockDataDestroy(&pIter->bData[1], 1);
+  taosArrayDestroy(pIter->aSstBlk);
+  taosMemoryFree(pIter);
+}
+
+extern int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData);
+
+void tLDataIterNextBlock(SLDataIter *pIter) {
+  int32_t step = pIter->backward ? -1 : 1;
+  pIter->iSstBlk += step;
+
+  int32_t index = -1;
+  size_t  size = taosArrayGetSize(pIter->aSstBlk);
+  for (int32_t i = pIter->iSstBlk; i < size && i >= 0; i += step) {
+    SSstBlk *p = taosArrayGet(pIter->aSstBlk, i);
+    if ((!pIter->backward) && p->minUid > pIter->uid) {
+      break;
+    }
+
+    if (pIter->backward && p->maxUid < pIter->uid) {
+      break;
+    }
+
+    if (p->minUid <= pIter->uid && p->maxUid >= pIter->uid) {
+      index = i;
+      break;
+    }
+  }
+
+  if (index == -1) {
+    pIter->pSstBlk = NULL;
+  } else {
+    pIter->pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, pIter->iSstBlk);
+  }
+}
+
+static void findNextValidRow(SLDataIter *pIter) {
+  int32_t step = pIter->backward ? -1 : 1;
+
+  bool    hasVal = false;
+  int32_t i = pIter->iRow;
+  SBlockData* pBlockData = getCurrentBlock(pIter);
+
+  for (; i < pBlockData->nRow && i >= 0; i += step) {
+    if (pBlockData->aUid != NULL) {
+      if (!pIter->backward) {
+        if (pBlockData->aUid[i] < pIter->uid) {
+          continue;
+        } else if (pBlockData->aUid[i] > pIter->uid) {
+          break;
+        }
+      } else {
+        if (pBlockData->aUid[i] > pIter->uid) {
+          continue;
+        } else if (pBlockData->aUid[i] < pIter->uid) {
+          break;
+        }
+      }
+    }
+
+    int64_t ts = pBlockData->aTSKEY[i];
+    if (!pIter->backward) {               // asc
+      if (ts > pIter->timeWindow.ekey) {  // no more data
+        break;
+      } else if (ts < pIter->timeWindow.skey) {
+        continue;
+      }
+    } else {
+      if (ts < pIter->timeWindow.skey) {
+        break;
+      } else if (ts > pIter->timeWindow.ekey) {
+        continue;
+      }
+    }
+
+    int64_t ver = pBlockData->aVersion[i];
+    if (ver < pIter->verRange.minVer) {
+      continue;
+    }
+
+    // todo opt handle desc case
+    if (ver > pIter->verRange.maxVer) {
+      continue;
+    }
+
+    //  todo handle delete soon
+#if 0
+    TSDBKEY k = {.ts = ts, .version = ver};
+        if (hasBeenDropped(pBlockScanInfo->delSkyline, &pBlockScanInfo->lastBlockDelIndex, &k, pLastBlockReader->order)) {
+          continue;
+        }
+#endif
+
+    hasVal = true;
+    break;
+  }
+
+  pIter->iRow = (hasVal) ? i : -1;
+}
+
+bool tLDataIterNextRow(SLDataIter *pIter) {
+  int32_t code = 0;
+  int32_t step = pIter->backward ? -1 : 1;
+
+  // no qualified last file block in current file, no need to fetch row
+  if (pIter->pSstBlk == NULL) {
+    return false;
+  }
+
+  int32_t iBlockL = pIter->iSstBlk;
+  SBlockData* pBlockData = getCurrentBlock(pIter);
+
+  if (pBlockData->nRow == 0 && pIter->pSstBlk != NULL) {  // current block not loaded yet
+    pBlockData = getNextBlock(pIter);
+    code = tsdbReadSstBlockEx(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _exit;
+    }
+
+    pIter->iRow = (pIter->backward) ? pBlockData->nRow : -1;
+  }
+
+  pIter->iRow += step;
+
+  while (1) {
+    findNextValidRow(pIter);
+
+    if (pIter->iRow >= pBlockData->nRow || pIter->iRow < 0) {
+      tLDataIterNextBlock(pIter);
+      if (pIter->pSstBlk == NULL) {  // no more data
+        goto _exit;
+      }
+    } else {
+      break;
+    }
+
+    if (iBlockL != pIter->iSstBlk) {
+      pBlockData = getNextBlock(pIter);
+      code = tsdbReadSstBlockEx(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData);
+      if (code) {
+        goto _exit;
+      }
+      pIter->iRow = pIter->backward ? (pBlockData->nRow - 1) : 0;
+    }
+  }
+
+  pIter->rInfo.suid = pBlockData->suid;
+  pIter->rInfo.uid = pBlockData->uid;
+  pIter->rInfo.row = tsdbRowFromBlockData(pBlockData, pIter->iRow);
+
+_exit:
+  if (code != TSDB_CODE_SUCCESS) {
+    terrno = code;
+  }
+
+  return (code == TSDB_CODE_SUCCESS) && (pIter->pSstBlk != NULL);
+}
+
+SRowInfo *tLDataIterGet(SLDataIter *pIter) { return &pIter->rInfo; }
+
+// SMergeTree =================================================
+static FORCE_INLINE int32_t tLDataIterCmprFn(const void *p1, const void *p2) {
+  SLDataIter *pIter1 = (SLDataIter *)(((uint8_t *)p1) - sizeof(SRBTreeNode));
+  SLDataIter *pIter2 = (SLDataIter *)(((uint8_t *)p2) - sizeof(SRBTreeNode));
+
+  TSDBKEY key1 = TSDBROW_KEY(&pIter1->rInfo.row);
+  TSDBKEY key2 = TSDBROW_KEY(&pIter2->rInfo.row);
+
+  if (key1.ts < key2.ts) {
+    return -1;
+  } else if (key1.ts > key2.ts) {
+    return 1;
+  } else {
+    if (key1.version < key2.version) {
+      return -1;
+    } else if (key1.version > key2.version) {
+      return 1;
+    } else {
+      return 0;
+    }
+  }
+}
+
+int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t uid,
+                       STimeWindow *pTimeWindow, SVersionRange *pVerRange) {
+  pMTree->backward = backward;
+  pMTree->pIter = NULL;
+  pMTree->pIterList = taosArrayInit(4, POINTER_BYTES);
+  if (pMTree->pIterList == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn);
+  int32_t code = TSDB_CODE_OUT_OF_MEMORY;
+
+  struct SLDataIter *pIterList[TSDB_DEFAULT_LAST_FILE] = {0};
+  for (int32_t i = 0; i < pFReader->pSet->nSstF; ++i) {  // open all last file
+    code = tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, uid, pTimeWindow, pVerRange);
+    if (code != TSDB_CODE_SUCCESS) {
+      goto _end;
+    }
+
+    bool hasVal = tLDataIterNextRow(pIterList[i]);
+    if (hasVal) {
+      taosArrayPush(pMTree->pIterList, &pIterList[i]);
+      tMergeTreeAddIter(pMTree, pIterList[i]);
+    } else {
+      tLDataIterClose(pIterList[i]);
+    }
+  }
+
+  return code;
+
+_end:
+  tMergeTreeClose(pMTree);
+  return code;
+}
+
+void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter) { tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pIter); }
+
+bool tMergeTreeNext(SMergeTree *pMTree) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (pMTree->pIter) {
+    SLDataIter *pIter = pMTree->pIter;
+
+    bool hasVal = tLDataIterNextRow(pIter);
+    if (!hasVal) {
+      pMTree->pIter = NULL;
+    }
+
+    // compare with min in RB Tree
+    pIter = (SLDataIter *)tRBTreeMin(&pMTree->rbt);
+    if (pMTree->pIter && pIter) {
+      int32_t c = pMTree->rbt.cmprFn(RBTREE_NODE_PAYLOAD(&pMTree->pIter->node), RBTREE_NODE_PAYLOAD(&pIter->node));
+      if (c > 0) {
+        tRBTreePut(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);
+        pMTree->pIter = NULL;
+      } else {
+        ASSERT(c);
+      }
+    }
+  }
+
+  if (pMTree->pIter == NULL) {
+    pMTree->pIter = (SLDataIter *)tRBTreeMin(&pMTree->rbt);
+    if (pMTree->pIter) {
+      tRBTreeDrop(&pMTree->rbt, (SRBTreeNode *)pMTree->pIter);
+    }
+  }
+
+  return pMTree->pIter != NULL;
+}
+
+TSDBROW tMergeTreeGetRow(SMergeTree *pMTree) { return pMTree->pIter->rInfo.row; }
+
+void tMergeTreeClose(SMergeTree *pMTree) {
+  size_t size = taosArrayGetSize(pMTree->pIterList);
+  for (int32_t i = 0; i < size; ++i) {
+    SLDataIter *pIter = taosArrayGetP(pMTree->pIterList, i);
+    tLDataIterClose(pIter);
+  }
+
+  pMTree->pIterList = taosArrayDestroy(pMTree->pIterList);
+  pMTree->pIter = NULL;
+}

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->aSstF[0]);
       taosMemoryFree(pSet->pSmaF);
       taosArrayRemove(fs.aDFileSet, iSet);
       iSet--;

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/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/dataDispatcher.c

@@ -93,6 +93,8 @@ static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pIn
 
   pBuf->useSize = sizeof(SDataCacheEntry);
   blockEncode(pInput->pData, pEntry->data, &pEntry->dataLen, numOfCols, pEntry->compressed);
+  ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data+8));
+  ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data+8+4));
 
   pBuf->useSize += pEntry->dataLen;
 
@@ -170,7 +172,13 @@ static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, bool* pQueryE
   taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf);
   memcpy(&pDispatcher->nextOutput, pBuf, sizeof(SDataDispatchBuf));
   taosFreeQitem(pBuf);
-  *pLen = ((SDataCacheEntry*)(pDispatcher->nextOutput.pData))->dataLen;
+
+  SDataCacheEntry* pEntry = (SDataCacheEntry*)pDispatcher->nextOutput.pData;
+  *pLen = pEntry->dataLen;
+
+  ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data+8));
+  ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data+8+4));
+
   *pQueryEnd = pDispatcher->queryEnd;
   qDebug("got data len %" PRId64 ", row num %d in sink", *pLen, ((SDataCacheEntry*)(pDispatcher->nextOutput.pData))->numOfRows);
 }
@@ -191,6 +199,9 @@ static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) {
   pOutput->numOfCols = pEntry->numOfCols;
   pOutput->compressed = pEntry->compressed;
 
+  ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data+8));
+  ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data+8+4));
+
   atomic_sub_fetch_64(&pDispatcher->cachedSize, pEntry->dataLen);
   atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen);
 

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);

source/libs/executor/src/timewindowoperator.c

@@ -622,7 +622,8 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num
     if (pr->closed) {
       ASSERT(isResultRowInterpolated(pr, RESULT_ROW_START_INTERP) &&
              isResultRowInterpolated(pr, RESULT_ROW_END_INTERP));
-      tdListPopHead(pResultRowInfo->openWindow);
+      SListNode* pNode = tdListPopHead(pResultRowInfo->openWindow);
+      taosMemoryFree(pNode);
       continue;
     }
 
@@ -651,7 +652,8 @@ static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t num
 
     if (isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP)) {
       closeResultRow(pr);
-      tdListPopHead(pResultRowInfo->openWindow);
+      SListNode* pNode = tdListPopHead(pResultRowInfo->openWindow);
+      taosMemoryFree(pNode);
     } else {  // the remains are can not be closed yet.
       break;
     }
@@ -1731,6 +1733,10 @@ void destroyIntervalOperatorInfo(void* param) {
   SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)param;
   cleanupBasicInfo(&pInfo->binfo);
   cleanupAggSup(&pInfo->aggSup);
+  cleanupExprSupp(&pInfo->scalarSupp);
+
+  tdListFree(pInfo->binfo.resultRowInfo.openWindow);
+
   pInfo->pRecycledPages = taosArrayDestroy(pInfo->pRecycledPages);
   pInfo->pInterpCols = taosArrayDestroy(pInfo->pInterpCols);
   taosArrayDestroyEx(pInfo->pPrevValues, freeItem);

source/libs/monitor/src/monMsg.c

@@ -510,6 +510,7 @@ int32_t tSerializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInfo)
     SVnodeLoad *pLoad = taosArrayGet(pInfo->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;
@@ -544,6 +545,7 @@ int32_t tDeserializeSMonVloadInfo(void *buf, int32_t bufLen, SMonVloadInfo *pInf
     SVnodeLoad load = {0};
     if (tDecodeI32(&decoder, &load.vgId) < 0) return -1;
     if (tDecodeI32(&decoder, &load.syncState) < 0) return -1;
+    if (tDecodeI64(&decoder, &load.cacheUsage) < 0) return -1;
     if (tDecodeI64(&decoder, &load.numOfTables) < 0) return -1;
     if (tDecodeI64(&decoder, &load.numOfTimeSeries) < 0) return -1;
     if (tDecodeI64(&decoder, &load.totalStorage) < 0) return -1;
@@ -594,7 +596,6 @@ int32_t tDeserializeSMonMloadInfo(void *buf, int32_t bufLen, SMonMloadInfo *pInf
   return 0;
 }
 
-
 int32_t tSerializeSQnodeLoad(void *buf, int32_t bufLen, SQnodeLoad *pInfo) {
   SEncoder encoder = {0};
   tEncoderInit(&encoder, buf, bufLen);
@@ -639,5 +640,3 @@ int32_t tDeserializeSQnodeLoad(void *buf, int32_t bufLen, SQnodeLoad *pInfo) {
   tDecoderClear(&decoder);
   return 0;
 }
-
-

source/util/test/CMakeLists.txt

@@ -76,3 +76,11 @@ add_test(
     NAME taosbsearchTest
     COMMAND taosbsearchTest
 )
+
+# trbtreeTest
+add_executable(rbtreeTest "trbtreeTest.cpp")
+target_link_libraries(rbtreeTest os util gtest_main)   
+add_test(
+    NAME rbtreeTest
+    COMMAND rbtreeTest
+)
\ No newline at end of file

source/util/test/trbtreeTest.cpp

+#include <gtest/gtest.h>
+
+#include <stdio.h>
+#include <stdlib.h>
+
+#include "trbtree.h"
+
+static int32_t tCmprInteger(const void *p1, const void *p2) {
+  if (*(int *)p1 < *(int *)p2) {
+    return -1;
+  } else if (*(int *)p1 > *(int *)p2) {
+    return 1;
+  }
+  return 0;
+}
+
+TEST(trbtreeTest, rbtree_test1) {
+#if 0
+  SRBTree rt;
+  tRBTreeCreate(&rt, tCmprInteger);
+  int a[] = {1, 3, 4, 2, 7, 5, 8};
+
+  for (int i = 0; i < sizeof(a) / sizeof(a[0]); i++) {
+    SRBTreeNode *pNode = (SRBTreeNode *)taosMemoryMalloc(sizeof(*pNode) + sizeof(int));
+    *(int *)pNode->payload = a[i];
+
+    tRBTreePut(&rt, pNode);
+  }
+
+  SRBTreeIter  rti = tRBTreeIterCreate(&rt, 1);
+  SRBTreeNode *pNode = tRBTreeIterNext(&rti);
+  int          la = 0;
+  while (pNode) {
+    GTEST_ASSERT_GT(*(int *)pNode->payload, la);
+    la = *(int *)pNode->payload;
+    // printf("%d\n", la);
+    pNode = tRBTreeIterNext(&rti);
+  }
+#endif
+}
\ No newline at end of file

tests/script/tsim/parser/slimit_alter_tags.sim

@@ -128,6 +128,7 @@ if $rows != 5 then
   return -1
 endi
 if $data00 != $rowNum then
+  print expect $rowNum , actual: $data00
   return -1
 endi
 if $data10 != $rowNum then