merge 3.0

contrib/test/CMakeLists.txt

@@ -24,3 +24,4 @@ if(${BUILD_WITH_TRAFT})
 endif(${BUILD_WITH_TRAFT})
 
 add_subdirectory(tdev)
+add_subdirectory(lz4)

contrib/test/lz4/CMakeLists.txt

+add_executable(lz4_test "")
+target_sources(lz4_test
+    PRIVATE
+    "main.c"
+)
+target_link_libraries(lz4_test lz4_static)
\ No newline at end of file

contrib/test/lz4/main.c

+#include <stdio.h>
+
+#include "lz4.h"
+
+int main(int argc, char const *argv[]) {
+  printf("%d\n", LZ4_compressBound(1024));
+  return 0;
+}

rust @ 7ed7a977

-Subproject commit 1c8924dc668e6aa848214c2fc54e3ace3f5bf8df
+Subproject commit 7ed7a97715388fa144718764d6bf20f9bfc29a12

include/common/tcommon.h

@@ -73,15 +73,17 @@ typedef struct {
   uint64_t  suid;
 } STableListInfo;
 
+#pragma pack(push, 1)
 typedef struct SColumnDataAgg {
   int16_t colId;
-  int16_t maxIndex;
   int16_t minIndex;
+  int16_t maxIndex;
   int16_t numOfNull;
   int64_t sum;
   int64_t max;
   int64_t min;
 } SColumnDataAgg;
+#pragma pack(pop)
 
 typedef struct SDataBlockInfo {
   STimeWindow window;
@@ -122,13 +124,11 @@ typedef struct SColumnInfoData {
 } SColumnInfoData;
 
 typedef struct SQueryTableDataCond {
-  // STimeWindow  twindow;
   uint64_t     suid;
   int32_t      order;  // desc|asc order to iterate the data block
   int32_t      numOfCols;
   SColumnInfo* colList;
-  bool         loadExternalRows;  // load external rows or not
-  int32_t      type;              // data block load type:
+  int32_t      type;  // data block load type:
   int32_t      numOfTWindows;
   STimeWindow* twindows;
   int64_t      startVersion;

include/common/tdataformat.h

@@ -34,21 +34,40 @@ typedef struct SValue        SValue;
 typedef struct SColVal       SColVal;
 typedef struct STSRow2       STSRow2;
 typedef struct STSRowBuilder STSRowBuilder;
-typedef struct SColData      SColData;
 typedef struct STagVal       STagVal;
 typedef struct STag          STag;
 
+// bitmap
+#define N1(n)        ((1 << (n)) - 1)
+#define BIT1_SIZE(n) (((n)-1) / 8 + 1)
+#define BIT2_SIZE(n) (((n)-1) / 4 + 1)
+#define SET_BIT1(p, i, v)                            \
+  do {                                               \
+    (p)[(i) / 8] &= N1((i) % 8);                     \
+    (p)[(i) / 8] |= (((uint8_t)(v)) << (((i) % 8))); \
+  } while (0)
+
+#define GET_BIT1(p, i) (((p)[(i) / 8] >> ((i) % 8)) & ((uint8_t)1))
+#define SET_BIT2(p, i, v)                                \
+  do {                                                   \
+    p[(i) / 4] &= N1((i) % 4 * 2);                       \
+    (p)[(i) / 4] |= (((uint8_t)(v)) << (((i) % 4) * 2)); \
+  } while (0)
+#define GET_BIT2(p, i) (((p)[(i) / 4] >> (((i) % 4) * 2)) & ((uint8_t)3))
+
 // STSchema
 int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t nCols, STSchema **ppTSchema);
 void    tTSchemaDestroy(STSchema *pTSchema);
 
 // SValue
-int tValueCmprFn(const SValue *pValue1, const SValue *pValue2, int8_t type);
+int32_t tPutValue(uint8_t *p, SValue *pValue, int8_t type);
+int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type);
+int     tValueCmprFn(const SValue *pValue1, const SValue *pValue2, int8_t type);
 
 // STSRow2
-#define COL_VAL_NONE(CID)     ((SColVal){.cid = (CID), .isNone = 1})
-#define COL_VAL_NULL(CID)     ((SColVal){.cid = (CID), .isNull = 1})
-#define COL_VAL_VALUE(CID, V) ((SColVal){.cid = (CID), .value = (V)})
+#define COL_VAL_NONE(CID, TYPE)     ((SColVal){.cid = (CID), .type = (TYPE), .isNone = 1})
+#define COL_VAL_NULL(CID, TYPE)     ((SColVal){.cid = (CID), .type = (TYPE), .isNull = 1})
+#define COL_VAL_VALUE(CID, TYPE, V) ((SColVal){.cid = (CID), .type = (TYPE), .value = (V)})
 
 int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow);
 int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow);
@@ -140,6 +159,7 @@ struct SValue {
 
 struct SColVal {
   int16_t cid;
+  int8_t  type;
   int8_t  isNone;
   int8_t  isNull;
   SValue  value;
@@ -172,12 +192,6 @@ struct STag {
 };
 #pragma pack(pop)
 
-struct SColData {
-  int16_t  cid;
-  uint32_t nData;
-  uint8_t *pData;
-};
-
 #if 1  //================================================================================================================================================
 // Imported since 3.0 and use bitmap to demonstrate None/Null/Norm, while use Null/Norm below 3.0 without of bitmap.
 #define TD_SUPPORT_BITMAP

include/common/tmsg.h

@@ -3024,6 +3024,17 @@ typedef struct {
 int32_t tEncodeSVDeleteRsp(SEncoder* pCoder, const SVDeleteRsp* pReq);
 int32_t tDecodeSVDeleteRsp(SDecoder* pCoder, SVDeleteRsp* pReq);
 
+typedef struct SDeleteRes {
+  uint64_t suid;
+  SArray*  uidList;
+  int64_t  skey;
+  int64_t  ekey;
+  int64_t  affectedRows;
+} SDeleteRes;
+
+int32_t tEncodeDeleteRes(SEncoder* pCoder, const SDeleteRes* pRes);
+int32_t tDecodeDeleteRes(SDecoder* pCoder, SDeleteRes* pRes);
+
 #pragma pack(pop)
 
 #ifdef __cplusplus

include/common/tmsgdef.h

@@ -200,6 +200,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_ALTER_HASHRANGE, "alter-hashrange", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_COMPACT, "compact", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_DROP_TTL_TABLE, "drop-ttl-stb", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_COMMIT, "commit vnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_MAX_MSG, "vnd-max", NULL, NULL)
 
   TD_NEW_MSG_SEG(TDMT_SCH_MSG)

include/common/trow.h

@@ -299,6 +299,7 @@ int32_t tdAppendColValToRow(SRowBuilder *pBuilder, col_id_t colId, int8_t colTyp
 int32_t tdGetTpRowValOfCol(SCellVal *output, STSRow *pRow, void *pBitmap, int8_t colType, int32_t offset,
                            int16_t colIdx);
 int32_t tdGetKvRowValOfCol(SCellVal *output, STSRow *pRow, void *pBitmap, int32_t offset, int16_t colIdx);
+void    tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColVal);
 
 typedef struct {
   STSchema *pSchema;
@@ -312,6 +313,7 @@ typedef struct {
 
 void    tdSTSRowIterReset(STSRowIter *pIter, STSRow *pRow);
 void    tdSTSRowIterInit(STSRowIter *pIter, STSchema *pSchema);
+int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow);
 bool    tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal);
 bool    tdGetTpRowDataOfCol(STSRowIter *pIter, col_type_t colType, int32_t offset, SCellVal *pVal);
 bool    tdGetKvRowValOfColEx(STSRowIter *pIter, col_id_t colId, col_type_t colType, col_id_t *nIdx, SCellVal *pVal);
@@ -320,7 +322,7 @@ STSRow *mergeTwoRows(void *buffer, STSRow *row1, STSRow *row2, STSchema *pSchema
 int32_t tdGetColDataOfRow(SCellVal *pVal, SDataCol *pCol, int32_t row, int8_t bitmapMode);
 bool    tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t flen, uint32_t offset, col_id_t colIdx,
                        SCellVal *pVal);
-bool    tdSKvRowGetVal(STSRow *pRow, col_id_t colId, uint32_t offset, col_id_t colIdx, SCellVal *pVal);
+bool    tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCellVal *pVal);
 int32_t dataColGetNEleLen(SDataCol *pDataCol, int32_t rows, int8_t bitmapMode);
 void    tdSCellValPrint(SCellVal *pVal, int8_t colType);
 void    tdSRowPrint(STSRow *row, STSchema *pSchema, const char *tag);

include/common/ttokendef.h

@@ -73,200 +73,201 @@
 #define TK_MNODE          55
 #define TK_DATABASE       56
 #define TK_USE            57
-#define TK_IF             58
-#define TK_NOT            59
-#define TK_EXISTS         60
-#define TK_BUFFER         61
-#define TK_CACHELAST      62
-#define TK_COMP           63
-#define TK_DURATION       64
-#define TK_NK_VARIABLE    65
-#define TK_FSYNC          66
-#define TK_MAXROWS        67
-#define TK_MINROWS        68
-#define TK_KEEP           69
-#define TK_PAGES          70
-#define TK_PAGESIZE       71
-#define TK_PRECISION      72
-#define TK_REPLICA        73
-#define TK_STRICT         74
-#define TK_WAL            75
-#define TK_VGROUPS        76
-#define TK_SINGLE_STABLE  77
-#define TK_RETENTIONS     78
-#define TK_SCHEMALESS     79
-#define TK_NK_COLON       80
-#define TK_TABLE          81
-#define TK_NK_LP          82
-#define TK_NK_RP          83
-#define TK_STABLE         84
-#define TK_ADD            85
-#define TK_COLUMN         86
-#define TK_MODIFY         87
-#define TK_RENAME         88
-#define TK_TAG            89
-#define TK_SET            90
-#define TK_NK_EQ          91
-#define TK_USING          92
-#define TK_TAGS           93
-#define TK_COMMENT        94
-#define TK_BOOL           95
-#define TK_TINYINT        96
-#define TK_SMALLINT       97
-#define TK_INT            98
-#define TK_INTEGER        99
-#define TK_BIGINT         100
-#define TK_FLOAT          101
-#define TK_DOUBLE         102
-#define TK_BINARY         103
-#define TK_TIMESTAMP      104
-#define TK_NCHAR          105
-#define TK_UNSIGNED       106
-#define TK_JSON           107
-#define TK_VARCHAR        108
-#define TK_MEDIUMBLOB     109
-#define TK_BLOB           110
-#define TK_VARBINARY      111
-#define TK_DECIMAL        112
-#define TK_MAX_DELAY      113
-#define TK_WATERMARK      114
-#define TK_ROLLUP         115
-#define TK_TTL            116
-#define TK_SMA            117
-#define TK_FIRST          118
-#define TK_LAST           119
-#define TK_SHOW           120
-#define TK_DATABASES      121
-#define TK_TABLES         122
-#define TK_STABLES        123
-#define TK_MNODES         124
-#define TK_MODULES        125
-#define TK_QNODES         126
-#define TK_FUNCTIONS      127
-#define TK_INDEXES        128
-#define TK_ACCOUNTS       129
-#define TK_APPS           130
-#define TK_CONNECTIONS    131
-#define TK_LICENCE        132
-#define TK_GRANTS         133
-#define TK_QUERIES        134
-#define TK_SCORES         135
-#define TK_TOPICS         136
-#define TK_VARIABLES      137
-#define TK_BNODES         138
-#define TK_SNODES         139
-#define TK_CLUSTER        140
-#define TK_TRANSACTIONS   141
-#define TK_DISTRIBUTED    142
-#define TK_CONSUMERS      143
-#define TK_SUBSCRIPTIONS  144
-#define TK_LIKE           145
-#define TK_INDEX          146
-#define TK_FUNCTION       147
-#define TK_INTERVAL       148
-#define TK_TOPIC          149
-#define TK_AS             150
-#define TK_WITH           151
-#define TK_META           152
-#define TK_CONSUMER       153
-#define TK_GROUP          154
-#define TK_DESC           155
-#define TK_DESCRIBE       156
-#define TK_RESET          157
-#define TK_QUERY          158
-#define TK_CACHE          159
-#define TK_EXPLAIN        160
-#define TK_ANALYZE        161
-#define TK_VERBOSE        162
-#define TK_NK_BOOL        163
-#define TK_RATIO          164
-#define TK_NK_FLOAT       165
-#define TK_COMPACT        166
-#define TK_VNODES         167
-#define TK_IN             168
-#define TK_OUTPUTTYPE     169
-#define TK_AGGREGATE      170
-#define TK_BUFSIZE        171
-#define TK_STREAM         172
-#define TK_INTO           173
-#define TK_TRIGGER        174
-#define TK_AT_ONCE        175
-#define TK_WINDOW_CLOSE   176
-#define TK_IGNORE         177
-#define TK_EXPIRED        178
-#define TK_KILL           179
-#define TK_CONNECTION     180
-#define TK_TRANSACTION    181
-#define TK_BALANCE        182
-#define TK_VGROUP         183
-#define TK_MERGE          184
-#define TK_REDISTRIBUTE   185
-#define TK_SPLIT          186
-#define TK_SYNCDB         187
-#define TK_DELETE         188
-#define TK_NULL           189
-#define TK_NK_QUESTION    190
-#define TK_NK_ARROW       191
-#define TK_ROWTS          192
-#define TK_TBNAME         193
-#define TK_QSTARTTS       194
-#define TK_QENDTS         195
-#define TK_WSTARTTS       196
-#define TK_WENDTS         197
-#define TK_WDURATION      198
-#define TK_CAST           199
-#define TK_NOW            200
-#define TK_TODAY          201
-#define TK_TIMEZONE       202
-#define TK_CLIENT_VERSION 203
-#define TK_SERVER_VERSION 204
-#define TK_SERVER_STATUS  205
-#define TK_CURRENT_USER   206
-#define TK_COUNT          207
-#define TK_LAST_ROW       208
-#define TK_BETWEEN        209
-#define TK_IS             210
-#define TK_NK_LT          211
-#define TK_NK_GT          212
-#define TK_NK_LE          213
-#define TK_NK_GE          214
-#define TK_NK_NE          215
-#define TK_MATCH          216
-#define TK_NMATCH         217
-#define TK_CONTAINS       218
-#define TK_JOIN           219
-#define TK_INNER          220
-#define TK_SELECT         221
-#define TK_DISTINCT       222
-#define TK_WHERE          223
-#define TK_PARTITION      224
-#define TK_BY             225
-#define TK_SESSION        226
-#define TK_STATE_WINDOW   227
-#define TK_SLIDING        228
-#define TK_FILL           229
-#define TK_VALUE          230
-#define TK_NONE           231
-#define TK_PREV           232
-#define TK_LINEAR         233
-#define TK_NEXT           234
-#define TK_HAVING         235
-#define TK_RANGE          236
-#define TK_EVERY          237
-#define TK_ORDER          238
-#define TK_SLIMIT         239
-#define TK_SOFFSET        240
-#define TK_LIMIT          241
-#define TK_OFFSET         242
-#define TK_ASC            243
-#define TK_NULLS          244
-#define TK_ID             245
-#define TK_NK_BITNOT      246
-#define TK_INSERT         247
-#define TK_VALUES         248
-#define TK_IMPORT         249
-#define TK_NK_SEMI        250
-#define TK_FILE           251
+#define TK_FLUSH          58
+#define TK_IF             59
+#define TK_NOT            60
+#define TK_EXISTS         61
+#define TK_BUFFER         62
+#define TK_CACHELAST      63
+#define TK_COMP           64
+#define TK_DURATION       65
+#define TK_NK_VARIABLE    66
+#define TK_FSYNC          67
+#define TK_MAXROWS        68
+#define TK_MINROWS        69
+#define TK_KEEP           70
+#define TK_PAGES          71
+#define TK_PAGESIZE       72
+#define TK_PRECISION      73
+#define TK_REPLICA        74
+#define TK_STRICT         75
+#define TK_WAL            76
+#define TK_VGROUPS        77
+#define TK_SINGLE_STABLE  78
+#define TK_RETENTIONS     79
+#define TK_SCHEMALESS     80
+#define TK_NK_COLON       81
+#define TK_TABLE          82
+#define TK_NK_LP          83
+#define TK_NK_RP          84
+#define TK_STABLE         85
+#define TK_ADD            86
+#define TK_COLUMN         87
+#define TK_MODIFY         88
+#define TK_RENAME         89
+#define TK_TAG            90
+#define TK_SET            91
+#define TK_NK_EQ          92
+#define TK_USING          93
+#define TK_TAGS           94
+#define TK_COMMENT        95
+#define TK_BOOL           96
+#define TK_TINYINT        97
+#define TK_SMALLINT       98
+#define TK_INT            99
+#define TK_INTEGER        100
+#define TK_BIGINT         101
+#define TK_FLOAT          102
+#define TK_DOUBLE         103
+#define TK_BINARY         104
+#define TK_TIMESTAMP      105
+#define TK_NCHAR          106
+#define TK_UNSIGNED       107
+#define TK_JSON           108
+#define TK_VARCHAR        109
+#define TK_MEDIUMBLOB     110
+#define TK_BLOB           111
+#define TK_VARBINARY      112
+#define TK_DECIMAL        113
+#define TK_MAX_DELAY      114
+#define TK_WATERMARK      115
+#define TK_ROLLUP         116
+#define TK_TTL            117
+#define TK_SMA            118
+#define TK_FIRST          119
+#define TK_LAST           120
+#define TK_SHOW           121
+#define TK_DATABASES      122
+#define TK_TABLES         123
+#define TK_STABLES        124
+#define TK_MNODES         125
+#define TK_MODULES        126
+#define TK_QNODES         127
+#define TK_FUNCTIONS      128
+#define TK_INDEXES        129
+#define TK_ACCOUNTS       130
+#define TK_APPS           131
+#define TK_CONNECTIONS    132
+#define TK_LICENCE        133
+#define TK_GRANTS         134
+#define TK_QUERIES        135
+#define TK_SCORES         136
+#define TK_TOPICS         137
+#define TK_VARIABLES      138
+#define TK_BNODES         139
+#define TK_SNODES         140
+#define TK_CLUSTER        141
+#define TK_TRANSACTIONS   142
+#define TK_DISTRIBUTED    143
+#define TK_CONSUMERS      144
+#define TK_SUBSCRIPTIONS  145
+#define TK_LIKE           146
+#define TK_INDEX          147
+#define TK_FUNCTION       148
+#define TK_INTERVAL       149
+#define TK_TOPIC          150
+#define TK_AS             151
+#define TK_WITH           152
+#define TK_META           153
+#define TK_CONSUMER       154
+#define TK_GROUP          155
+#define TK_DESC           156
+#define TK_DESCRIBE       157
+#define TK_RESET          158
+#define TK_QUERY          159
+#define TK_CACHE          160
+#define TK_EXPLAIN        161
+#define TK_ANALYZE        162
+#define TK_VERBOSE        163
+#define TK_NK_BOOL        164
+#define TK_RATIO          165
+#define TK_NK_FLOAT       166
+#define TK_COMPACT        167
+#define TK_VNODES         168
+#define TK_IN             169
+#define TK_OUTPUTTYPE     170
+#define TK_AGGREGATE      171
+#define TK_BUFSIZE        172
+#define TK_STREAM         173
+#define TK_INTO           174
+#define TK_TRIGGER        175
+#define TK_AT_ONCE        176
+#define TK_WINDOW_CLOSE   177
+#define TK_IGNORE         178
+#define TK_EXPIRED        179
+#define TK_KILL           180
+#define TK_CONNECTION     181
+#define TK_TRANSACTION    182
+#define TK_BALANCE        183
+#define TK_VGROUP         184
+#define TK_MERGE          185
+#define TK_REDISTRIBUTE   186
+#define TK_SPLIT          187
+#define TK_SYNCDB         188
+#define TK_DELETE         189
+#define TK_INSERT         190
+#define TK_NULL           191
+#define TK_NK_QUESTION    192
+#define TK_NK_ARROW       193
+#define TK_ROWTS          194
+#define TK_TBNAME         195
+#define TK_QSTARTTS       196
+#define TK_QENDTS         197
+#define TK_WSTARTTS       198
+#define TK_WENDTS         199
+#define TK_WDURATION      200
+#define TK_CAST           201
+#define TK_NOW            202
+#define TK_TODAY          203
+#define TK_TIMEZONE       204
+#define TK_CLIENT_VERSION 205
+#define TK_SERVER_VERSION 206
+#define TK_SERVER_STATUS  207
+#define TK_CURRENT_USER   208
+#define TK_COUNT          209
+#define TK_LAST_ROW       210
+#define TK_BETWEEN        211
+#define TK_IS             212
+#define TK_NK_LT          213
+#define TK_NK_GT          214
+#define TK_NK_LE          215
+#define TK_NK_GE          216
+#define TK_NK_NE          217
+#define TK_MATCH          218
+#define TK_NMATCH         219
+#define TK_CONTAINS       220
+#define TK_JOIN           221
+#define TK_INNER          222
+#define TK_SELECT         223
+#define TK_DISTINCT       224
+#define TK_WHERE          225
+#define TK_PARTITION      226
+#define TK_BY             227
+#define TK_SESSION        228
+#define TK_STATE_WINDOW   229
+#define TK_SLIDING        230
+#define TK_FILL           231
+#define TK_VALUE          232
+#define TK_NONE           233
+#define TK_PREV           234
+#define TK_LINEAR         235
+#define TK_NEXT           236
+#define TK_HAVING         237
+#define TK_RANGE          238
+#define TK_EVERY          239
+#define TK_ORDER          240
+#define TK_SLIMIT         241
+#define TK_SOFFSET        242
+#define TK_LIMIT          243
+#define TK_OFFSET         244
+#define TK_ASC            245
+#define TK_NULLS          246
+#define TK_ID             247
+#define TK_NK_BITNOT      248
+#define TK_VALUES         249
+#define TK_IMPORT         250
+#define TK_NK_SEMI        251
+#define TK_FILE           252
 
 #define TK_NK_SPACE   300
 #define TK_NK_COMMENT 301

include/libs/command/command.h

@@ -13,6 +13,9 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#ifndef TDENGINE_COMMAND_H
+#define TDENGINE_COMMAND_H
+
 #include "cmdnodes.h"
 #include "tmsg.h"
 #include "plannodes.h"
@@ -27,4 +30,4 @@ int32_t qExecExplainEnd(SExplainCtx *pCtx, SRetrieveTableRsp **pRsp);
 int32_t qExplainUpdateExecInfo(SExplainCtx *pCtx, SExplainRsp *pRspMsg, int32_t groupId, SRetrieveTableRsp **pRsp);
 void    qExplainFreeCtx(SExplainCtx *pCtx);
 
-
+#endif

include/libs/executor/executor.h

@@ -36,6 +36,8 @@ typedef struct SReadHandle {
   void*   vnode;
   void*   mnd;
   SMsgCb* pMsgCb;
+
+//  int8_t  initTsdbReader;
   bool    tqReader;
 } SReadHandle;
 

include/libs/index/index.h

@@ -227,7 +227,7 @@ int32_t doFilterTag(SNode* pFilterNode, SIndexMetaArg* metaArg, SArray* result,
  * destory index env
  *
  */
-void indexCleanUp();
+void indexCleanup();
 
 #ifdef __cplusplus
 }

include/libs/nodes/cmdnodes.h

@@ -97,6 +97,11 @@ typedef struct SAlterDatabaseStmt {
   SDatabaseOptions* pOptions;
 } SAlterDatabaseStmt;
 
+typedef struct SFlushDatabaseStmt {
+  ENodeType type;
+  char      dbName[TSDB_DB_NAME_LEN];
+} SFlushDatabaseStmt;
+
 typedef struct STableOptions {
   ENodeType  type;
   bool       commentNull;

include/libs/nodes/nodes.h

@@ -111,6 +111,7 @@ typedef enum ENodeType {
   QUERY_NODE_CREATE_DATABASE_STMT,
   QUERY_NODE_DROP_DATABASE_STMT,
   QUERY_NODE_ALTER_DATABASE_STMT,
+  QUERY_NODE_FLUSH_DATABASE_STMT,
   QUERY_NODE_CREATE_TABLE_STMT,
   QUERY_NODE_CREATE_SUBTABLE_CLAUSE,
   QUERY_NODE_CREATE_MULTI_TABLE_STMT,
@@ -194,6 +195,7 @@ typedef enum ENodeType {
   QUERY_NODE_KILL_QUERY_STMT,
   QUERY_NODE_KILL_TRANSACTION_STMT,
   QUERY_NODE_DELETE_STMT,
+  QUERY_NODE_INSERT_STMT,
   QUERY_NODE_QUERY,
 
   // logic plan node
@@ -247,6 +249,7 @@ typedef enum ENodeType {
   QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC,
   QUERY_NODE_PHYSICAL_PLAN_DISPATCH,
   QUERY_NODE_PHYSICAL_PLAN_INSERT,
+  QUERY_NODE_PHYSICAL_PLAN_QUERY_INSERT,
   QUERY_NODE_PHYSICAL_PLAN_DELETE,
   QUERY_NODE_PHYSICAL_SUBPLAN,
   QUERY_NODE_PHYSICAL_PLAN

include/libs/nodes/plannodes.h

@@ -128,9 +128,12 @@ typedef struct SVnodeModifyLogicNode {
   SVgDataBlocks*   pVgDataBlocks;
   SNode*           pAffectedRows;  // SColumnNode
   uint64_t         tableId;
+  uint64_t         stableId;
   int8_t           tableType;  // table type
   char             tableFName[TSDB_TABLE_FNAME_LEN];
   STimeWindow      deleteTimeRange;
+  SVgroupsInfo*    pVgroupList;
+  SNodeList*       pInsertCols;
 } SVnodeModifyLogicNode;
 
 typedef struct SExchangeLogicNode {
@@ -456,6 +459,17 @@ typedef struct SDataInserterNode {
   char*         pData;
 } SDataInserterNode;
 
+typedef struct SQueryInserterNode {
+  SDataSinkNode sink;
+  SNodeList*    pCols;
+  uint64_t      tableId;
+  uint64_t      stableId;
+  int8_t        tableType;  // table type
+  char          tableFName[TSDB_TABLE_FNAME_LEN];
+  int32_t       vgId;
+  SEpSet        epSet;
+} SQueryInserterNode;
+
 typedef struct SDataDeleterNode {
   SDataSinkNode sink;
   uint64_t      tableId;

include/libs/nodes/querynodes.h

@@ -302,6 +302,14 @@ typedef struct SDeleteStmt {
   bool        deleteZeroRows;
 } SDeleteStmt;
 
+typedef struct SInsertStmt {
+  ENodeType  type;  // QUERY_NODE_INSERT_STMT
+  SNode*     pTable;
+  SNodeList* pCols;
+  SNode*     pQuery;
+  uint8_t    precision;
+} SInsertStmt;
+
 typedef enum {
   PAYLOAD_TYPE_KV = 0,
   PAYLOAD_TYPE_RAW = 1,

include/libs/parser/parser.h

@@ -56,7 +56,7 @@ typedef struct SParseContext {
 } SParseContext;
 
 int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery);
-bool    qIsInsertSql(const char* pStr, size_t length);
+bool    qIsInsertValuesSql(const char* pStr, size_t length);
 
 // for async mode
 int32_t qParseSqlSyntax(SParseContext* pCxt, SQuery** pQuery, struct SCatalogReq* pCatalogReq);

include/libs/qcom/query.h

@@ -29,12 +29,13 @@ extern "C" {
 
 typedef enum {
   JOB_TASK_STATUS_NULL = 0,
-  JOB_TASK_STATUS_NOT_START = 1,
-  JOB_TASK_STATUS_EXECUTING,
-  JOB_TASK_STATUS_PARTIAL_SUCCEED,
-  JOB_TASK_STATUS_SUCCEED,
-  JOB_TASK_STATUS_FAILED,
-  JOB_TASK_STATUS_DROPPING,
+  JOB_TASK_STATUS_INIT,
+  JOB_TASK_STATUS_EXEC,
+  JOB_TASK_STATUS_PART_SUCC,
+  JOB_TASK_STATUS_SUCC,
+  JOB_TASK_STATUS_FAIL,
+  JOB_TASK_STATUS_DROP,
+  JOB_TASK_STATUS_MAX,
 } EJobTaskType;
 
 typedef enum {
@@ -59,10 +60,6 @@ typedef struct STableComInfo {
   int32_t  rowSize;       // row size of the schema
 } STableComInfo;
 
-typedef struct SQueryExecRes {
-  int32_t msgType;
-  void*   res;
-} SQueryExecRes;
 
 typedef struct SIndexMeta {
 #if defined(WINDOWS) || defined(_TD_DARWIN_64)
@@ -71,6 +68,13 @@ typedef struct SIndexMeta {
 
 } SIndexMeta;
 
+typedef struct SExecResult {
+  int32_t         code;
+  uint64_t        numOfRows;
+  int32_t         msgType;
+  void*           res;
+} SExecResult;
+
 typedef struct STbVerInfo {
   char    tbFName[TSDB_TABLE_FNAME_LEN];
   int32_t sversion;
@@ -210,7 +214,7 @@ char*   jobTaskStatusStr(int32_t status);
 
 SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name);
 
-void    destroyQueryExecRes(SQueryExecRes* pRes);
+void    destroyQueryExecRes(SExecResult* pRes);
 int32_t dataConverToStr(char* str, int type, void* buf, int32_t bufSize, int32_t* len);
 char*   parseTagDatatoJson(void* p);
 int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst);

include/libs/qworker/qworker.h

@@ -20,9 +20,9 @@
 extern "C" {
 #endif
 
+#include "executor.h"
 #include "tmsgcb.h"
 #include "trpc.h"
-#include "executor.h"
 
 enum {
   NODE_TYPE_VNODE = 1,
@@ -31,13 +31,6 @@ enum {
   NODE_TYPE_MNODE,
 };
 
-typedef struct SDeleteRes {
-  uint64_t suid;
-  SArray*  uidList;
-  int64_t  skey;
-  int64_t  ekey;
-} SDeleteRes;
-
 typedef struct SQWorkerCfg {
   uint32_t maxSchedulerNum;
   uint32_t maxTaskNum;
@@ -46,19 +39,19 @@ typedef struct SQWorkerCfg {
 
 typedef struct {
   uint64_t cacheDataSize;
-  
+
   uint64_t queryProcessed;
   uint64_t cqueryProcessed;
   uint64_t fetchProcessed;
   uint64_t dropProcessed;
   uint64_t hbProcessed;
   uint64_t deleteProcessed;
-  
+
   uint64_t numOfQueryInQueue;
   uint64_t numOfFetchInQueue;
   uint64_t timeInQueryQueue;
   uint64_t timeInFetchQueue;
-  
+
   uint64_t numOfErrors;
 } SQWorkerStat;
 
@@ -82,7 +75,7 @@ int32_t qWorkerProcessDropMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int6
 
 int32_t qWorkerProcessHbMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, int64_t ts);
 
-int32_t qWorkerProcessDeleteMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SRpcMsg *pRsp, SDeleteRes *pRes);
+int32_t qWorkerProcessDeleteMsg(void *node, void *qWorkerMgmt, SRpcMsg *pMsg, SDeleteRes *pRes);
 
 void qWorkerDestroy(void **qWorkerMgmt);
 

include/libs/scheduler/scheduler.h

@@ -53,12 +53,6 @@ typedef struct SQueryProfileSummary {
   uint64_t resultSize;   // generated result size in Kb.
 } SQueryProfileSummary;
 
-typedef struct SQueryResult {
-  int32_t         code;
-  uint64_t        numOfRows;
-  SQueryExecRes   res;
-} SQueryResult;
-
 typedef struct STaskInfo {
   SQueryNodeAddr addr;
   SSubQueryMsg  *msg;
@@ -69,50 +63,34 @@ typedef struct SSchdFetchParam {
   int32_t* code;
 } SSchdFetchParam;
 
-typedef void (*schedulerExecFp)(SQueryResult* pResult, void* param, int32_t code);
+typedef void (*schedulerExecFp)(SExecResult* pResult, void* param, int32_t code);
 typedef void (*schedulerFetchFp)(void* pResult, void* param, int32_t code);
 typedef bool (*schedulerChkKillFp)(void* param);
 
 typedef struct SSchedulerReq {
+  bool                  syncReq;
   SRequestConnInfo     *pConn;
   SArray               *pNodeList;
   SQueryPlan           *pDag;
   const char           *sql;
   int64_t               startTs;
   schedulerExecFp       execFp;
-  void*                 execParam;
+  schedulerFetchFp      fetchFp;
+  void*                 cbParam;
   schedulerChkKillFp    chkKillFp;
   void*                 chkKillParam;
+  SExecResult*          pExecRes;
+  void**                pFetchRes;
 } SSchedulerReq;
 
 
 int32_t schedulerInit(SSchedulerCfg *cfg);
 
-/**
- * Process the query job, generated according to the query physical plan.
- * This is a synchronized API, and is also thread-safety.
- * @param nodeList  Qnode/Vnode address list, element is SQueryNodeAddr
- * @return
- */
-int32_t schedulerExecJob(SSchedulerReq *pReq, int64_t *pJob, SQueryResult *pRes);
-
-/**
- * Process the query job, generated according to the query physical plan.
- * This is a asynchronized API, and is also thread-safety.
- * @param pNodeList  Qnode/Vnode address list, element is SQueryNodeAddr
- * @return
- */
-  int32_t schedulerAsyncExecJob(SSchedulerReq *pReq, int64_t *pJob);
+int32_t schedulerExecJob(SSchedulerReq *pReq, int64_t *pJob);
 
-/**
- * Fetch query result from the remote query executor
- * @param pJob
- * @param data
- * @return
- */
-int32_t schedulerFetchRows(int64_t job, void **data);
+int32_t schedulerFetchRows(int64_t jobId, SSchedulerReq *pReq);
 
-void schedulerAsyncFetchRows(int64_t job, schedulerFetchFp fp, void* param);
+void schedulerFetchRowsA(int64_t job, schedulerFetchFp fp, void* param);
 
 int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub);
 
@@ -134,7 +112,7 @@ void schedulerFreeJob(int64_t* job, int32_t errCode);
 
 void schedulerDestroy(void);
 
-void schdExecCallback(SQueryResult* pResult, void* param, int32_t code);
+void schdExecCallback(SExecResult* pResult, void* param, int32_t code);
 
 #ifdef __cplusplus
 }

include/libs/sync/sync.h

@@ -26,7 +26,7 @@ extern "C" {
 
 extern bool gRaftDetailLog;
 
-#define SYNC_MAX_BATCH_SIZE 100
+#define SYNC_MAX_BATCH_SIZE 500
 #define SYNC_INDEX_BEGIN 0
 #define SYNC_INDEX_INVALID -1
 #define SYNC_TERM_INVALID 0xFFFFFFFFFFFFFFFF
@@ -215,6 +215,7 @@ int32_t     syncProposeBatch(int64_t rid, SRpcMsg* pMsgArr, bool* pIsWeakArr, in
 bool        syncEnvIsStart();
 const char* syncStr(ESyncState state);
 bool        syncIsRestoreFinish(int64_t rid);
+int32_t     syncGetSnapshotByIndex(int64_t rid, SyncIndex index, SSnapshot* pSnapshot);
 
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pNewCfg);
 

include/libs/transport/trpc.h

@@ -27,7 +27,7 @@ extern "C" {
 
 #define TAOS_CONN_SERVER 0
 #define TAOS_CONN_CLIENT 1
-#define IsReq(pMsg)      (pMsg->msgType & 1U)
+#define IsReq(pMsg) (pMsg->msgType & 1U)
 
 extern int32_t tsRpcHeadSize;
 
@@ -35,12 +35,13 @@ typedef struct {
   uint32_t clientIp;
   uint16_t clientPort;
   int64_t  applyIndex;
+  uint64_t applyTerm;
   char     user[TSDB_USER_LEN];
 } SRpcConnInfo;
 
 typedef struct SRpcHandleInfo {
   // rpc info
-  void *   handle;         // rpc handle returned to app
+  void    *handle;         // rpc handle returned to app
   int64_t  refId;          // refid, used by server
   int32_t  noResp;         // has response or not(default 0, 0: resp, 1: no resp);
   int32_t  persistHandle;  // persist handle or not
@@ -53,7 +54,7 @@ typedef struct SRpcHandleInfo {
   void *node;     // node mgmt handle
 
   // resp info
-  void *  rsp;
+  void   *rsp;
   int32_t rspLen;
 
   // conn info
@@ -62,7 +63,7 @@ typedef struct SRpcHandleInfo {
 
 typedef struct SRpcMsg {
   tmsg_t         msgType;
-  void *         pCont;
+  void          *pCont;
   int32_t        contLen;
   int32_t        code;
   SRpcHandleInfo info;
@@ -74,7 +75,7 @@ typedef bool (*RpcRfp)(int32_t code, tmsg_t msgType);
 typedef struct SRpcInit {
   char     localFqdn[TSDB_FQDN_LEN];
   uint16_t localPort;     // local port
-  char *   label;         // for debug purpose
+  char    *label;         // for debug purpose
   int32_t  numOfThreads;  // number of threads to handle connections
   int32_t  sessions;      // number of sessions allowed
   int8_t   connType;      // TAOS_CONN_UDP, TAOS_CONN_TCPC, TAOS_CONN_TCPS
@@ -99,12 +100,12 @@ typedef struct {
 
 typedef struct {
   int32_t msgType;
-  void *  val;
+  void   *val;
   int32_t (*clone)(void *src, void **dst);
 } SRpcBrokenlinkVal;
 
 typedef struct {
-  SHashObj *        args;
+  SHashObj         *args;
   SRpcBrokenlinkVal brokenVal;
   void (*freeFunc)(const void *arg);
 } SRpcCtx;

include/os/osMemory.h

@@ -32,7 +32,7 @@ extern "C" {
 void *taosMemoryMalloc(int32_t size);
 void *taosMemoryCalloc(int32_t num, int32_t size);
 void *taosMemoryRealloc(void *ptr, int32_t size);
-void *taosMemoryStrDup(void *ptr);
+void *taosMemoryStrDup(const char *ptr);
 void taosMemoryFree(void *ptr);
 int32_t taosMemorySize(void *ptr);
 void taosPrintBackTrace();

include/util/tmallocator.h → include/util/tRealloc.h

@@ -13,8 +13,8 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
-#ifndef _TD_UTIL_MALLOCATOR_H_
-#define _TD_UTIL_MALLOCATOR_H_
+#ifndef _TD_UTIL_TREALLOC_H_
+#define _TD_UTIL_TREALLOC_H_
 
 #include "os.h"
 
@@ -22,34 +22,43 @@
 extern "C" {
 #endif
 
-// Memory allocator
-#define TD_MEM_ALCT(TYPE)                           \
-  struct {                                          \
-    void *(*malloc_)(struct TYPE *, uint64_t size); \
-    void (*free_)(struct TYPE *, void *ptr);        \
+static FORCE_INLINE int32_t tRealloc(uint8_t **ppBuf, int64_t size) {
+  int32_t  code = 0;
+  int64_t  bsize = 0;
+  uint8_t *pBuf;
+
+  if (*ppBuf) {
+    bsize = *(int64_t *)((*ppBuf) - sizeof(int64_t));
+  }
+
+  if (bsize >= size) goto _exit;
+
+  if (bsize == 0) bsize = 64;
+  while (bsize < size) {
+    bsize *= 2;
   }
-#define TD_MA_MALLOC_FUNC(TMA) (TMA)->malloc_
-#define TD_MA_FREE_FUNC(TMA)   (TMA)->free_
 
-#define TD_MA_MALLOC(TMA, SIZE) (*((TMA)->malloc_))(TMA, (SIZE))
-#define TD_MA_FREE(TMA, PTR)    (*((TMA)->free_))(TMA, (PTR))
+  pBuf = (uint8_t *)taosMemoryRealloc(*ppBuf ? (*ppBuf) - sizeof(int64_t) : *ppBuf, bsize + sizeof(int64_t));
+  if (pBuf == NULL) {
+    code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
 
-typedef struct SMemAllocator {
-  void *impl;
-  TD_MEM_ALCT(SMemAllocator);
-} SMemAllocator;
+  *(int64_t *)pBuf = bsize;
+  *ppBuf = pBuf + sizeof(int64_t);
 
-#define tMalloc(pMA, SIZE) TD_MA_MALLOC(PMA, SIZE)
-#define tFree(pMA, PTR)    TD_MA_FREE(PMA, PTR)
+_exit:
+  return code;
+}
 
-typedef struct SMemAllocatorFactory {
-  void *impl;
-  SMemAllocator *(*create)(struct SMemAllocatorFactory *);
-  void (*destroy)(struct SMemAllocatorFactory *, SMemAllocator *);
-} SMemAllocatorFactory;
+static FORCE_INLINE void tFree(uint8_t *pBuf) {
+  if (pBuf) {
+    taosMemoryFree(pBuf - sizeof(int64_t));
+  }
+}
 
 #ifdef __cplusplus
 }
 #endif
 
-#endif /*_TD_UTIL_MALLOCATOR_H_*/
+#endif /*_TD_UTIL_TREALLOC_H_*/
\ No newline at end of file

include/util/taoserror.h

@@ -72,6 +72,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_INVALID_TIMESTAMP             TAOS_DEF_ERROR_CODE(0, 0x0030)
 #define TSDB_CODE_MSG_DECODE_ERROR              TAOS_DEF_ERROR_CODE(0, 0x0031)
 #define TSDB_CODE_NO_AVAIL_DISK                 TAOS_DEF_ERROR_CODE(0, 0x0032)
+#define TSDB_CODE_NOT_FOUND                     TAOS_DEF_ERROR_CODE(0, 0x0033)
 
 #define TSDB_CODE_REF_NO_MEMORY                 TAOS_DEF_ERROR_CODE(0, 0x0040)
 #define TSDB_CODE_REF_FULL                      TAOS_DEF_ERROR_CODE(0, 0x0041)
@@ -389,10 +390,10 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_QRY_TASK_MSG_ERROR            TAOS_DEF_ERROR_CODE(0, 0x0719)
 #define TSDB_CODE_QRY_JOB_FREED                 TAOS_DEF_ERROR_CODE(0, 0x071A)
 #define TSDB_CODE_QRY_TASK_STATUS_ERROR         TAOS_DEF_ERROR_CODE(0, 0x071B)
-//json
 #define TSDB_CODE_QRY_JSON_IN_ERROR             TAOS_DEF_ERROR_CODE(0, 0x071C)
 #define TSDB_CODE_QRY_JSON_NOT_SUPPORT_ERROR    TAOS_DEF_ERROR_CODE(0, 0x071D)
 #define TSDB_CODE_QRY_JSON_IN_GROUP_ERROR       TAOS_DEF_ERROR_CODE(0, 0x071E)
+#define TSDB_CODE_QRY_JOB_NOT_EXIST             TAOS_DEF_ERROR_CODE(0, 0x071F)
 
 // grant
 #define TSDB_CODE_GRANT_EXPIRED                 TAOS_DEF_ERROR_CODE(0, 0x0800)

source/client/inc/clientInt.h

@@ -156,7 +156,7 @@ typedef struct SResultColumn {
 } SResultColumn;
 
 typedef struct SReqResultInfo {
-  SQueryExecRes  execRes;
+  SExecResult    execRes;
   const char*    pRspMsg;
   const char*    pData;
   TAOS_FIELD*    fields;      // todo, column names are not needed.

source/client/src/clientImpl.c

@@ -627,22 +627,26 @@ _return:
 int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList) {
   void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
 
-  SQueryResult     res = {0};
+  SExecResult     res = {0};
   SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
                            .requestId = pRequest->requestId,
                            .requestObjRefId = pRequest->self};
-  SSchedulerReq    req = {.pConn = &conn,
-                       .pNodeList = pNodeList,
-                       .pDag = pDag,
-                       .sql = pRequest->sqlstr,
-                       .startTs = pRequest->metric.start,
-                       .execFp = NULL,
-                       .execParam = NULL,
-                       .chkKillFp = chkRequestKilled,
-                       .chkKillParam = (void*)pRequest->self};
-
-  int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob, &res);
-  pRequest->body.resInfo.execRes = res.res;
+  SSchedulerReq    req = {
+    .syncReq = true,
+    .pConn = &conn,
+    .pNodeList = pNodeList,
+    .pDag = pDag,
+    .sql = pRequest->sqlstr,
+    .startTs = pRequest->metric.start,
+    .execFp = NULL,
+    .cbParam = NULL,
+    .chkKillFp = chkRequestKilled,
+    .chkKillParam = (void*)pRequest->self,
+    .pExecRes = &res,
+  };
+
+  int32_t code = schedulerExecJob(&req, &pRequest->body.queryJob);
+  memcpy(&pRequest->body.resInfo.execRes, &res, sizeof(res));
 
   if (code != TSDB_CODE_SUCCESS) {
     schedulerFreeJob(&pRequest->body.queryJob, 0);
@@ -753,7 +757,7 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
   }
 
   SEpSet         epset = getEpSet_s(&pAppInfo->mgmtEp);
-  SQueryExecRes* pRes = &pRequest->body.resInfo.execRes;
+  SExecResult* pRes = &pRequest->body.resInfo.execRes;
 
   switch (pRes->msgType) {
     case TDMT_VND_ALTER_TABLE:
@@ -779,10 +783,10 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
   return code;
 }
 
-void schedulerExecCb(SQueryResult* pResult, void* param, int32_t code) {
+void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
   SRequestObj* pRequest = (SRequestObj*)param;
   pRequest->code = code;
-  pRequest->body.resInfo.execRes = pResult->res;
+  memcpy(&pRequest->body.resInfo.execRes, pResult, sizeof(*pResult));
 
   if (TDMT_VND_SUBMIT == pRequest->type || TDMT_VND_DELETE == pRequest->type ||
       TDMT_VND_CREATE_TABLE == pRequest->type) {
@@ -939,16 +943,20 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
 
         SRequestConnInfo conn = {
             .pTrans = pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
-        SSchedulerReq req = {.pConn = &conn,
-                             .pNodeList = pNodeList,
-                             .pDag = pDag,
-                             .sql = pRequest->sqlstr,
-                             .startTs = pRequest->metric.start,
-                             .execFp = schedulerExecCb,
-                             .execParam = pRequest,
-                             .chkKillFp = chkRequestKilled,
-                             .chkKillParam = (void*)pRequest->self};
-        code = schedulerAsyncExecJob(&req, &pRequest->body.queryJob);
+        SSchedulerReq req = {
+          .syncReq = false,
+          .pConn = &conn,
+          .pNodeList = pNodeList,
+          .pDag = pDag,
+          .sql = pRequest->sqlstr,
+          .startTs = pRequest->metric.start,
+          .execFp = schedulerExecCb,
+          .cbParam = pRequest,
+          .chkKillFp = chkRequestKilled,
+          .chkKillParam = (void*)pRequest->self,
+          .pExecRes = NULL,
+        };
+        code = schedulerExecJob(&req, &pRequest->body.queryJob);
         taosArrayDestroy(pNodeList);
       } else {
         tscDebug("0x%" PRIx64 " plan not executed, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
@@ -1387,7 +1395,11 @@ void* doFetchRows(SRequestObj* pRequest, bool setupOneRowPtr, bool convertUcs4)
     }
 
     SReqResultInfo* pResInfo = &pRequest->body.resInfo;
-    pRequest->code = schedulerFetchRows(pRequest->body.queryJob, (void**)&pResInfo->pData);
+    SSchedulerReq req = {
+      .syncReq = true,
+      .pFetchRes = (void**)&pResInfo->pData,
+    };
+    pRequest->code = schedulerFetchRows(pRequest->body.queryJob, &req);
     if (pRequest->code != TSDB_CODE_SUCCESS) {
       pResultInfo->numOfRows = 0;
       return NULL;

source/client/src/clientMain.c

@@ -843,35 +843,46 @@ void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
   pRequest->body.param = param;
 
   SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
-  if (taos_num_fields(pRequest) == 0) {
+
+  // this query has no results or error exists, return directly
+  if (taos_num_fields(pRequest) == 0 || pRequest->code != TSDB_CODE_SUCCESS) {
     pResultInfo->numOfRows = 0;
     pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
     return;
   }
 
-  if (pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) {
-    // All data has returned to App already, no need to try again
-    if (pResultInfo->completed) {
-      pResultInfo->numOfRows = 0;
-      pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
-      return;
-    }
+  // all data has returned to App already, no need to try again
+  if ((pResultInfo->pData == NULL || pResultInfo->current >= pResultInfo->numOfRows) && pResultInfo->completed) {
+    pResultInfo->numOfRows = 0;
+    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
+    return;
+  }
+
+  // it is a local executed query, no need to do async fetch
+  if (pResultInfo->current < pResultInfo->numOfRows && pRequest->body.queryJob == 0) {
+    pRequest->body.fetchFp(param, pRequest, pResultInfo->numOfRows);
+    return;
   }
 
-  schedulerAsyncFetchRows(pRequest->body.queryJob, fetchCallback, pRequest);
+  SSchedulerReq req = {
+    .syncReq = false,
+    .fetchFp = fetchCallback,
+    .cbParam = pRequest,
+  };
+  schedulerFetchRows(pRequest->body.queryJob, &req);
 }
 
 void taos_fetch_raw_block_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
   ASSERT(res != NULL && fp != NULL);
   ASSERT(TD_RES_QUERY(res));
-  SRequestObj *pRequest = res;
-
-  pRequest->body.resInfo.convertUcs4 = false;
 
+  SRequestObj *pRequest = res;
   SReqResultInfo *pResultInfo = &pRequest->body.resInfo;
 
   // set the current block is all consumed
   pResultInfo->current = pResultInfo->numOfRows;
+  pResultInfo->convertUcs4 = false;
+
   taos_fetch_rows_a(res, fp, param);
 }
 

source/client/src/clientMsgHandler.c

@@ -266,7 +266,7 @@ int32_t processAlterStbRsp(void* param, SDataBuf* pMsg, int32_t code) {
   }
 
   if (pRequest->body.queryFp != NULL) {
-    SQueryExecRes* pRes = &pRequest->body.resInfo.execRes;
+    SExecResult* pRes = &pRequest->body.resInfo.execRes;
 
     if (code == TSDB_CODE_SUCCESS) {
       SCatalog* pCatalog = NULL;

source/client/src/clientStmt.c

@@ -324,9 +324,9 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
 }
 
 int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks* pDataBlock, STableDataBlocks** newBlock, uint64_t uid) {
-  SEpSet      ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
-  SVgroupInfo vgInfo = {0};
-  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter, 
+  SEpSet           ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
+  SVgroupInfo      vgInfo = {0};
+  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
                            .requestId = pStmt->exec.pRequest->requestId,
                            .requestObjRefId = pStmt->exec.pRequest->self,
                            .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
@@ -391,13 +391,12 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     STMT_RET(stmtCleanBindInfo(pStmt));
   }
 
-  STableMeta* pTableMeta = NULL;
-  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter, 
+  STableMeta*      pTableMeta = NULL;
+  SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
                            .requestId = pStmt->exec.pRequest->requestId,
                            .requestObjRefId = pStmt->exec.pRequest->self,
                            .mgmtEps = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp)};
-  int32_t     code =
-      catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
+  int32_t          code = catalogGetTableMeta(pStmt->pCatalog, &conn, &pStmt->bInfo.sname, &pTableMeta);
   if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
     STMT_ERR_RET(stmtCleanBindInfo(pStmt));
 
@@ -849,7 +848,7 @@ int stmtIsInsert(TAOS_STMT* stmt, int* insert) {
   if (pStmt->sql.type) {
     *insert = (STMT_TYPE_INSERT == pStmt->sql.type || STMT_TYPE_MULTI_INSERT == pStmt->sql.type);
   } else {
-    *insert = qIsInsertSql(pStmt->sql.sqlStr, 0);
+    *insert = qIsInsertValuesSql(pStmt->sql.sqlStr, pStmt->sql.sqlLen);
   }
 
   return TSDB_CODE_SUCCESS;
@@ -861,7 +860,7 @@ int stmtGetTagFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
     STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
   }
-  
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
 
   if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
@@ -893,7 +892,7 @@ int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
     STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
   }
-  
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
 
   if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
@@ -919,7 +918,6 @@ int stmtGetColFields(TAOS_STMT* stmt, int* nums, TAOS_FIELD_E** fields) {
   return TSDB_CODE_SUCCESS;
 }
 
-
 int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
@@ -952,13 +950,13 @@ int stmtGetParamNum(TAOS_STMT* stmt, int* nums) {
   return TSDB_CODE_SUCCESS;
 }
 
-int stmtGetParam(TAOS_STMT *stmt, int idx, int *type, int *bytes) {
+int stmtGetParam(TAOS_STMT* stmt, int idx, int* type, int* bytes) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
     STMT_RET(TSDB_CODE_TSC_STMT_API_ERROR);
   }
-  
+
   STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_FETCH_FIELDS));
 
   if (pStmt->bInfo.needParse && pStmt->sql.runTimes && pStmt->sql.type > 0 &&
@@ -979,8 +977,8 @@ int stmtGetParam(TAOS_STMT *stmt, int idx, int *type, int *bytes) {
     STMT_ERR_RET(stmtParseSql(pStmt));
   }
 
-  int32_t nums = 0;
-  TAOS_FIELD_E *pField = NULL;
+  int32_t       nums = 0;
+  TAOS_FIELD_E* pField = NULL;
   STMT_ERR_RET(stmtFetchColFields(stmt, &nums, &pField));
   if (idx >= nums) {
     tscError("idx %d is too big", idx);

source/client/test/clientTests.cpp

@@ -669,13 +669,13 @@ TEST(testCase, projection_query_tables) {
   TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
   ASSERT_NE(pConn, nullptr);
 
-//  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 2");
-//  if (taos_errno(pRes) != 0) {
-//    printf("error in create db, reason:%s\n", taos_errstr(pRes));
-//  }
-//  taos_free_result(pRes);
+  TAOS_RES* pRes = taos_query(pConn, "create database if not exists abc1 vgroups 1");
+  if (taos_errno(pRes) != 0) {
+    printf("error in create db, reason:%s\n", taos_errstr(pRes));
+  }
+  taos_free_result(pRes);
 
-  TAOS_RES* pRes = taos_query(pConn, "use abc1");
+  pRes = taos_query(pConn, "use abc1");
   taos_free_result(pRes);
 
   pRes = taos_query(pConn, "create stable st1 (ts timestamp, k int) tags(a int)");
@@ -700,54 +700,55 @@ TEST(testCase, projection_query_tables) {
     printf("create table :%d\n", i);
     createNewTable(pConn, i);
   }
-//  pRes = taos_query(pConn, "select * from tu");
-//  if (taos_errno(pRes) != 0) {
-//    printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
-//    taos_free_result(pRes);
-//    ASSERT_TRUE(false);
-//  }
 
-//  TAOS_ROW    pRow = NULL;
-//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-//  int32_t     numOfFields = taos_num_fields(pRes);
-//
-//  char str[512] = {0};
-//  while ((pRow = taos_fetch_row(pRes)) != NULL) {
-//    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
-//    printf("%s\n", str);
-//  }
+  pRes = taos_query(pConn, "select * from tu");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
+    taos_free_result(pRes);
+    ASSERT_TRUE(false);
+  }
 
-//  taos_free_result(pRes);
+  TAOS_ROW    pRow = NULL;
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  int32_t     numOfFields = taos_num_fields(pRes);
+
+  char str[512] = {0};
+  while ((pRow = taos_fetch_row(pRes)) != NULL) {
+    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+    printf("%s\n", str);
+  }
+
+  taos_free_result(pRes);
   taos_close(pConn);
 }
 
-//TEST(testCase, projection_query_stables) {
-//  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
-//  ASSERT_NE(pConn, nullptr);
-//
-//  TAOS_RES* pRes = taos_query(pConn, "use abc1");
-//  taos_free_result(pRes);
-//
-//  pRes = taos_query(pConn, "select ts from st1");
-//  if (taos_errno(pRes) != 0) {
-//    printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
-//    taos_free_result(pRes);
-//    ASSERT_TRUE(false);
-//  }
-//
-//  TAOS_ROW    pRow = NULL;
-//  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
-//  int32_t     numOfFields = taos_num_fields(pRes);
-//
-//  char str[512] = {0};
-//  while ((pRow = taos_fetch_row(pRes)) != NULL) {
-//    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
-//    printf("%s\n", str);
-//  }
-//
-//  taos_free_result(pRes);
-//  taos_close(pConn);
-//}
+TEST(testCase, projection_query_stables) {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  TAOS_RES* pRes = taos_query(pConn, "use abc1");
+  taos_free_result(pRes);
+
+  pRes = taos_query(pConn, "select ts from st1");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
+    taos_free_result(pRes);
+    ASSERT_TRUE(false);
+  }
+
+  TAOS_ROW    pRow = NULL;
+  TAOS_FIELD* pFields = taos_fetch_fields(pRes);
+  int32_t     numOfFields = taos_num_fields(pRes);
+
+  char str[512] = {0};
+  while ((pRow = taos_fetch_row(pRes)) != NULL) {
+    int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
+    printf("%s\n", str);
+  }
+
+  taos_free_result(pRes);
+  taos_close(pConn);
+}
 
 
 TEST(testCase, agg_query_tables) {
@@ -773,7 +774,7 @@ TEST(testCase, agg_query_tables) {
   taos_free_result(pRes);
   taos_close(pConn);
 }
-#endif
+
 /*
 --- copy the following script in the shell to setup the environment ---
 
@@ -819,5 +820,27 @@ TEST(testCase, async_api_test) {
   getchar();
   taos_close(pConn);
 }
+#endif
+
+TEST(testCase, update_test) {
+  TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(pConn, nullptr);
+
+  taos_query(pConn, "use abc1");
+
+  TAOS_RES* pRes = taos_query(pConn, "create table tup (ts timestamp, k int);");
+  if (taos_errno(pRes) != 0) {
+    printf("failed to create table, reason:%s", taos_errstr(pRes));
+  }
+
+  taos_free_result(pRes);
+
+  char s[256]  = {0};
+  for(int32_t i = 0; i < 7000; ++i) {
+    sprintf(s, "insert into tup values('2020-1-1 1:1:1', %d)", i);
+    pRes = taos_query(pConn, s);
+    taos_free_result(pRes);
+  }
+}
 
 #pragma GCC diagnostic pop

source/common/src/tdatablock.c

@@ -1831,6 +1831,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
     pSubmitBlk->suid = suid;
     pSubmitBlk->uid = pDataBlock->info.groupId;
     pSubmitBlk->numOfRows = rows;
+    pSubmitBlk->sversion = pTSchema->version;
 
     msgLen += sizeof(SSubmitBlk);
     int32_t dataLen = 0;

source/common/src/tdataformat.c

@@ -29,15 +29,9 @@ typedef struct {
 #pragma pack(pop)
 
 #define TSROW_IS_KV_ROW(r) ((r)->flags & TSROW_KV_ROW)
-#define BIT1_SIZE(n)       (((n)-1) / 8 + 1)
-#define BIT2_SIZE(n)       (((n)-1) / 4 + 1)
-#define SET_BIT1(p, i, v)  ((p)[(i) / 8] = (p)[(i) / 8] & (~(((uint8_t)1) << ((i) % 8))) | ((v) << ((i) % 8)))
-#define SET_BIT2(p, i, v)  ((p)[(i) / 4] = (p)[(i) / 4] & (~(((uint8_t)3) << ((i) % 4))) | ((v) << ((i) % 4)))
-#define GET_BIT1(p, i)     (((p)[(i) / 8] >> ((i) % 8)) & ((uint8_t)1))
-#define GET_BIT2(p, i)     (((p)[(i) / 4] >> ((i) % 4)) & ((uint8_t)3))
 
 // SValue
-static FORCE_INLINE int32_t tPutValue(uint8_t *p, SValue *pValue, int8_t type) {
+int32_t tPutValue(uint8_t *p, SValue *pValue, int8_t type) {
   int32_t n = 0;
 
   if (IS_VAR_DATA_TYPE(type)) {
@@ -88,7 +82,7 @@ static FORCE_INLINE int32_t tPutValue(uint8_t *p, SValue *pValue, int8_t type) {
   return n;
 }
 
-static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type) {
+int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type) {
   int32_t n = 0;
 
   if (IS_VAR_DATA_TYPE(type)) {
@@ -421,7 +415,7 @@ int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, S
     _set_none:
       if ((flags & 0xf0) == 0) {
         setBitMap(pb, 0, iColumn - 1, flags);
-        if (flags & TSROW_HAS_VAL) { // set 0
+        if (flags & TSROW_HAS_VAL) {  // set 0
           if (IS_VAR_DATA_TYPE(pTColumn->type)) {
             *(VarDataOffsetT *)(pf + pTColumn->offset) = 0;
           } else {
@@ -434,7 +428,7 @@ int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, S
     _set_null:
       if ((flags & 0xf0) == 0) {
         setBitMap(pb, 1, iColumn - 1, flags);
-        if (flags & TSROW_HAS_VAL) { // set 0
+        if (flags & TSROW_HAS_VAL) {  // set 0
           if (IS_VAR_DATA_TYPE(pTColumn->type)) {
             *(VarDataOffsetT *)(pf + pTColumn->offset) = 0;
           } else {
@@ -639,15 +633,15 @@ void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal
   }
 
 _return_none:
-  *pColVal = COL_VAL_NONE(pTColumn->colId);
+  *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
   return;
 
 _return_null:
-  *pColVal = COL_VAL_NULL(pTColumn->colId);
+  *pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
   return;
 
 _return_value:
-  *pColVal = COL_VAL_VALUE(pTColumn->colId, value);
+  *pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, value);
   return;
 }
 
@@ -1105,9 +1099,9 @@ _err:
 #if 1  // ===================================================================================================================
 static void dataColSetNEleNull(SDataCol *pCol, int nEle);
 int         tdAllocMemForCol(SDataCol *pCol, int maxPoints) {
-  int spaceNeeded = pCol->bytes * maxPoints;
-  if (IS_VAR_DATA_TYPE(pCol->type)) {
-    spaceNeeded += sizeof(VarDataOffsetT) * maxPoints;
+          int spaceNeeded = pCol->bytes * maxPoints;
+          if (IS_VAR_DATA_TYPE(pCol->type)) {
+            spaceNeeded += sizeof(VarDataOffsetT) * maxPoints;
   }
 #ifdef TD_SUPPORT_BITMAP
   int32_t nBitmapBytes = (int32_t)TD_BITMAP_BYTES(maxPoints);

source/common/src/tmsg.c

@@ -5445,6 +5445,37 @@ int32_t tDecodeSTqOffset(SDecoder *pDecoder, STqOffset *pOffset) {
   return 0;
 }
 
+int32_t tEncodeDeleteRes(SEncoder *pCoder, const SDeleteRes *pRes) {
+  int32_t nUid = taosArrayGetSize(pRes->uidList);
+
+  if (tEncodeU64(pCoder, pRes->suid) < 0) return -1;
+  if (tEncodeI32v(pCoder, nUid) < 0) return -1;
+  for (int32_t iUid = 0; iUid < nUid; iUid++) {
+    if (tEncodeU64(pCoder, *(uint64_t *)taosArrayGet(pRes->uidList, iUid)) < 0) return -1;
+  }
+  if (tEncodeI64(pCoder, pRes->skey) < 0) return -1;
+  if (tEncodeI64(pCoder, pRes->ekey) < 0) return -1;
+  if (tEncodeI64v(pCoder, pRes->affectedRows) < 0) return -1;
+
+  return 0;
+}
+
+int32_t tDecodeDeleteRes(SDecoder *pCoder, SDeleteRes *pRes) {
+  int32_t  nUid;
+  uint64_t uid;
+
+  if (tDecodeU64(pCoder, &pRes->suid) < 0) return -1;
+  if (tDecodeI32v(pCoder, &nUid) < 0) return -1;
+  for (int32_t iUid = 0; iUid < nUid; iUid++) {
+    if (tDecodeU64(pCoder, &uid) < 0) return -1;
+    taosArrayPush(pRes->uidList, &uid);
+  }
+  if (tDecodeI64(pCoder, &pRes->skey) < 0) return -1;
+  if (tDecodeI64(pCoder, &pRes->ekey) < 0) return -1;
+  if (tDecodeI64v(pCoder, &pRes->affectedRows) < 0) return -1;
+
+  return 0;
+}
 int32_t tEncodeSMqDataRsp(SEncoder *pEncoder, const SMqDataRsp *pRsp) {
   if (tEncodeSTqOffsetVal(pEncoder, &pRsp->reqOffset) < 0) return -1;
   if (tEncodeSTqOffsetVal(pEncoder, &pRsp->rspOffset) < 0) return -1;

source/common/src/trow.c

@@ -34,6 +34,7 @@ const uint8_t tdVTypeByte[2][3] = {{
 // declaration
 static uint8_t tdGetBitmapByte(uint8_t byte);
 static int32_t tdCompareColId(const void *arg1, const void *arg2);
+static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2);
 
 // static void dataColSetNEleNull(SDataCol *pCol, int nEle);
 
@@ -339,9 +340,9 @@ int32_t tdSetBitmapValTypeN(void *pBitmap, int16_t nEle, TDRowValT valType, int8
 }
 
 bool tdIsBitmapBlkNorm(const void *pBitmap, int32_t numOfBits, int8_t bitmapMode) {
-  int32_t nBytes = (bitmapMode == 0 ? numOfBits / TD_VTYPE_PARTS : numOfBits / TD_VTYPE_PARTS_I);
-  uint8_t vTypeByte = tdVTypeByte[bitmapMode][TD_VTYPE_NORM];
-  uint8_t *qBitmap = (uint8_t*)pBitmap;
+  int32_t  nBytes = (bitmapMode == 0 ? numOfBits / TD_VTYPE_PARTS : numOfBits / TD_VTYPE_PARTS_I);
+  uint8_t  vTypeByte = tdVTypeByte[bitmapMode][TD_VTYPE_NORM];
+  uint8_t *qBitmap = (uint8_t *)pBitmap;
   for (int i = 0; i < nBytes; ++i) {
     if (*qBitmap != vTypeByte) {
       return false;
@@ -1045,13 +1046,28 @@ int32_t dataColGetNEleLen(SDataCol *pDataCol, int32_t rows, int8_t bitmapMode) {
   return result;
 }
 
-bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, uint32_t offset, col_id_t colIdx, SCellVal *pVal) {
+bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCellVal *pVal) {
   if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
     tdRowSetVal(pVal, TD_VTYPE_NORM, TD_ROW_KEY_ADDR(pRow));
     return true;
   }
+  int16_t nCols = tdRowGetNCols(pRow) - 1;
+  if (nCols <= 0) {
+    pVal->valType = TD_VTYPE_NONE;
+    return true;
+  }
+
+  SKvRowIdx *pColIdx =
+      (SKvRowIdx *)taosbsearch(&colId, TD_ROW_COL_IDX(pRow), nCols, sizeof(SKvRowIdx), compareKvRowColId, TD_EQ);
+
+  if (!pColIdx) {
+    pVal->valType = TD_VTYPE_NONE;
+    return true;
+  }
+
   void *pBitmap = tdGetBitmapAddrKv(pRow, tdRowGetNCols(pRow));
-  tdGetKvRowValOfCol(pVal, pRow, pBitmap, offset, colIdx);
+  tdGetKvRowValOfCol(pVal, pRow, pBitmap, pColIdx->offset,
+                     POINTER_DISTANCE(pColIdx, TD_ROW_COL_IDX(pRow)) / sizeof(SKvRowIdx));
   return true;
 }
 
@@ -1204,6 +1220,112 @@ static FORCE_INLINE int32_t compareKvRowColId(const void *key1, const void *key2
   }
 }
 
+int32_t tdSTSRowNew(SArray *pArray, STSchema *pTSchema, STSRow **ppRow) {
+  STColumn *pTColumn;
+  SColVal  *pColVal;
+  int32_t   nColVal = taosArrayGetSize(pArray);
+  int32_t   varDataLen = 0;
+  int32_t   maxVarDataLen = 0;
+  int32_t   iColVal = 0;
+  void     *varBuf = NULL;
+
+  ASSERT(nColVal > 1);
+
+  for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; ++iColumn) {
+    pTColumn = &pTSchema->columns[iColumn];
+    if (iColVal < nColVal) {
+      pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
+    } else {
+      pColVal = NULL;
+    }
+
+    if (iColumn == 0) {
+      ASSERT(pColVal->cid == pTColumn->colId);
+      ASSERT(pTColumn->type == TSDB_DATA_TYPE_TIMESTAMP);
+      ASSERT(pTColumn->colId == PRIMARYKEY_TIMESTAMP_COL_ID);
+    } else {
+      if (IS_VAR_DATA_TYPE(pTColumn->type)) {
+        if (pColVal) {
+          varDataLen += (pColVal->value.nData + sizeof(VarDataLenT));
+          if (maxVarDataLen < (pColVal->value.nData + sizeof(VarDataLenT))) {
+            maxVarDataLen = pColVal->value.nData + sizeof(VarDataLenT);
+          }
+        } else {
+          varDataLen += sizeof(VarDataLenT);
+          if (pTColumn->type == TSDB_DATA_TYPE_VARCHAR) {
+            varDataLen += CHAR_BYTES;
+            if (maxVarDataLen < CHAR_BYTES + sizeof(VarDataLenT)) {
+              maxVarDataLen = CHAR_BYTES + sizeof(VarDataLenT);
+            }
+          } else {
+            varDataLen += INT_BYTES;
+            if (maxVarDataLen < INT_BYTES + sizeof(VarDataLenT)) {
+              maxVarDataLen = INT_BYTES + sizeof(VarDataLenT);
+            }
+          }
+        }
+      }
+    }
+
+    ++iColVal;
+  }
+
+  *ppRow = (STSRow *)taosMemoryCalloc(
+      1, sizeof(STSRow) + pTSchema->flen + varDataLen + TD_BITMAP_BYTES(pTSchema->numOfCols - 1));
+
+  if (!(*ppRow)) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return -1;
+  }
+
+  if (maxVarDataLen > 0) {
+    varBuf = taosMemoryMalloc(maxVarDataLen);
+    if (!varBuf) {
+      taosMemoryFreeClear(*ppRow);
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      return -1;
+    }
+  }
+
+  SRowBuilder rb = {0};
+  tdSRowInit(&rb, pTSchema->version);
+  tdSRowSetInfo(&rb, pTSchema->numOfCols, pTSchema->numOfCols, pTSchema->flen);
+  tdSRowResetBuf(&rb, *ppRow);
+
+  iColVal = 0;
+  for (int32_t iColumn = 0; iColumn < pTSchema->numOfCols; ++iColumn) {
+    pTColumn = &pTSchema->columns[iColumn];
+
+    TDRowValT   valType = TD_VTYPE_NORM;
+    const void *val = NULL;
+    if (iColVal < nColVal) {
+      pColVal = (SColVal *)taosArrayGet(pArray, iColVal);
+      if (pColVal->isNone) {
+        valType = TD_VTYPE_NONE;
+      } else if (pColVal->isNull) {
+        valType = TD_VTYPE_NULL;
+      } else if (IS_VAR_DATA_TYPE(pTColumn->type)) {
+        varDataSetLen(varBuf, pColVal->value.nData);
+        memcpy(varDataVal(varBuf), pColVal->value.pData, pColVal->value.nData);
+        val = varBuf;
+      } else {
+        val = (const void *)&pColVal->value.i64;
+      }
+    } else {
+      pColVal = NULL;
+      valType = TD_VTYPE_NONE;
+    }
+
+    tdAppendColValToRow(&rb, pTColumn->colId, pTColumn->type, valType, val, true, pTColumn->offset, iColVal);
+
+    ++iColVal;
+  }
+
+  taosMemoryFreeClear(varBuf);
+
+  return 0;
+}
+
 bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCellVal *pVal) {
   if (colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
     pVal->val = &pIter->pRow->ts;
@@ -1593,7 +1715,6 @@ int32_t tdSRowSetExtendedInfo(SRowBuilder *pBuilder, int32_t nCols, int32_t nBou
   } else {
     pBuilder->rowType = TD_ROW_TP;
   }
-
   pBuilder->flen = flen;
   pBuilder->nCols = nCols;
   pBuilder->nBoundCols = nBoundCols;
@@ -1855,4 +1976,36 @@ void tdSTSRowIterReset(STSRowIter *pIter, STSRow *pRow) {
 void tdSTSRowIterInit(STSRowIter *pIter, STSchema *pSchema) {
   pIter->pSchema = pSchema;
   pIter->maxColId = pSchema->columns[pSchema->numOfCols - 1].colId;
-}
+}
\ No newline at end of file
+
+void tTSRowGetVal(STSRow *pRow, STSchema *pTSchema, int16_t iCol, SColVal *pColVal) {
+  STColumn *pTColumn = &pTSchema->columns[iCol];
+  SCellVal  cv;
+  SValue    value;
+
+  ASSERT(iCol > 0);
+
+  if (TD_IS_TP_ROW(pRow)) {
+    tdSTpRowGetVal(pRow, pTColumn->colId, pTColumn->type, pTSchema->flen, pTColumn->offset, iCol - 1, &cv);
+  } else if (TD_IS_KV_ROW(pRow)) {
+    ASSERT(iCol > 0);
+    tdSKvRowGetVal(pRow, pTColumn->colId, iCol - 1, &cv);
+  } else {
+    ASSERT(0);
+  }
+
+  if (tdValTypeIsNone(cv.valType)) {
+    *pColVal = COL_VAL_NONE(pTColumn->colId, pTColumn->type);
+  } else if (tdValTypeIsNull(cv.valType)) {
+    *pColVal = COL_VAL_NULL(pTColumn->colId, pTColumn->type);
+  } else {
+    if (IS_VAR_DATA_TYPE(pTColumn->type)) {
+      value.nData = varDataLen(cv.val);
+      value.pData = varDataVal(cv.val);
+    } else {
+      tGetValue(cv.val, &value, pTColumn->type);
+    }
+
+    *pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, value);
+  }
+}

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

@@ -350,6 +350,7 @@ SArray *vmGetMsgHandles() {
   if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_COMMIT_OFFSET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
+  if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
   if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
 
   if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;

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

@@ -107,7 +107,7 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf
     const STraceId *trace = &pMsg->info.traceId;
     dGTrace("vgId:%d, msg:%p get from vnode-sync queue", pVnode->vgId, pMsg);
 
-    int32_t code = vnodeProcessSyncReq(pVnode->pImpl, pMsg, NULL);  // no response here
+    int32_t code = vnodeProcessSyncMsg(pVnode->pImpl, pMsg, NULL);  // no response here
     dGTrace("vgId:%d, msg:%p is freed, code:0x%x", pVnode->vgId, pMsg, code);
     rpcFreeCont(pMsg->pCont);
     taosFreeQitem(pMsg);

source/dnode/mgmt/node_mgmt/src/dmMgmt.c

@@ -16,6 +16,7 @@
 #define _DEFAULT_SOURCE
 #include "dmMgmt.h"
 #include "dmNodes.h"
+#include "index.h"
 #include "qworker.h"
 
 static bool dmRequireNode(SDnode *pDnode, SMgmtWrapper *pWrapper) {
@@ -213,6 +214,7 @@ void dmCleanupDnode(SDnode *pDnode) {
   dmCleanupServer(pDnode);
   dmClearVars(pDnode);
   rpcCleanup();
+  indexCleanup();
   dDebug("dnode is closed, ptr:%p", pDnode);
 }
 

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

@@ -1373,9 +1373,9 @@ char *buildRetension(SArray *pRetension) {
 static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, int32_t rows, int64_t numOfTables,
                            bool sysDb, ESdbStatus objStatus, bool sysinfo) {
   int32_t cols = 0;
+  int32_t bytes = pShow->pMeta->pSchemas[cols].bytes;
+  char   *buf = taosMemoryMalloc(bytes);
 
-  int32_t     bytes = pShow->pMeta->pSchemas[cols].bytes;
-  char       *buf = taosMemoryMalloc(bytes);
   const char *name = mndGetDbStr(pDb->name);
   if (name != NULL) {
     STR_WITH_MAXSIZE_TO_VARSTR(buf, name, bytes);
@@ -1383,11 +1383,11 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
     STR_WITH_MAXSIZE_TO_VARSTR(buf, "NULL", bytes);
   }
 
-  char *status = "ready";
-  if (objStatus == SDB_STATUS_CREATING) status = "creating";
-  if (objStatus == SDB_STATUS_DROPPING) status = "dropping";
-  char statusB[24] = {0};
-  STR_WITH_SIZE_TO_VARSTR(statusB, status, strlen(status));
+  char *statusStr = "ready";
+  if (objStatus == SDB_STATUS_CREATING) statusStr = "creating";
+  if (objStatus == SDB_STATUS_DROPPING) statusStr = "dropping";
+  char statusVstr[24] = {0};
+  STR_WITH_SIZE_TO_VARSTR(statusVstr, statusStr, strlen(statusStr));
 
   if (sysDb || !sysinfo) {
     for (int32_t i = 0; i < pShow->numOfColumns; ++i) {
@@ -1397,7 +1397,7 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
       } else if (i == 3) {
         colDataAppend(pColInfo, rows, (const char *)&numOfTables, false);
       } else if (i == 20) {
-        colDataAppend(pColInfo, rows, statusB, false);
+        colDataAppend(pColInfo, rows, statusVstr, false);
       } else {
         colDataAppendNULL(pColInfo, rows);
       }
@@ -1405,7 +1405,6 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
   } else {
     SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, rows, buf, false);
-    taosMemoryFree(buf);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, rows, (const char *)&pDb->createdTime, false);
@@ -1419,30 +1418,29 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.replications, false);
 
-    const char *src = pDb->cfg.strict ? "strict" : "no_strict";
-    char        strict[24] = {0};
-    STR_WITH_SIZE_TO_VARSTR(strict, src, strlen(src));
+    const char *strictStr = pDb->cfg.strict ? "strict" : "no_strict";
+    char        strictVstr[24] = {0};
+    STR_WITH_SIZE_TO_VARSTR(strictVstr, strictStr, strlen(strictStr));
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, rows, (const char *)strict, false);
+    colDataAppend(pColInfo, rows, (const char *)strictVstr, false);
 
-    char    tmp[128] = {0};
-    int32_t len = 0;
-    len = sprintf(&tmp[VARSTR_HEADER_SIZE], "%dm", pDb->cfg.daysPerFile);
-    varDataSetLen(tmp, len);
+    char    durationVstr[128] = {0};
+    int32_t len = sprintf(&durationVstr[VARSTR_HEADER_SIZE], "%dm", pDb->cfg.daysPerFile);
+    varDataSetLen(durationVstr, len);
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, rows, (const char *)tmp, false);
+    colDataAppend(pColInfo, rows, (const char *)durationVstr, false);
 
+    char keepVstr[128] = {0};
     if (pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep1 || pDb->cfg.daysToKeep0 > pDb->cfg.daysToKeep2) {
-      len = sprintf(&tmp[VARSTR_HEADER_SIZE], "%dm,%dm,%dm", pDb->cfg.daysToKeep1, pDb->cfg.daysToKeep2,
+      len = sprintf(&keepVstr[VARSTR_HEADER_SIZE], "%dm,%dm,%dm", pDb->cfg.daysToKeep1, pDb->cfg.daysToKeep2,
                     pDb->cfg.daysToKeep0);
     } else {
-      len = sprintf(&tmp[VARSTR_HEADER_SIZE], "%dm,%dm,%dm", pDb->cfg.daysToKeep0, pDb->cfg.daysToKeep1,
+      len = sprintf(&keepVstr[VARSTR_HEADER_SIZE], "%dm,%dm,%dm", pDb->cfg.daysToKeep0, pDb->cfg.daysToKeep1,
                     pDb->cfg.daysToKeep2);
     }
-
-    varDataSetLen(tmp, len);
+    varDataSetLen(keepVstr, len);
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, rows, (const char *)tmp, false);
+    colDataAppend(pColInfo, rows, (const char *)keepVstr, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.buffer, false);
@@ -1469,68 +1467,49 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in
     colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.compression, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-
-    STR_WITH_SIZE_TO_VARSTR(strict, src, strlen(src));
-#if 0
-    char cacheModel[24] = {0};
-    bool null = false;
-    if (pDb->cfg.cacheLastRow == 0) {
-      STR_TO_VARSTR(cacheModel, "no_cache");
-    } else if (pDb->cfg.cacheLastRow == 1) {
-      STR_TO_VARSTR(cacheModel, "last_row_cache")
-    } else {
-      null = true;
-    }
-    colDataAppend(pColInfo, rows, cacheModel, null);
-#endif
     colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.cacheLastRow, false);
 
-    char *prec = NULL;
+    const char *precStr = NULL;
     switch (pDb->cfg.precision) {
       case TSDB_TIME_PRECISION_MILLI:
-        prec = TSDB_TIME_PRECISION_MILLI_STR;
+        precStr = TSDB_TIME_PRECISION_MILLI_STR;
         break;
       case TSDB_TIME_PRECISION_MICRO:
-        prec = TSDB_TIME_PRECISION_MICRO_STR;
+        precStr = TSDB_TIME_PRECISION_MICRO_STR;
         break;
       case TSDB_TIME_PRECISION_NANO:
-        prec = TSDB_TIME_PRECISION_NANO_STR;
+        precStr = TSDB_TIME_PRECISION_NANO_STR;
         break;
       default:
-        prec = "none";
+        precStr = "none";
         break;
     }
-
-    char t[10] = {0};
-    STR_WITH_SIZE_TO_VARSTR(t, prec, 2);
+    char precVstr[10] = {0};
+    STR_WITH_SIZE_TO_VARSTR(precVstr, precStr, 2);
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, rows, (const char *)t, false);
+    colDataAppend(pColInfo, rows, (const char *)precVstr, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
     colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.numOfStables, false);
 
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    colDataAppend(pColInfo, rows, (const char *)statusB, false);
-
-    //    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
-    //    colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.schemaless, false);
-
-    char *p = buildRetension(pDb->cfg.pRetensions);
+    colDataAppend(pColInfo, rows, (const char *)statusVstr, false);
 
+    char *rentensionVstr = buildRetension(pDb->cfg.pRetensions);
     pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
-    if (p == NULL) {
+    if (rentensionVstr == NULL) {
       colDataAppendNULL(pColInfo, rows);
     } else {
-      colDataAppend(pColInfo, rows, (const char *)p, false);
-      taosMemoryFree(p);
+      colDataAppend(pColInfo, rows, (const char *)rentensionVstr, false);
+      taosMemoryFree(rentensionVstr);
     }
   }
+
+  taosMemoryFree(buf);
 }
 
 static void setInformationSchemaDbCfg(SDbObj *pDbObj) {
-  ASSERT(pDbObj != NULL);
-  strncpy(pDbObj->name, TSDB_INFORMATION_SCHEMA_DB, tListLen(pDbObj->name));
-
+  tstrncpy(pDbObj->name, TSDB_INFORMATION_SCHEMA_DB, tListLen(pDbObj->name));
   pDbObj->createdTime = 0;
   pDbObj->cfg.numOfVgroups = 0;
   pDbObj->cfg.strict = 1;
@@ -1539,9 +1518,7 @@ static void setInformationSchemaDbCfg(SDbObj *pDbObj) {
 }
 
 static void setPerfSchemaDbCfg(SDbObj *pDbObj) {
-  ASSERT(pDbObj != NULL);
-  strncpy(pDbObj->name, TSDB_PERFORMANCE_SCHEMA_DB, tListLen(pDbObj->name));
-
+  tstrncpy(pDbObj->name, TSDB_PERFORMANCE_SCHEMA_DB, tListLen(pDbObj->name));
   pDbObj->createdTime = 0;
   pDbObj->cfg.numOfVgroups = 0;
   pDbObj->cfg.strict = 1;
@@ -1585,14 +1562,11 @@ static int32_t mndRetrieveDbs(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
 
   while (numOfRows < rowsCapacity) {
     pShow->pIter = sdbFetchAll(pSdb, SDB_DB, pShow->pIter, (void **)&pDb, &objStatus);
-    if (pShow->pIter == NULL) {
-      break;
-    }
+    if (pShow->pIter == NULL) break;
 
     if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_READ_OR_WRITE_DB, pDb) == 0) {
       int32_t numOfTables = 0;
       sdbTraverse(pSdb, SDB_VGROUP, mndGetTablesOfDbFp, &numOfTables, NULL, NULL);
-
       dumpDbInfoData(pBlock, pDb, pShow, numOfRows, numOfTables, false, objStatus, sysinfo);
       numOfRows++;
     }

source/dnode/vnode/CMakeLists.txt

@@ -44,8 +44,12 @@ target_sources(
     "src/tsdb/tsdbMemTable.c"
     "src/tsdb/tsdbRead.c"
     "src/tsdb/tsdbReadImpl.c"
+    "src/tsdb/tsdbCache.c"
     "src/tsdb/tsdbWrite.c"
+    "src/tsdb/tsdbReaderWriter.c"
+    "src/tsdb/tsdbUtil.c"
     "src/tsdb/tsdbSnapshot.c"
+    "src/tsdb/tsdbCacheRead.c"
 
     # tq
     "src/tq/tq.c"

source/dnode/vnode/inc/vnode.h

@@ -28,7 +28,6 @@
 
 #include "tcommon.h"
 #include "tfs.h"
-#include "tmallocator.h"
 #include "tmsg.h"
 #include "trow.h"
 
@@ -52,10 +51,10 @@ int32_t vnodeCreate(const char *path, SVnodeCfg *pCfg, STfs *pTfs);
 void    vnodeDestroy(const char *path, STfs *pTfs);
 SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb);
 void    vnodeClose(SVnode *pVnode);
-int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg);
+int32_t vnodePreProcessReq(SVnode *pVnode, SRpcMsg *pMsg);
 int32_t vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg *pRsp);
 int32_t vnodeProcessCMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
-int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
+int32_t vnodeProcessSyncMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp);
 int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
 int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg);
 int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo);
@@ -70,6 +69,10 @@ int32_t vnodeSnapshotReaderOpen(SVnode *pVnode, SVSnapshotReader **ppReader, int
 int32_t vnodeSnapshotReaderClose(SVSnapshotReader *pReader);
 int32_t vnodeSnapshotRead(SVSnapshotReader *pReader, const void **ppData, uint32_t *nData);
 int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen);
+int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list);
+int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list);
+void *  vnodeGetIdx(SVnode *pVnode);
+void *  vnodeGetIvtIdx(SVnode *pVnode);
 
 void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs);
 void vnodeApplyMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs);
@@ -89,7 +92,7 @@ typedef struct SMetaFltParam {
   tb_uid_t suid;
   int16_t  cid;
   int16_t  type;
-  char    *val;
+  char *   val;
   bool     reverse;
   int (*filterFunc)(void *a, void *b, int16_t type);
 
@@ -110,33 +113,33 @@ int32_t     metaTbCursorNext(SMTbCursor *pTbCur);
 
 // tsdb
 // typedef struct STsdb STsdb;
-typedef void *tsdbReaderT;
-
-#define BLOCK_LOAD_OFFSET_SEQ_ORDER 1
-#define BLOCK_LOAD_TABLE_SEQ_ORDER  2
-#define BLOCK_LOAD_TABLE_RR_ORDER   3
-
-int32_t     tsdbSetTableId(tsdbReaderT reader, int64_t uid);
-int32_t     tsdbSetTableList(tsdbReaderT reader, SArray *tableList);
-tsdbReaderT tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, SArray *tableList, uint64_t qId,
-                           uint64_t taskId);
-tsdbReaderT tsdbQueryCacheLast(SVnode *pVnode, SQueryTableDataCond *pCond, STableListInfo *groupList, uint64_t qId,
-                               void *pMemRef);
-int32_t     tsdbGetFileBlocksDistInfo(tsdbReaderT *pReader, STableBlockDistInfo *pTableBlockInfo);
-bool        isTsdbCacheLastRow(tsdbReaderT *pReader);
-int32_t     tsdbGetAllTableList(SMeta *pMeta, uint64_t uid, SArray *list);
-int32_t     tsdbGetCtbIdList(SMeta *pMeta, int64_t suid, SArray *list);
-int32_t     tsdbGetStbIdList(SMeta *pMeta, int64_t suid, SArray *list);
-void       *tsdbGetIdx(SMeta *pMeta);
-void       *tsdbGetIvtIdx(SMeta *pMeta);
-int64_t     tsdbGetNumOfRowsInMemTable(tsdbReaderT *pHandle);
-
-bool    tsdbNextDataBlock(tsdbReaderT pTsdbReadHandle);
-void    tsdbRetrieveDataBlockInfo(tsdbReaderT *pTsdbReadHandle, SDataBlockInfo *pBlockInfo);
-int32_t tsdbRetrieveDataBlockStatisInfo(tsdbReaderT *pTsdbReadHandle, SColumnDataAgg ***pBlockStatis, bool *allHave);
-SArray *tsdbRetrieveDataBlock(tsdbReaderT *pTsdbReadHandle, SArray *pColumnIdList);
-void    tsdbResetReadHandle(tsdbReaderT queryHandle, SQueryTableDataCond *pCond, int32_t tWinIdx);
-void    tsdbCleanupReadHandle(tsdbReaderT queryHandle);
+typedef struct STsdbReader STsdbReader;
+
+#define BLOCK_LOAD_OFFSET_ORDER   1
+#define BLOCK_LOAD_TABLESEQ_ORDER 2
+#define BLOCK_LOAD_EXTERN_ORDER   3
+
+#define LASTROW_RETRIEVE_TYPE_ALL    0x1
+#define LASTROW_RETRIEVE_TYPE_SINGLE 0x2
+
+int32_t tsdbSetTableId(STsdbReader *pReader, int64_t uid);
+int32_t tsdbReaderOpen(SVnode *pVnode, SQueryTableDataCond *pCond, SArray *pTableList, STsdbReader **ppReader,
+                       const char *idstr);
+void    tsdbReaderClose(STsdbReader *pReader);
+bool    tsdbNextDataBlock(STsdbReader *pReader);
+void    tsdbRetrieveDataBlockInfo(STsdbReader *pReader, SDataBlockInfo *pDataBlockInfo);
+int32_t tsdbRetrieveDataBlockStatisInfo(STsdbReader *pReader, SColumnDataAgg ***pBlockStatis, bool *allHave);
+SArray *tsdbRetrieveDataBlock(STsdbReader *pTsdbReadHandle, SArray *pColumnIdList);
+int32_t tsdbReaderReset(STsdbReader *pReader, SQueryTableDataCond *pCond, int32_t tWinIdx);
+int32_t tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo);
+int64_t tsdbGetNumOfRowsInMemTable(STsdbReader *pHandle);
+void *  tsdbGetIdx(SMeta *pMeta);
+void *  tsdbGetIvtIdx(SMeta *pMeta);
+
+int32_t tsdbLastRowReaderOpen(void *pVnode, int32_t type, SArray *pTableIdList, int32_t *colId, int32_t numOfCols,
+                              void **pReader);
+int32_t tsdbRetrieveLastRow(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds);
+int32_t tsdbLastrowReaderClose(void *pReader);
 
 // tq
 
@@ -208,7 +211,7 @@ struct SMetaEntry {
   int8_t   type;
   int8_t   flags;  // TODO: need refactor?
   tb_uid_t uid;
-  char    *name;
+  char *   name;
   union {
     struct {
       SSchemaWrapper schemaRow;
@@ -219,7 +222,7 @@ struct SMetaEntry {
       int64_t  ctime;
       int32_t  ttlDays;
       int32_t  commentLen;
-      char    *comment;
+      char *   comment;
       tb_uid_t suid;
       uint8_t *pTags;
     } ctbEntry;
@@ -227,7 +230,7 @@ struct SMetaEntry {
       int64_t        ctime;
       int32_t        ttlDays;
       int32_t        commentLen;
-      char          *comment;
+      char *         comment;
       int32_t        ncid;  // next column id
       SSchemaWrapper schemaRow;
     } ntbEntry;
@@ -241,17 +244,17 @@ struct SMetaEntry {
 
 struct SMetaReader {
   int32_t    flags;
-  SMeta     *pMeta;
+  SMeta *    pMeta;
   SDecoder   coder;
   SMetaEntry me;
-  void      *pBuf;
+  void *     pBuf;
   int32_t    szBuf;
 };
 
 struct SMTbCursor {
-  TBC        *pDbc;
-  void       *pKey;
-  void       *pVal;
+  TBC *       pDbc;
+  void *      pKey;
+  void *      pVal;
   int32_t     kLen;
   int32_t     vLen;
   SMetaReader mr;

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

@@ -62,6 +62,7 @@ struct STSmaStat {
 
 struct SRSmaStat {
   SSma     *pSma;
+  int64_t   submitVer;
   int64_t   refId;         // shared by fetch tasks
   void     *tmrHandle;     // shared by fetch tasks
   int8_t    triggerStat;   // shared by fetch tasks
@@ -84,6 +85,7 @@ struct SSmaStat {
 #define RSMA_TRIGGER_STAT(r) (&(r)->triggerStat)
 #define RSMA_RUNNING_STAT(r) (&(r)->runningStat)
 #define RSMA_REF_ID(r)       ((r)->refId)
+#define RSMA_SUBMIT_VER(r)   ((r)->submitVer)
 
 enum {
   TASK_TRIGGER_STAT_INIT = 0,
@@ -208,11 +210,15 @@ struct STFInfo {
   // specific fields
   union {
     struct {
-      int64_t applyVer[2];
+      int64_t submitVer;
     } qTaskInfo;
   };
 };
 
+enum {
+  TD_FTYPE_RSMA_QTASKINFO = 0,
+};
+
 struct STFile {
   uint8_t   state;
   STFInfo   info;

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

@@ -20,6 +20,7 @@
 #include "filter.h"
 #include "qworker.h"
 #include "sync.h"
+#include "tRealloc.h"
 #include "tchecksum.h"
 #include "tcoding.h"
 #include "tcompare.h"
@@ -27,15 +28,15 @@
 #include "tdatablock.h"
 #include "tdb.h"
 #include "tencode.h"
-#include "tref.h"
 #include "tfs.h"
 #include "tglobal.h"
 #include "tjson.h"
 #include "tlist.h"
 #include "tlockfree.h"
 #include "tlosertree.h"
-#include "tmallocator.h"
+#include "tlrucache.h"
 #include "tmsgcb.h"
+#include "tref.h"
 #include "tskiplist.h"
 #include "tstream.h"
 #include "ttime.h"
@@ -94,6 +95,7 @@ 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);
 STSchema*       metaGetTbTSchema(SMeta* pMeta, tb_uid_t uid, int32_t sver);
+int32_t         metaGetTbTSchemaEx(SMeta* pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema** ppTSchema);
 int             metaGetTableEntryByName(SMetaReader* pReader, const char* name);
 tb_uid_t        metaGetTableEntryUidByName(SMeta* pMeta, const char* name);
 int             metaGetTbNum(SMeta* pMeta);
@@ -127,9 +129,7 @@ int         tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSub
 int32_t     tsdbInsertTableData(STsdb* pTsdb, int64_t version, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock,
                                 SSubmitBlkRsp* pRsp);
 int32_t     tsdbDeleteTableData(STsdb* pTsdb, int64_t version, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKEY eKey);
-tsdbReaderT tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, SArray* tableList, uint64_t qId,
-                           uint64_t taskId);
-tsdbReaderT tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableListInfo* tableList, uint64_t qId,
+STsdbReader tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableListInfo* tableList, uint64_t qId,
                                 void* pMemRef);
 int32_t     tsdbSnapshotReaderOpen(STsdb* pTsdb, STsdbSnapshotReader** ppReader, int64_t sver, int64_t ever);
 int32_t     tsdbSnapshotReaderClose(STsdbSnapshotReader* pReader);
@@ -157,6 +157,7 @@ int32_t tqProcessTaskDispatchRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRecoverRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRetrieveRsp(STQ* pTq, SRpcMsg* pMsg);
+int32_t tsdbGetStbIdList(SMeta* pMeta, int64_t suid, SArray* list);
 
 SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pSchema, bool createTb, int64_t suid,
                             const char* stbFullName, int32_t vgId);
@@ -171,6 +172,7 @@ int32_t smaPostCommit(SSma* pSma);
 
 int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
 int32_t tdProcessTSmaInsert(SSma* pSma, int64_t indexUid, const char* msg);
+int64_t tdRSmaGetMaxSubmitVer(SSma* pSma, int8_t level);
 
 int32_t tdProcessRSmaCreate(SVnode* pVnode, SVCreateStbReq* pReq);
 int32_t tdProcessRSmaSubmit(SSma* pSma, void* pMsg, int32_t inputType);
@@ -196,9 +198,9 @@ typedef struct {
 
 // SVState
 struct SVState {
-  // int64_t processed;
   int64_t committed;
   int64_t applied;
+  int64_t commitID;
 };
 
 struct SVnodeInfo {
@@ -240,7 +242,7 @@ struct SVnode {
   SSink*     pSink;
   tsem_t     canCommit;
   int64_t    sync;
-  int32_t    syncCount;
+  int32_t    blockCount;
   tsem_t     syncSem;
   SQHandle*  pQuery;
 };

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

@@ -31,7 +31,7 @@ void metaReaderClear(SMetaReader *pReader) {
 }
 
 int metaGetTableEntryByVersion(SMetaReader *pReader, int64_t version, tb_uid_t uid) {
-  SMeta *  pMeta = pReader->pMeta;
+  SMeta   *pMeta = pReader->pMeta;
   STbDbKey tbDbKey = {.version = version, .uid = uid};
 
   // query table.db
@@ -54,7 +54,7 @@ _err:
 }
 
 int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {
-  SMeta * pMeta = pReader->pMeta;
+  SMeta  *pMeta = pReader->pMeta;
   int64_t version;
 
   // query uid.idx
@@ -68,7 +68,7 @@ int metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid) {
 }
 
 int metaGetTableEntryByName(SMetaReader *pReader, const char *name) {
-  SMeta *  pMeta = pReader->pMeta;
+  SMeta   *pMeta = pReader->pMeta;
   tb_uid_t uid;
 
   // query name.idx
@@ -82,7 +82,7 @@ int metaGetTableEntryByName(SMetaReader *pReader, const char *name) {
 }
 
 tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name) {
-  void *   pData = NULL;
+  void    *pData = NULL;
   int      nData = 0;
   tb_uid_t uid = 0;
 
@@ -138,7 +138,7 @@ void metaCloseTbCursor(SMTbCursor *pTbCur) {
 
 int metaTbCursorNext(SMTbCursor *pTbCur) {
   int    ret;
-  void * pBuf;
+  void  *pBuf;
   STbCfg tbCfg;
 
   for (;;) {
@@ -159,7 +159,7 @@ int metaTbCursorNext(SMTbCursor *pTbCur) {
 }
 
 SSchemaWrapper *metaGetTableSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver, bool isinline) {
-  void *          pData = NULL;
+  void           *pData = NULL;
   int             nData = 0;
   int64_t         version;
   SSchemaWrapper  schema = {0};
@@ -218,9 +218,9 @@ _err:
   return NULL;
 }
 
-int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList){
-  TBC *    pCur;
-  int ret = tdbTbcOpen(pMeta->pTtlIdx, &pCur, NULL);
+int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList) {
+  TBC *pCur;
+  int  ret = tdbTbcOpen(pMeta->pTtlIdx, &pCur, NULL);
   if (ret < 0) {
     return ret;
   }
@@ -235,13 +235,13 @@ int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList){
   }
 
   void *pKey = NULL;
-  int kLen = 0;
-  while(1){
+  int   kLen = 0;
+  while (1) {
     ret = tdbTbcPrev(pCur, &pKey, &kLen, NULL, NULL);
     if (ret < 0) {
       break;
     }
-    ttlKey = *(STtlIdxKey*)pKey;
+    ttlKey = *(STtlIdxKey *)pKey;
     taosArrayPush(uidList, &ttlKey.uid);
   }
   tdbTbcClose(pCur);
@@ -252,11 +252,11 @@ int metaTtlSmaller(SMeta *pMeta, uint64_t ttl, SArray *uidList){
 }
 
 struct SMCtbCursor {
-  SMeta *  pMeta;
-  TBC *    pCur;
+  SMeta   *pMeta;
+  TBC     *pCur;
   tb_uid_t suid;
-  void *   pKey;
-  void *   pVal;
+  void    *pKey;
+  void    *pVal;
   int      kLen;
   int      vLen;
 };
@@ -388,15 +388,15 @@ tb_uid_t metaStbCursorNext(SMStbCursor *pStbCur) {
   if (ret < 0) {
     return 0;
   }
-  return *(tb_uid_t*)pStbCur->pKey;
+  return *(tb_uid_t *)pStbCur->pKey;
 }
 
 STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
   // SMetaReader     mr = {0};
-  STSchema *      pTSchema = NULL;
+  STSchema       *pTSchema = NULL;
   SSchemaWrapper *pSW = NULL;
   STSchemaBuilder sb = {0};
-  SSchema *       pSchema;
+  SSchema        *pSchema;
 
   pSW = metaGetTableSchema(pMeta, uid, sver, 0);
   if (!pSW) return NULL;
@@ -415,6 +415,51 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
   return pTSchema;
 }
 
+int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sver, STSchema **ppTSchema) {
+  int32_t   code = 0;
+  STSchema *pTSchema = NULL;
+  SSkmDbKey skmDbKey = {.uid = suid ? suid : uid, .sver = sver};
+  void     *pData = NULL;
+  int       nData = 0;
+
+  // query
+  metaRLock(pMeta);
+  if (tdbTbGet(pMeta->pSkmDb, &skmDbKey, sizeof(skmDbKey), &pData, &nData) < 0) {
+    code = TSDB_CODE_NOT_FOUND;
+    metaULock(pMeta);
+    goto _err;
+  }
+  metaULock(pMeta);
+
+  // decode
+  SDecoder        dc = {0};
+  SSchemaWrapper  schema;
+  SSchemaWrapper *pSchemaWrapper = &schema;
+
+  tDecoderInit(&dc, pData, nData);
+  tDecodeSSchemaWrapper(&dc, pSchemaWrapper);
+  tDecoderClear(&dc);
+
+  // convert
+  STSchemaBuilder sb = {0};
+
+  tdInitTSchemaBuilder(&sb, pSchemaWrapper->version);
+  for (int i = 0; i < pSchemaWrapper->nCols; i++) {
+    SSchema *pSchema = pSchemaWrapper->pSchema + i;
+    tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes);
+  }
+  pTSchema = tdGetSchemaFromBuilder(&sb);
+  tdDestroyTSchemaBuilder(&sb);
+
+  *ppTSchema = pTSchema;
+  taosMemoryFree(pSchemaWrapper->pSchema);
+  return code;
+
+_err:
+  *ppTSchema = NULL;
+  return code;
+}
+
 int metaGetTbNum(SMeta *pMeta) {
   // TODO
   // ASSERT(0);
@@ -422,11 +467,11 @@ int metaGetTbNum(SMeta *pMeta) {
 }
 
 typedef struct {
-  SMeta *  pMeta;
-  TBC *    pCur;
+  SMeta   *pMeta;
+  TBC     *pCur;
   tb_uid_t uid;
-  void *   pKey;
-  void *   pVal;
+  void    *pKey;
+  void    *pVal;
   int      kLen;
   int      vLen;
 } SMSmaCursor;
@@ -498,7 +543,7 @@ tb_uid_t metaSmaCursorNext(SMSmaCursor *pSmaCur) {
 
 STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) {
   STSmaWrapper *pSW = NULL;
-  SArray *      pSmaIds = NULL;
+  SArray       *pSmaIds = NULL;
 
   if (!(pSmaIds = metaGetSmaIdsByTable(pMeta, uid))) {
     return NULL;
@@ -522,7 +567,7 @@ STSmaWrapper *metaGetSmaInfoByTable(SMeta *pMeta, tb_uid_t uid, bool deepCopy) {
   metaReaderInit(&mr, pMeta, 0);
   int64_t smaId;
   int     smaIdx = 0;
-  STSma * pTSma = NULL;
+  STSma  *pTSma = NULL;
   for (int i = 0; i < pSW->number; ++i) {
     smaId = *(tb_uid_t *)taosArrayGet(pSmaIds, i);
     if (metaGetTableEntryByUid(&mr, smaId) < 0) {
@@ -570,7 +615,7 @@ _err:
 }
 
 STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {
-  STSma *     pTSma = NULL;
+  STSma      *pTSma = NULL;
   SMetaReader mr = {0};
   metaReaderInit(&mr, pMeta, 0);
   if (metaGetTableEntryByUid(&mr, indexUid) < 0) {
@@ -592,7 +637,7 @@ STSma *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid) {
 }
 
 SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) {
-  SArray *    pUids = NULL;
+  SArray     *pUids = NULL;
   SSmaIdxKey *pSmaIdxKey = NULL;
 
   SMSmaCursor *pCur = metaOpenSmaCursor(pMeta, uid);
@@ -630,7 +675,7 @@ SArray *metaGetSmaIdsByTable(SMeta *pMeta, tb_uid_t uid) {
 }
 
 SArray *metaGetSmaTbUids(SMeta *pMeta) {
-  SArray *    pUids = NULL;
+  SArray     *pUids = NULL;
   SSmaIdxKey *pSmaIdxKey = NULL;
   tb_uid_t    lastUid = 0;
 
@@ -689,20 +734,20 @@ const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t type, STagVal *val) {
 }
 
 typedef struct {
-  SMeta *  pMeta;
-  TBC *    pCur;
+  SMeta   *pMeta;
+  TBC     *pCur;
   tb_uid_t suid;
   int16_t  cid;
   int16_t  type;
-  void *   pKey;
-  void *   pVal;
+  void    *pKey;
+  void    *pVal;
   int32_t  kLen;
   int32_t  vLen;
 } SIdxCursor;
 
 int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
   SIdxCursor *pCursor = NULL;
-  char *      buf = NULL;
+  char       *buf = NULL;
   int32_t     maxSize = 0;
 
   int32_t ret = 0, valid = 0;
@@ -721,7 +766,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
   int32_t     nKey = 0;
 
   int32_t nTagData = 0;
-  void *  tagData = NULL;
+  void   *tagData = NULL;
 
   if (param->val == NULL) {
     metaError("vgId:%d, failed to filter NULL data", TD_VID(pMeta->pVnode));
@@ -757,7 +802,7 @@ int32_t metaFilteTableIds(SMeta *pMeta, SMetaFltParam *param, SArray *pUids) {
     goto END;
   }
 
-  void *  entryKey = NULL, *entryVal = NULL;
+  void   *entryKey = NULL, *entryVal = NULL;
   int32_t nEntryKey, nEntryVal;
   bool    first = true;
   while (1) {

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

@@ -523,6 +523,17 @@ static void tdDestroySDataBlockArray(SArray *pArray) {
   taosArrayDestroy(pArray);
 }
 
+int64_t tdRSmaGetMaxSubmitVer(SSma *pSma, int8_t level) {
+  if (level == TSDB_RETENTION_L0) {
+    return pSma->pVnode->state.applied;
+  }
+
+  SSmaEnv   *pRSmaEnv = SMA_RSMA_ENV(pSma);
+  SRSmaStat *pRSmaStat = (SRSmaStat *)(SMA_ENV_STAT(pRSmaEnv));
+
+  return atomic_load_64(&pRSmaStat->submitVer);
+}
+
 static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType) {
   SArray    *pResult = NULL;
   SRSmaInfo *pRSmaInfo = pItem->pRsmaInfo;
@@ -562,7 +573,7 @@ static int32_t tdFetchAndSubmitRSmaResult(SRSmaInfoItem *pItem, int8_t blkType)
       goto _err;
     }
 
-    if (pReq && tdProcessSubmitReq(sinkTsdb, INT64_MAX, pReq) < 0) {
+    if (pReq && tdProcessSubmitReq(sinkTsdb, atomic_add_fetch_64(&pRSmaInfo->pStat->submitVer, 1), pReq) < 0) {
       taosMemoryFreeClear(pReq);
       goto _err;
     }
@@ -814,6 +825,7 @@ static int32_t tdRSmaRestoreQTaskInfoReload(SSma *pSma, int64_t *committed) {
   }
 
   if (!taosCheckExistFile(TD_TFILE_FULL_NAME(&tFile))) {
+    *committed = 0;
     if (pVnode->state.committed > 0) {
       smaWarn("vgId:%d, rsma restore for version %" PRIi64 ", not start as %s not exist", TD_VID(pVnode),
               pVnode->state.committed, TD_TFILE_FULL_NAME(&tFile));
@@ -828,6 +840,18 @@ static int32_t tdRSmaRestoreQTaskInfoReload(SSma *pSma, int64_t *committed) {
     goto _err;
   }
 
+  STFInfo tFileInfo = {0};
+  if (tdLoadTFileHeader(&tFile, &tFileInfo) < 0) {
+    goto _err;
+  }
+
+  ASSERT(tFileInfo.qTaskInfo.submitVer > 0);
+
+  SSmaEnv   *pRSmaEnv = pSma->pRSmaEnv;
+  SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pRSmaEnv);
+  atomic_store_64(&pRSmaStat->submitVer, tFileInfo.qTaskInfo.submitVer);
+  smaDebug("%s:%d tFileInfo.qTaskInfo.submitVer = %" PRIi64, __func__, __LINE__, tFileInfo.qTaskInfo.submitVer);
+
   SRSmaQTaskInfoIter fIter = {0};
   if (tdRSmaQTaskInfoIterInit(&fIter, &tFile) < 0) {
     tdRSmaQTaskInfoIterDestroy(&fIter);
@@ -1094,6 +1118,22 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
   }
 
   STFile tFile = {0};
+  if (RSMA_SUBMIT_VER(pRSmaStat) > 0) {
+    char qTaskInfoFName[TSDB_FILENAME_LEN];
+    tdRSmaQTaskInfoGetFName(vid, pSma->pVnode->state.applied, qTaskInfoFName);
+    if (tdInitTFile(&tFile, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFName) < 0) {
+      smaError("vgId:%d, rsma persit, init %s failed since %s", vid, qTaskInfoFName, terrstr());
+      goto _err;
+    }
+    if (tdCreateTFile(&tFile, true, TD_FTYPE_RSMA_QTASKINFO) < 0) {
+      smaError("vgId:%d, rsma persit, create %s failed since %s", vid, TD_TFILE_FULL_NAME(&tFile), terrstr());
+      goto _err;
+    }
+    smaDebug("vgId:%d, rsma, serialize qTaskInfo, file %s created", vid, TD_TFILE_FULL_NAME(&tFile));
+
+    isFileCreated = true;
+  }
+
   while (infoHash) {
     SRSmaInfo *pRSmaInfo = *(SRSmaInfo **)infoHash;
     for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
@@ -1129,12 +1169,12 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
           smaError("vgId:%d, rsma persit, init %s failed since %s", vid, qTaskInfoFName, terrstr());
           goto _err;
         }
-        if (tdCreateTFile(&tFile, true, -1) < 0) {
+        if (tdCreateTFile(&tFile, true, TD_FTYPE_RSMA_QTASKINFO) < 0) {
           smaError("vgId:%d, rsma persit, create %s failed since %s", vid, TD_TFILE_FULL_NAME(&tFile), terrstr());
           goto _err;
         }
-        smaDebug("vgId:%d, rsma, table %" PRIi64 " level %d serialize qTaskInfo, file %s created", vid, pRSmaInfo->suid,
-                 i + 1, TD_TFILE_FULL_NAME(&tFile));
+        smaDebug("vgId:%d, rsma, table %" PRIi64 " serialize qTaskInfo, file %s created", vid, pRSmaInfo->suid,
+                 TD_TFILE_FULL_NAME(&tFile));
 
         isFileCreated = true;
       }
@@ -1161,6 +1201,7 @@ int32_t tdRSmaPersistExecImpl(SRSmaStat *pRSmaStat) {
   }
 
   if (isFileCreated) {
+    tFile.info.qTaskInfo.submitVer = atomic_load_64(&pRSmaStat->submitVer);
     if (tdUpdateTFileHeader(&tFile) < 0) {
       smaError("vgId:%d, rsma, failed to update tfile %s header since %s", vid, TD_TFILE_FULL_NAME(&tFile),
                tstrerror(terrno));

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

@@ -32,6 +32,9 @@ static int32_t tdEncodeTFInfo(void **buf, STFInfo *pInfo) {
   tlen += taosEncodeFixedU32(buf, pInfo->ftype);
   tlen += taosEncodeFixedU32(buf, pInfo->fver);
   tlen += taosEncodeFixedI64(buf, pInfo->fsize);
+  if (pInfo->ftype == TD_FTYPE_RSMA_QTASKINFO) {
+    tlen += taosEncodeFixedI64(buf, pInfo->qTaskInfo.submitVer);
+  }
 
   return tlen;
 }
@@ -41,6 +44,11 @@ static void *tdDecodeTFInfo(void *buf, STFInfo *pInfo) {
   buf = taosDecodeFixedU32(buf, &(pInfo->ftype));
   buf = taosDecodeFixedU32(buf, &(pInfo->fver));
   buf = taosDecodeFixedI64(buf, &(pInfo->fsize));
+  // specific
+  if (pInfo->ftype == TD_FTYPE_RSMA_QTASKINFO) {
+    buf = taosDecodeFixedI64(buf, &(pInfo->qTaskInfo.submitVer));
+  }
+
   return buf;
 }
 

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

@@ -462,8 +462,8 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
     } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) {
       pHandle->execHandle.execTb.suid = req.suid;
       SArray* tbUidList = taosArrayInit(0, sizeof(int64_t));
-      tsdbGetCtbIdList(pTq->pVnode->pMeta, req.suid, tbUidList);
-      tqDebug("vg %d, tq try get suid: %ld", TD_VID(pTq->pVnode), req.suid);
+      vnodeGetCtbIdList(pTq->pVnode, req.suid, tbUidList);
+      tqDebug("vg %d, tq try get suid: %ld", pTq->pVnode->config.vgId, req.suid);
       for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) {
         int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i);
         tqDebug("vg %d, idx %d, uid: %ld", TD_VID(pTq->pVnode), i, tbUid);

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

@@ -87,6 +87,7 @@ int32_t tqMetaOpen(STQ* pTq) {
             .reader = handle.execHandle.pExecReader[i],
             .meta = pTq->pVnode->pMeta,
             .pMsgCb = &pTq->pVnode->msgCb,
+            .vnode  = pTq->pVnode,
         };
         handle.execHandle.execCol.task[i] = qCreateStreamExecTaskInfo(handle.execHandle.execCol.qmsg, &reader);
         ASSERT(handle.execHandle.execCol.task[i]);

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

@@ -85,6 +85,7 @@ STqOffsetStore* tqOffsetOpen(STQ* pTq) {
 void tqOffsetClose(STqOffsetStore* pStore) {
   tqOffsetSnapshot(pStore);
   taosHashCleanup(pStore->pHash);
+  taosMemoryFree(pStore);
 }
 
 STqOffset* tqOffsetRead(STqOffsetStore* pStore, const char* subscribeKey) {

source/dnode/vnode/src/tsdb/tsdbDelete.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/>.
- */
\ No newline at end of file

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

@@ -17,7 +17,6 @@
 
 static int tsdbSetKeepCfg(STsdbKeepCfg *pKeepCfg, STsdbCfg *pCfg);
 
-
 // implementation
 
 static int tsdbSetKeepCfg(STsdbKeepCfg *pKeepCfg, STsdbCfg *pCfg) {
@@ -42,7 +41,7 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
   int    slen = 0;
 
   *ppTsdb = NULL;
-  slen = strlen(tfsGetPrimaryPath(pVnode->pTfs)) + strlen(pVnode->path) + strlen(dir) + 3;
+  slen = strlen(pVnode->path) + strlen(dir) + 2;
 
   // create handle
   pTsdb = (STsdb *)taosMemoryCalloc(1, sizeof(*pTsdb) + slen);
@@ -51,10 +50,8 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
     return -1;
   }
 
-  ASSERT(strlen(dir) < TSDB_DATA_DIR_LEN);
-  memcpy(pTsdb->dir, dir, strlen(dir));
   pTsdb->path = (char *)&pTsdb[1];
-  sprintf(pTsdb->path, "%s%s%s%s%s", tfsGetPrimaryPath(pVnode->pTfs), TD_DIRSEP, pVnode->path, TD_DIRSEP, dir);
+  sprintf(pTsdb->path, "%s%s%s", pVnode->path, TD_DIRSEP, dir);
   taosRealPath(pTsdb->path, NULL, slen);
   pTsdb->pVnode = pVnode;
   pTsdb->repoLocked = false;
@@ -64,13 +61,17 @@ int tsdbOpen(SVnode *pVnode, STsdb **ppTsdb, const char *dir, STsdbKeepCfg *pKee
   } else {
     memcpy(&pTsdb->keepCfg, pKeepCfg, sizeof(STsdbKeepCfg));
   }
-  pTsdb->fs = tsdbNewFS(REPO_KEEP_CFG(pTsdb));
+  // pTsdb->fs = tsdbNewFS(REPO_KEEP_CFG(pTsdb));
 
-  // create dir (TODO: use tfsMkdir)
-  taosMkDir(pTsdb->path);
+  // create dir
+  tfsMkdir(pVnode->pTfs, pTsdb->path);
 
   // open tsdb
-  if (tsdbOpenFS(pTsdb) < 0) {
+  if (tsdbFSOpen(pTsdb, &pTsdb->fs) < 0) {
+    goto _err;
+  }
+
+  if (tsdbOpenCache(pTsdb) < 0) {
     goto _err;
   }
 
@@ -87,10 +88,9 @@ _err:
 
 int tsdbClose(STsdb **pTsdb) {
   if (*pTsdb) {
-    // TODO: destroy mem/imem
     taosThreadMutexDestroy(&(*pTsdb)->mutex);
-    tsdbCloseFS(*pTsdb);
-    tsdbFreeFS((*pTsdb)->fs);
+    tsdbFSClose((*pTsdb)->fs);
+    tsdbCloseCache((*pTsdb)->lruCache);
     taosMemoryFreeClear(*pTsdb);
   }
   return 0;
@@ -99,7 +99,7 @@ int tsdbClose(STsdb **pTsdb) {
 int tsdbLockRepo(STsdb *pTsdb) {
   int code = taosThreadMutexLock(&pTsdb->mutex);
   if (code != 0) {
-    tsdbError("vgId:%d, failed to lock tsdb since %s", REPO_ID(pTsdb), strerror(errno));
+    tsdbError("vgId:%d, failed to lock tsdb since %s", TD_VID(pTsdb->pVnode), strerror(errno));
     terrno = TAOS_SYSTEM_ERROR(code);
     return -1;
   }
@@ -108,13 +108,13 @@ int tsdbLockRepo(STsdb *pTsdb) {
 }
 
 int tsdbUnlockRepo(STsdb *pTsdb) {
-  ASSERT(IS_REPO_LOCKED(pTsdb));
+  // ASSERT(IS_REPO_LOCKED(pTsdb));
   pTsdb->repoLocked = false;
   int code = taosThreadMutexUnlock(&pTsdb->mutex);
   if (code != 0) {
-    tsdbError("vgId:%d, failed to unlock tsdb since %s", REPO_ID(pTsdb), strerror(errno));
+    tsdbError("vgId:%d, failed to unlock tsdb since %s", TD_VID(pTsdb->pVnode), strerror(errno));
     terrno = TAOS_SYSTEM_ERROR(code);
     return -1;
   }
   return 0;
-}
+}
\ No newline at end of file

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

@@ -28,7 +28,7 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *
   // scan and convert
   if (tsdbScanAndConvertSubmitMsg(pTsdb, pMsg) < 0) {
     if (terrno != TSDB_CODE_TDB_TABLE_RECONFIGURE) {
-      tsdbError("vgId:%d, failed to insert data since %s", REPO_ID(pTsdb), tstrerror(terrno));
+      tsdbError("vgId:%d, failed to insert data since %s", TD_VID(pTsdb->pVnode), tstrerror(terrno));
     }
     return -1;
   }
@@ -77,7 +77,7 @@ static FORCE_INLINE int tsdbCheckRowRange(STsdb *pTsdb, tb_uid_t uid, STSRow *ro
   if (rowKey < minKey || rowKey > maxKey) {
     tsdbError("vgId:%d, table uid %" PRIu64 " timestamp is out of range! now %" PRId64 " minKey %" PRId64
               " maxKey %" PRId64 " row key %" PRId64,
-              REPO_ID(pTsdb), uid, now, minKey, maxKey, rowKey);
+              TD_VID(pTsdb->pVnode), uid, now, minKey, maxKey, rowKey);
     terrno = TSDB_CODE_TDB_TIMESTAMP_OUT_OF_RANGE;
     return -1;
   }
@@ -92,7 +92,7 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
   SSubmitBlk    *pBlock = NULL;
   SSubmitBlkIter blkIter = {0};
   STSRow        *row = NULL;
-  STsdbKeepCfg  *pCfg = REPO_KEEP_CFG(pTsdb);
+  STsdbKeepCfg  *pCfg = &pTsdb->keepCfg;
   TSKEY          now = taosGetTimestamp(pCfg->precision);
   TSKEY          minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep2;
   TSKEY          maxKey = now + tsTickPerMin[pCfg->precision] * pCfg->days;

source/libs/executor/src/tsort.c

@@ -545,6 +545,7 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
   return 0;
 }
 
+// TODO consider the page meta size
 int32_t getProperSortPageSize(size_t rowSize) {
   uint32_t defaultPageSize = 4096;