Merge branch '3.0' of https://github.com/taosdata/TDengine into feat/tdlite

README.md

@@ -15,7 +15,7 @@
 [![Coverage Status](https://coveralls.io/repos/github/taosdata/TDengine/badge.svg?branch=develop)](https://coveralls.io/github/taosdata/TDengine?branch=develop)
 [![CII Best Practices](https://bestpractices.coreinfrastructure.org/projects/4201/badge)](https://bestpractices.coreinfrastructure.org/projects/4201)
 
-English | [简体中文](README-CN.md) | [Learn more about TSDB](https://tdengine.com/tsdb/)
+English | [简体中文](README-CN.md) | [TDengine Cloud](https://cloud.tdengine.com) | [Learn more about TSDB](https://tdengine.com/tsdb/)
 
 # What is TDengine？
 
@@ -33,7 +33,7 @@ TDengine is an open source, high-performance, cloud native [time-series database
 
 - **[Open Source](https://tdengine.com/tdengine/open-source-time-series-database/)**: TDengine’s core modules, including cluster feature, are all available under open source licenses. It has gathered 18.8k stars on GitHub. There is an active developer community, and over 139k running instances worldwide.
 
-For a full list of TDengine competitive advantages, please [check here](https://tdengine.com/tdengine/)
+For a full list of TDengine competitive advantages, please [check here](https://tdengine.com/tdengine/). The easiest way to experience TDengine is through [TDengine Cloud](https://cloud.tdengine.com). 
 
 # Documentation
 

contrib/CMakeLists.txt

@@ -37,6 +37,11 @@ if(${BUILD_WITH_ICONV})
     cat("${TD_SUPPORT_DIR}/iconv_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 endif()
 
+# jemalloc
+if(${JEMALLOC_ENABLED})
+    cat("${TD_SUPPORT_DIR}/jemalloc_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+endif()
+
 # msvc regex
 if(${BUILD_MSVCREGEX})
     cat("${TD_SUPPORT_DIR}/msvcregex_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
@@ -258,6 +263,19 @@ if(${BUILD_PTHREAD})
     target_link_libraries(pthread INTERFACE libpthreadVC3)
 endif()
 
+# jemalloc
+if(${JEMALLOC_ENABLED})
+    include(ExternalProject)
+    ExternalProject_Add(jemalloc
+        PREFIX        "jemalloc"
+        SOURCE_DIR    ${CMAKE_CURRENT_SOURCE_DIR}/jemalloc
+        BUILD_IN_SOURCE     1
+        CONFIGURE_COMMAND   ./autogen.sh COMMAND ./configure --prefix=${CMAKE_BINARY_DIR}/build/ --disable-initial-exec-tls
+        BUILD_COMMAND       ${MAKE}
+    )
+    INCLUDE_DIRECTORIES(${CMAKE_BINARY_DIR}/build/include)
+endif()
+
 # crashdump
 if(${BUILD_CRASHDUMP})
     add_executable(dumper "crashdump/dumper/dumper.c")

docs/en/14-reference/05-taosbenchmark.md

@@ -23,7 +23,7 @@ There are two ways to install taosBenchmark:
 
 TaosBenchmark needs to be executed on the terminal of the operating system, it supports two configuration methods: [Command-line arguments](#command-line-arguments-in-detail) and [JSON configuration file](#configuration-file-parameters-in-detail). These two methods are mutually exclusive. Users can use `-f <json file>` to specify a configuration file. When running taosBenchmark with command-line arguments to control its behavior, users should use other parameters for configuration, but not the `-f` parameter. In addition, taosBenchmark offers a special way of running without parameters.
 
-taosBenchmark supports the complete performance testing of TDengine by providing functionally to write, query, and subscribe. These three functions are mutually exclusive, users can only select one of them each time taosBenchmark runs. The query and subscribe functionalities are only configurable using a json configuration file by specifying the parameter `filetype`, while write can be performed through both the command-line and a configuration file. If you want to test the performance of queries or data subscriptionm configure taosBenchmark with the configuration file. You can modify the value of the `filetype` parameter to specify the function that you want to test.
+taosBenchmark supports the complete performance testing of TDengine by providing functionally to write, query, and subscribe. These three functions are mutually exclusive, users can only select one of them each time taosBenchmark runs. The query and subscribe functionalities are only configurable using a json configuration file by specifying the parameter `filetype`, while write can be performed through both the command-line and a configuration file. If you want to test the performance of queries or data subscription configure taosBenchmark with the configuration file. You can modify the value of the `filetype` parameter to specify the function that you want to test.
 
 **Make sure that the TDengine cluster is running correctly before running taosBenchmark. **
 
@@ -340,7 +340,7 @@ The configuration parameters for specifying super table tag columns and data col
 
 - **values**: The value field of the nchar/binary column/label, which will be chosen randomly from the values.
 
-- **sma**: Insert the column into the BSMA. Enter `yes` or `no`. The default is `no`.
+- **sma**: Insert the column into the SMA. Enter `yes` or `no`. The default is `no`.
 
 #### insertion behavior configuration parameters
 

docs/examples/c/tmq_example.c

@@ -184,22 +184,54 @@ void tmq_commit_cb_print(tmq_t* tmq, int32_t code, void* param) {
 tmq_t* build_consumer() {
   tmq_conf_res_t code;
   tmq_conf_t*    conf = tmq_conf_new();
+
   code = tmq_conf_set(conf, "enable.auto.commit", "true");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
+
   code = tmq_conf_set(conf, "auto.commit.interval.ms", "1000");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
+
   code = tmq_conf_set(conf, "group.id", "cgrpName");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
+
   code = tmq_conf_set(conf, "client.id", "user defined name");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
+
   code = tmq_conf_set(conf, "td.connect.user", "root");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
+
   code = tmq_conf_set(conf, "td.connect.pass", "taosdata");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
+
   code = tmq_conf_set(conf, "auto.offset.reset", "earliest");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
+
   code = tmq_conf_set(conf, "experimental.snapshot.enable", "false");
-  if (TMQ_CONF_OK != code) return NULL;
+  if (TMQ_CONF_OK != code) {
+    tmq_conf_destroy(conf);
+    return NULL;
+  }
 
   tmq_conf_set_auto_commit_cb(conf, tmq_commit_cb_print, NULL);
 

docs/zh/14-reference/05-taosbenchmark.md

@@ -340,7 +340,7 @@ taosBenchmark -A INT,DOUBLE,NCHAR,BINARY\(16\)
 
 - **values** : nchar/binary 列/标签的值域，将从值中随机选择。
 
-- **sma**: 将该列加入bsma中，值为 "yes" 或者 "no"，默认为 "no"。
+- **sma**: 将该列加入 SMA 中，值为 "yes" 或者 "no"，默认为 "no"。
 
 #### 插入行为配置参数
 

include/common/tglobal.h

@@ -103,6 +103,7 @@ extern bool    tsKeepColumnName;
 // client
 extern int32_t tsMinSlidingTime;
 extern int32_t tsMinIntervalTime;
+extern int32_t tsMaxMemUsedByInsert;
 
 // build info
 extern char version[];

include/libs/catalog/catalog.h

@@ -303,6 +303,10 @@ int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, SRequestConnInfo* pConn, uint64_t re
 
 int32_t catalogClearCache(void);
 
+SMetaData* catalogCloneMetaData(SMetaData* pData);
+
+void catalogFreeMetaData(SMetaData* pData);
+
 /**
  * Destroy catalog and relase all resources
  */

include/libs/nodes/querynodes.h

@@ -385,7 +385,6 @@ typedef struct SCmdMsgInfo {
   SEpSet  epSet;
   void*   pMsg;
   int32_t msgLen;
-  void*   pExtension;  // todo remove it soon
 } SCmdMsgInfo;
 
 typedef enum EQueryExecMode {

include/libs/parser/parser.h

@@ -33,6 +33,13 @@ typedef struct SStmtCallback {
   int32_t (*getExecInfoFn)(TAOS_STMT*, SHashObj**, SHashObj**);
 } SStmtCallback;
 
+typedef struct SParseCsvCxt {
+  TdFilePtr   fp;           // last parsed file
+  int32_t     tableNo;      // last parsed table
+  SName       tableName;    // last parsed table
+  const char* pLastSqlPos;  // the location of the last parsed sql
+} SParseCsvCxt;
+
 typedef struct SParseContext {
   uint64_t         requestId;
   int64_t          requestRid;
@@ -57,6 +64,8 @@ typedef struct SParseContext {
   SArray*          pTableMetaPos;    // sql table pos => catalog data pos
   SArray*          pTableVgroupPos;  // sql table pos => catalog data pos
   int64_t          allocatorId;
+  bool             needMultiParse;
+  SParseCsvCxt     csvCxt;
 } SParseContext;
 
 int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery);
@@ -67,6 +76,8 @@ int32_t qParseSqlSyntax(SParseContext* pCxt, SQuery** pQuery, struct SCatalogReq
 int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCatalogReq,
                             const struct SMetaData* pMetaData, SQuery* pQuery);
 
+void qDestroyParseContext(SParseContext* pCxt);
+
 void qDestroyQuery(SQuery* pQueryNode);
 
 int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);

include/os/os.h

@@ -51,6 +51,9 @@ extern "C" {
 #endif
 #else
 
+#ifndef __func__
+#define __func__ __FUNCTION__ 
+#endif
 #include <malloc.h>
 #include <time.h>
 #ifndef TD_USE_WINSOCK

source/client/inc/clientInt.h

@@ -380,9 +380,16 @@ void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);
 // --- mq
 void hbMgrInitMqHbRspHandle();
 
+typedef struct SSqlCallbackWrapper {
+  SParseContext* pParseCtx;
+  SCatalogReq*   pCatalogReq;
+  SMetaData*     pResultMeta;
+  SRequestObj*   pRequest;
+} SSqlCallbackWrapper;
+
 SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res);
 int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
-void         launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta);
+void    launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta, SSqlCallbackWrapper* pWrapper);
 int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest);
 int32_t updateQnodeList(SAppInstInfo* pInfo, SArray* pNodeList);
 void    doAsyncQuery(SRequestObj* pRequest, bool forceUpdateMeta);
@@ -390,6 +397,8 @@ int32_t      removeMeta(STscObj* pTscObj, SArray* tbList);
 int32_t handleAlterTbExecRes(void* res, struct SCatalog* pCatalog);
 int32_t handleCreateTbExecRes(void* res, SCatalog* pCatalog);
 bool    qnodeRequired(SRequestObj* pRequest);
+int32_t continueInsertFromCsv(SSqlCallbackWrapper* pWrapper, SRequestObj* pRequest);
+void    destorySqlCallbackWrapper(SSqlCallbackWrapper* pWrapper);
 
 #ifdef __cplusplus
 }

source/client/src/clientImpl.c

@@ -870,7 +870,8 @@ int32_t handleQueryExecRsp(SRequestObj* pRequest) {
 
 // todo refacto the error code  mgmt
 void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
-  SRequestObj* pRequest = (SRequestObj*)param;
+  SSqlCallbackWrapper* pWrapper = param;
+  SRequestObj*         pRequest = pWrapper->pRequest;
   STscObj*             pTscObj = pRequest->pTscObj;
 
   pRequest->code = code;
@@ -882,7 +883,7 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
   int32_t type = pRequest->type;
   if (TDMT_VND_SUBMIT == type || TDMT_VND_DELETE == type || TDMT_VND_CREATE_TABLE == type) {
     if (pResult) {
-      pRequest->body.resInfo.numOfRows = pResult->numOfRows;
+      pRequest->body.resInfo.numOfRows += pResult->numOfRows;
 
       // record the insert rows
       if (TDMT_VND_SUBMIT == type) {
@@ -899,12 +900,13 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
            pRequest->requestId);
 
   if (code != TSDB_CODE_SUCCESS && NEED_CLIENT_HANDLE_ERROR(code) && pRequest->sqlstr != NULL) {
-    tscDebug("0x%" PRIx64 " client retry to handle the error, code:%s, tryCount:%d, reqId:0x%" PRIx64,
+    tscDebug("0x%" PRIx64 " client retry to handle the error, code:%s, tryCount:%d, reqId:0x%" PRIx64, pRequest->self,
-             pRequest->self, tstrerror(code), pRequest->retry, pRequest->requestId);
+             tstrerror(code), pRequest->retry, pRequest->requestId);
     pRequest->prevCode = code;
     schedulerFreeJob(&pRequest->body.queryJob, 0);
     qDestroyQuery(pRequest->pQuery);
     pRequest->pQuery = NULL;
+    destorySqlCallbackWrapper(pWrapper);
     doAsyncQuery(pRequest, true);
     return;
   }
@@ -920,6 +922,15 @@ void schedulerExecCb(SExecResult* pResult, void* param, int32_t code) {
     pRequest->code = code1;
   }
 
+  if (pRequest->code == TSDB_CODE_SUCCESS && NULL != pWrapper->pParseCtx && pWrapper->pParseCtx->needMultiParse) {
+    code = continueInsertFromCsv(pWrapper, pRequest);
+    if (TSDB_CODE_SUCCESS == code) {
+      return;
+    }
+  }
+
+  destorySqlCallbackWrapper(pWrapper);
+
   // return to client
   pRequest->body.queryFp(pRequest->body.param, pRequest, code);
 }
@@ -1020,23 +1031,11 @@ SRequestObj* launchQuery(uint64_t connId, const char* sql, int sqlLen, bool vali
   return launchQueryImpl(pRequest, pQuery, false, NULL);
 }
 
-void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta) {
+static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta,
-  int32_t code = 0;
+                                 SSqlCallbackWrapper* pWrapper) {
-
-  pRequest->body.execMode = pQuery->execMode;
-
-  switch (pQuery->execMode) {
-    case QUERY_EXEC_MODE_LOCAL:
-      asyncExecLocalCmd(pRequest, pQuery);
-      return;
-    case QUERY_EXEC_MODE_RPC:
-      code = asyncExecDdlQuery(pRequest, pQuery);
-      break;
-    case QUERY_EXEC_MODE_SCHEDULE: {
-      SArray* pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad));
-
   pRequest->type = pQuery->msgType;
 
+  SArray*      pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad));
   SPlanContext cxt = {.queryId = pRequest->requestId,
                       .acctId = pRequest->pTscObj->acctId,
                       .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
@@ -1047,10 +1046,8 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
                       .pUser = pRequest->pTscObj->user,
                       .sysInfo = pRequest->pTscObj->sysInfo,
                       .allocatorId = pRequest->allocatorRefId};
-
-      SAppInstInfo* pAppInfo = getAppInfo(pRequest);
   SQueryPlan*  pDag = NULL;
-      code = qCreateQueryPlan(&cxt, &pDag, pMnodeList);
+  int32_t      code = qCreateQueryPlan(&cxt, &pDag, pMnodeList);
   if (code) {
     tscError("0x%" PRIx64 " failed to create query plan, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
              pRequest->requestId);
@@ -1059,12 +1056,14 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
   }
 
   pRequest->metric.planEnd = taosGetTimestampUs();
+
   if (TSDB_CODE_SUCCESS == code && !pRequest->validateOnly) {
     SArray* pNodeList = NULL;
     buildAsyncExecNodeList(pRequest, &pNodeList, pMnodeList, pResultMeta);
 
-        SRequestConnInfo conn = {
+    SRequestConnInfo conn = {.pTrans = getAppInfo(pRequest)->pTransporter,
-            .pTrans = pAppInfo->pTransporter, .requestId = pRequest->requestId, .requestObjRefId = pRequest->self};
+                             .requestId = pRequest->requestId,
+                             .requestObjRefId = pRequest->self};
     SSchedulerReq    req = {
            .syncReq = false,
            .localReq = (tsQueryPolicy == QUERY_POLICY_CLIENT),
@@ -1075,7 +1074,7 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
            .sql = pRequest->sqlstr,
            .startTs = pRequest->metric.start,
            .execFp = schedulerExecCb,
-            .cbParam = pRequest,
+           .cbParam = pWrapper,
            .chkKillFp = chkRequestKilled,
            .chkKillParam = (void*)pRequest->self,
            .pExecRes = NULL,
@@ -1085,11 +1084,33 @@ void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultM
   } else {
     tscDebug("0x%" PRIx64 " plan not executed, code:%s 0x%" PRIx64, pRequest->self, tstrerror(code),
              pRequest->requestId);
+    destorySqlCallbackWrapper(pWrapper);
     pRequest->body.queryFp(pRequest->body.param, pRequest, code);
   }
 
   // todo not to be released here
   taosArrayDestroy(pMnodeList);
+
+  return code;
+}
+
+void launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta, SSqlCallbackWrapper* pWrapper) {
+  int32_t code = 0;
+
+  pRequest->body.execMode = pQuery->execMode;
+  if (QUERY_EXEC_MODE_SCHEDULE != pRequest->body.execMode) {
+    destorySqlCallbackWrapper(pWrapper);
+  }
+
+  switch (pQuery->execMode) {
+    case QUERY_EXEC_MODE_LOCAL:
+      asyncExecLocalCmd(pRequest, pQuery);
+      break;
+    case QUERY_EXEC_MODE_RPC:
+      code = asyncExecDdlQuery(pRequest, pQuery);
+      break;
+    case QUERY_EXEC_MODE_SCHEDULE: {
+      code = asyncExecSchQuery(pRequest, pQuery, pResultMeta, pWrapper);
       break;
     }
     case QUERY_EXEC_MODE_EMPTY_RESULT:

source/client/src/clientMain.c

@@ -667,35 +667,39 @@ const char *taos_get_server_info(TAOS *taos) {
   return pTscObj->sDetailVer;
 }
 
-typedef struct SqlParseWrapper {
-  SParseContext *pCtx;
-  SCatalogReq    catalogReq;
-  SRequestObj   *pRequest;
-} SqlParseWrapper;
-
 static void destoryTablesReq(void *p) {
   STablesReq *pRes = (STablesReq *)p;
   taosArrayDestroy(pRes->pTables);
 }
 
-static void destorySqlParseWrapper(SqlParseWrapper *pWrapper) {
+static void destoryCatalogReq(SCatalogReq *pCatalogReq) {
-  taosArrayDestroy(pWrapper->catalogReq.pDbVgroup);
+  if (NULL == pCatalogReq) {
-  taosArrayDestroy(pWrapper->catalogReq.pDbCfg);
+    return;
-  taosArrayDestroy(pWrapper->catalogReq.pDbInfo);
+  }
-  taosArrayDestroyEx(pWrapper->catalogReq.pTableMeta, destoryTablesReq);
+  taosArrayDestroy(pCatalogReq->pDbVgroup);
-  taosArrayDestroyEx(pWrapper->catalogReq.pTableHash, destoryTablesReq);
+  taosArrayDestroy(pCatalogReq->pDbCfg);
-  taosArrayDestroy(pWrapper->catalogReq.pUdf);
+  taosArrayDestroy(pCatalogReq->pDbInfo);
-  taosArrayDestroy(pWrapper->catalogReq.pIndex);
+  taosArrayDestroyEx(pCatalogReq->pTableMeta, destoryTablesReq);
-  taosArrayDestroy(pWrapper->catalogReq.pUser);
+  taosArrayDestroyEx(pCatalogReq->pTableHash, destoryTablesReq);
-  taosArrayDestroy(pWrapper->catalogReq.pTableIndex);
+  taosArrayDestroy(pCatalogReq->pUdf);
-  taosArrayDestroy(pWrapper->pCtx->pTableMetaPos);
+  taosArrayDestroy(pCatalogReq->pIndex);
-  taosArrayDestroy(pWrapper->pCtx->pTableVgroupPos);
+  taosArrayDestroy(pCatalogReq->pUser);
-  taosMemoryFree(pWrapper->pCtx);
+  taosArrayDestroy(pCatalogReq->pTableIndex);
+  taosMemoryFree(pCatalogReq);
+}
+
+void destorySqlCallbackWrapper(SSqlCallbackWrapper *pWrapper) {
+  if (NULL == pWrapper) {
+    return;
+  }
+  destoryCatalogReq(pWrapper->pCatalogReq);
+  qDestroyParseContext(pWrapper->pParseCtx);
+  catalogFreeMetaData(pWrapper->pResultMeta);
   taosMemoryFree(pWrapper);
 }
 
 void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
-  SqlParseWrapper *pWrapper = (SqlParseWrapper *)param;
+  SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)param;
   SRequestObj         *pRequest = pWrapper->pRequest;
   SQuery              *pQuery = pRequest->pQuery;
 
@@ -703,7 +707,7 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
   qDebug("0x%" PRIx64 " start to semantic analysis, reqId:0x%" PRIx64, pRequest->self, pRequest->requestId);
 
   if (code == TSDB_CODE_SUCCESS) {
-    code = qAnalyseSqlSemantic(pWrapper->pCtx, &pWrapper->catalogReq, pResultMeta, pQuery);
+    code = qAnalyseSqlSemantic(pWrapper->pParseCtx, pWrapper->pCatalogReq, pResultMeta, pQuery);
     pRequest->stableQuery = pQuery->stableQuery;
     if (pQuery->pRoot) {
       pRequest->stmtType = pQuery->pRoot->type;
@@ -712,6 +716,13 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
 
   pRequest->metric.semanticEnd = taosGetTimestampUs();
 
+  if (code == TSDB_CODE_SUCCESS && pWrapper->pParseCtx->needMultiParse) {
+    pWrapper->pResultMeta = catalogCloneMetaData(pResultMeta);
+    if (NULL == pWrapper->pResultMeta) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+    }
+  }
+
   if (code == TSDB_CODE_SUCCESS) {
     if (pQuery->haveResultSet) {
       setResSchemaInfo(&pRequest->body.resInfo, pQuery->pResSchema, pQuery->numOfResCols);
@@ -722,15 +733,13 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
     TSWAP(pRequest->tableList, (pQuery)->pTableList);
     TSWAP(pRequest->targetTableList, (pQuery)->pTargetTableList);
 
-    destorySqlParseWrapper(pWrapper);
-
     double el = (pRequest->metric.semanticEnd - pRequest->metric.ctgEnd) / 1000.0;
     tscDebug("0x%" PRIx64 " analysis semantics completed, start async query, elapsed time:%.2f ms, reqId:0x%" PRIx64,
              pRequest->self, el, pRequest->requestId);
 
-    launchAsyncQuery(pRequest, pQuery, pResultMeta);
+    launchAsyncQuery(pRequest, pQuery, pResultMeta, pWrapper);
   } else {
-    destorySqlParseWrapper(pWrapper);
+    destorySqlCallbackWrapper(pWrapper);
     qDestroyQuery(pRequest->pQuery);
     pRequest->pQuery = NULL;
 
@@ -750,6 +759,16 @@ void retrieveMetaCallback(SMetaData *pResultMeta, void *param, int32_t code) {
   }
 }
 
+int32_t continueInsertFromCsv(SSqlCallbackWrapper *pWrapper, SRequestObj *pRequest) {
+  qDestroyQuery(pRequest->pQuery);
+  pRequest->pQuery = (SQuery *)nodesMakeNode(QUERY_NODE_QUERY);
+  if (NULL == pRequest->pQuery) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  retrieveMetaCallback(pWrapper->pResultMeta, pWrapper, TSDB_CODE_SUCCESS);
+  return TSDB_CODE_SUCCESS;
+}
+
 void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) {
   int64_t connId = *(int64_t *)taos;
   taosAsyncQueryImpl(connId, sql, fp, param, false);
@@ -786,37 +805,48 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
 }
 
 void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
-  SParseContext *pCxt = NULL;
   STscObj             *pTscObj = pRequest->pTscObj;
-  int32_t        code = 0;
+  SSqlCallbackWrapper *pWrapper = NULL;
+  int32_t              code = TSDB_CODE_SUCCESS;
 
   if (pRequest->retry++ > REQUEST_TOTAL_EXEC_TIMES) {
     code = pRequest->prevCode;
-    goto _error;
   }
 
-  code = createParseContext(pRequest, &pCxt);
+  if (TSDB_CODE_SUCCESS == code) {
-  if (code != TSDB_CODE_SUCCESS) {
+    pWrapper = taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
-    goto _error;
+    if (pWrapper == NULL) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
+    } else {
+      pWrapper->pRequest = pRequest;
+    }
   }
 
-  pCxt->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
+  if (TSDB_CODE_SUCCESS == code) {
-  code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pCxt->pCatalog);
+    code = createParseContext(pRequest, &pWrapper->pParseCtx);
-  if (code != TSDB_CODE_SUCCESS) {
-    goto _error;
   }
 
+  if (TSDB_CODE_SUCCESS == code) {
+    pWrapper->pParseCtx->mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
+    code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &pWrapper->pParseCtx->pCatalog);
+  }
+
+  if (TSDB_CODE_SUCCESS == code) {
     pRequest->metric.syntaxStart = taosGetTimestampUs();
 
-  SCatalogReq catalogReq = {.forceUpdate = updateMetaForce, .qNodeRequired = qnodeRequired(pRequest)};
+    pWrapper->pCatalogReq = taosMemoryCalloc(1, sizeof(SCatalogReq));
-  code = qParseSqlSyntax(pCxt, &pRequest->pQuery, &catalogReq);
+    if (pWrapper->pCatalogReq == NULL) {
-  if (code != TSDB_CODE_SUCCESS) {
+      code = TSDB_CODE_OUT_OF_MEMORY;
-    goto _error;
+    } else {
+      pWrapper->pCatalogReq->forceUpdate = updateMetaForce;
+      pWrapper->pCatalogReq->qNodeRequired = qnodeRequired(pRequest);
+      code = qParseSqlSyntax(pWrapper->pParseCtx, &pRequest->pQuery, pWrapper->pCatalogReq);
     }
 
     pRequest->metric.syntaxEnd = taosGetTimestampUs();
+  }
 
-  if (!updateMetaForce) {
+  if (TSDB_CODE_SUCCESS == code && !updateMetaForce) {
     SAppClusterSummary *pActivity = &pTscObj->pAppInfo->summary;
     if (NULL == pRequest->pQuery->pRoot) {
       atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);
@@ -825,39 +855,27 @@ void doAsyncQuery(SRequestObj *pRequest, bool updateMetaForce) {
     }
   }
 
-  SqlParseWrapper *pWrapper = taosMemoryCalloc(1, sizeof(SqlParseWrapper));
+  if (TSDB_CODE_SUCCESS == code) {
-  if (pWrapper == NULL) {
+    SRequestConnInfo conn = {.pTrans = pWrapper->pParseCtx->pTransporter,
-    code = TSDB_CODE_OUT_OF_MEMORY;
+                             .requestId = pWrapper->pParseCtx->requestId,
-    goto _error;
+                             .requestObjRefId = pWrapper->pParseCtx->requestRid,
-  }
+                             .mgmtEps = pWrapper->pParseCtx->mgmtEpSet};
-
-  pWrapper->pCtx = pCxt;
-  pWrapper->pRequest = pRequest;
-  pWrapper->catalogReq = catalogReq;
-
-  SRequestConnInfo conn = {.pTrans = pCxt->pTransporter,
-                           .requestId = pCxt->requestId,
-                           .requestObjRefId = pCxt->requestRid,
-                           .mgmtEps = pCxt->mgmtEpSet};
 
     pRequest->metric.ctgStart = taosGetTimestampUs();
 
-  code = catalogAsyncGetAllMeta(pCxt->pCatalog, &conn, &catalogReq, retrieveMetaCallback, pWrapper,
+    code = catalogAsyncGetAllMeta(pWrapper->pParseCtx->pCatalog, &conn, pWrapper->pCatalogReq, retrieveMetaCallback,
-                                &pRequest->body.queryJob);
+                                  pWrapper, &pRequest->body.queryJob);
-  pCxt = NULL;
-  if (code == TSDB_CODE_SUCCESS) {
-    return;
   }
 
-_error:
+  if (TSDB_CODE_SUCCESS != code) {
     tscError("0x%" PRIx64 " error happens, code:%d - %s, reqId:0x%" PRIx64, pRequest->self, code, tstrerror(code),
              pRequest->requestId);
-  taosMemoryFree(pCxt);
+    destorySqlCallbackWrapper(pWrapper);
-
     terrno = code;
     pRequest->code = code;
     pRequest->body.queryFp(pRequest->body.param, pRequest, code);
   }
+}
 
 static void fetchCallback(void *pResult, void *param, int32_t code) {
   SRequestObj *pRequest = (SRequestObj *)param;

source/client/src/clientSml.c

@@ -314,7 +314,8 @@ static int32_t getBytes(uint8_t type, int32_t length){
   }
 }
 
-static int32_t smlBuildFieldsList(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols, SArray* results, int32_t numOfCols, bool isTag) {
+static int32_t smlBuildFieldsList(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
+                                  SArray *results, int32_t numOfCols, bool isTag) {
   for (int j = 0; j < taosArrayGetSize(cols); ++j) {
     SSmlKv       *kv = (SSmlKv *)taosArrayGetP(cols, j);
     ESchemaAction action = SCHEMA_ACTION_NULL;
@@ -338,9 +339,8 @@ static int32_t smlBuildFieldsList(SSmlHandle *info, SSchema *schemaField, SHashO
 
 // static int32_t smlSendMetaMsg(SSmlHandle *info, SName *pName, SSmlSTableMeta *sTableData,
 //                               int32_t colVer, int32_t tagVer, int8_t source, uint64_t suid){
-static int32_t  smlSendMetaMsg(SSmlHandle *info, SName *pName, SArray* pColumns, SArray* pTags,
+static int32_t smlSendMetaMsg(SSmlHandle *info, SName *pName, SArray *pColumns, SArray *pTags, STableMeta *pTableMeta,
-                               STableMeta *pTableMeta, ESchemaAction action){
+                              ESchemaAction action) {
-
   SRequestObj   *pRequest = NULL;
   SMCreateStbReq pReq = {0};
   int32_t        code = TSDB_CODE_SUCCESS;
@@ -463,8 +463,8 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) {
       }
       info->cost.numOfCreateSTables++;
     } else if (code == TSDB_CODE_SUCCESS) {
-      hashTmp = taosHashInit(pTableMeta->tableInfo.numOfTags,
+      hashTmp = taosHashInit(pTableMeta->tableInfo.numOfTags, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true,
-                                       taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
+                             HASH_NO_LOCK);
       for (uint16_t i = pTableMeta->tableInfo.numOfColumns;
            i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
         taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
@@ -476,8 +476,10 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) {
         goto end;
       }
       if (action != SCHEMA_ACTION_NULL) {
-        SArray* pColumns = taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
+        SArray *pColumns =
-        SArray* pTags = taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
+            taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
+        SArray *pTags =
+            taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
 
         for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
           SField field = {0};
@@ -490,7 +492,8 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) {
             taosArrayPush(pTags, &field);
           }
         }
-        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->tags, pTags, pTableMeta->tableInfo.numOfColumns, true);
+        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->tags, pTags,
+                           pTableMeta->tableInfo.numOfColumns, true);
 
         code = smlSendMetaMsg(info, &pName, pColumns, pTags, pTableMeta, action);
         if (code != TSDB_CODE_SUCCESS) {
@@ -519,8 +522,10 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) {
         goto end;
       }
       if (action != SCHEMA_ACTION_NULL) {
-        SArray* pColumns = taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
+        SArray *pColumns =
-        SArray* pTags = taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
+            taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
+        SArray *pTags =
+            taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
 
         for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
           SField field = {0};
@@ -534,7 +539,8 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) {
           }
         }
 
-        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->cols, pColumns, pTableMeta->tableInfo.numOfColumns, false);
+        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->cols, pColumns,
+                           pTableMeta->tableInfo.numOfColumns, false);
 
         code = smlSendMetaMsg(info, &pName, pColumns, pTags, pTableMeta, action);
         if (code != TSDB_CODE_SUCCESS) {
@@ -1830,7 +1836,8 @@ static int32_t smlConvertJSONString(SSmlKv *pVal, char *typeStr, cJSON *value) {
   if (pVal->type == TSDB_DATA_TYPE_BINARY && pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
     return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
   }
-  if (pVal->type == TSDB_DATA_TYPE_NCHAR  && pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
+  if (pVal->type == TSDB_DATA_TYPE_NCHAR &&
+      pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
     return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
   }
 
@@ -2314,7 +2321,8 @@ static int32_t smlInsertData(SSmlHandle *info) {
     (*pMeta)->tableMeta->uid = tableData->uid;  // one table merge data block together according uid
 
     code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat,
-                       (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen, info->msgBuf.buf, info->msgBuf.len);
+                       (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen,
+                       info->msgBuf.buf, info->msgBuf.len);
     if (code != TSDB_CODE_SUCCESS) {
       uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
       return code;
@@ -2336,7 +2344,12 @@ static int32_t smlInsertData(SSmlHandle *info) {
   SAppClusterSummary *pActivity = &info->taos->pAppInfo->summary;
   atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);
 
-  launchAsyncQuery(info->pRequest, info->pQuery, NULL);
+  SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
+  if (pWrapper == NULL) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  pWrapper->pRequest = info->pRequest;
+  launchAsyncQuery(info->pRequest, info->pQuery, NULL, pWrapper);
   return TSDB_CODE_SUCCESS;
 }
 

source/common/src/tdataformat.c

@@ -411,7 +411,7 @@ _exit:
 
 int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow) {
   int32_t code = 0;
-  int32_t rLen;
+  int32_t rLen = 0;
 
   TSROW_LEN(pRow, rLen);
   (*ppRow) = (STSRow2 *)taosMemoryMalloc(rLen);
@@ -1678,8 +1678,8 @@ int32_t tColDataCopy(SColData *pColDataSrc, SColData *pColDataDest) {
   int32_t size;
 
   ASSERT(pColDataSrc->nVal > 0);
-  ASSERT(pColDataDest->cid = pColDataSrc->cid);
+  ASSERT(pColDataDest->cid == pColDataSrc->cid);
-  ASSERT(pColDataDest->type = pColDataSrc->type);
+  ASSERT(pColDataDest->type == pColDataSrc->type);
 
   pColDataDest->smaOn = pColDataSrc->smaOn;
   pColDataDest->nVal = pColDataSrc->nVal;

source/common/src/tglobal.c

@@ -85,7 +85,8 @@ uint16_t tsTelemPort = 80;
 char tsSmlTagName[TSDB_COL_NAME_LEN] = "_tag_null";
 char tsSmlChildTableName[TSDB_TABLE_NAME_LEN] = "";  // user defined child table name can be specified in tag value.
                                                      // If set to empty system will generate table name using MD5 hash.
-bool tsSmlDataFormat = false;  // true means that the name and order of cols in each line are the same(only for influx protocol)
+// true means that the name and order of cols in each line are the same(only for influx protocol)
+bool tsSmlDataFormat = false;
 
 // query
 int32_t tsQueryPolicy = 1;
@@ -125,6 +126,9 @@ int32_t tsMaxNumOfDistinctResults = 1000 * 10000;
 // 1 database precision unit for interval time range, changed accordingly
 int32_t tsMinIntervalTime = 1;
 
+// maximum memory allowed to be allocated for a single csv load (in MB)
+int32_t tsMaxMemUsedByInsert = 1024;
+
 // the maximum allowed query buffer size during query processing for each data node.
 // -1 no limit (default)
 // 0  no query allowed, queries are disabled
@@ -296,6 +300,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   if (cfgAddString(pCfg, "smlChildTableName", "", 1) != 0) return -1;
   if (cfgAddString(pCfg, "smlTagName", tsSmlTagName, 1) != 0) return -1;
   if (cfgAddBool(pCfg, "smlDataFormat", tsSmlDataFormat, 1) != 0) return -1;
+  if (cfgAddInt32(pCfg, "maxMemUsedByInsert", tsMaxMemUsedByInsert, 1, INT32_MAX, true) != 0) return -1;
 
   tsNumOfTaskQueueThreads = tsNumOfCores / 2;
   tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4);
@@ -648,6 +653,8 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
   tstrncpy(tsSmlTagName, cfgGetItem(pCfg, "smlTagName")->str, TSDB_COL_NAME_LEN);
   tsSmlDataFormat = cfgGetItem(pCfg, "smlDataFormat")->bval;
 
+  tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32;
+
   tsShellActivityTimer = cfgGetItem(pCfg, "shellActivityTimer")->i32;
   tsCompressMsgSize = cfgGetItem(pCfg, "compressMsgSize")->i32;
   tsCompressColData = cfgGetItem(pCfg, "compressColData")->i32;
@@ -877,6 +884,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) {
             tsMaxShellConns = cfgGetItem(pCfg, "maxShellConns")->i32;
           } else if (strcasecmp("maxNumOfDistinctRes", name) == 0) {
             tsMaxNumOfDistinctResults = cfgGetItem(pCfg, "maxNumOfDistinctRes")->i32;
+          } else if (strcasecmp("maxMemUsedByInsert", name) == 0) {
+            tsMaxMemUsedByInsert = cfgGetItem(pCfg, "maxMemUsedByInsert")->i32;
           }
           break;
         }

source/common/src/ttszip.c

@@ -52,10 +52,13 @@ STSBuf* tsBufCreate(bool autoDelete, int32_t order) {
   }
 
   if (!autoDelete) {
-    taosRemoveFile(pTSBuf->path);
+    if (taosRemoveFile(pTSBuf->path) != 0) {
+      taosMemoryFree(pTSBuf);
+      return NULL;
+    }
   }
 
-  if (NULL == allocResForTSBuf(pTSBuf)) {
+  if (allocResForTSBuf(pTSBuf) == NULL) {
     return NULL;
   }
 

source/dnode/mgmt/CMakeLists.txt

@@ -14,7 +14,4 @@ target_include_directories(
     taosd
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/node_mgmt/inc"
 )
-IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
-    add_dependencies(taosd jemalloc)
-ENDIF ()
 target_link_libraries(taosd dnode)

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

@@ -201,14 +201,13 @@ _OVER:
   taosCloseFile(&pFile);
   taosMemoryFree(content);
 
+  if (ppVnodes != NULL) {
     for (int32_t i = 0; i < numOfVnodes; ++i) {
       SVnodeObj *pVnode = ppVnodes[i];
       if (pVnode != NULL) {
         vmReleaseVnode(pMgmt, pVnode);
       }
     }
-
-  if (ppVnodes != NULL) {
     taosMemoryFree(ppVnodes);
   }
 

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

@@ -16,7 +16,7 @@
 #define _DEFAULT_SOURCE
 #include "vmInt.h"
 
-void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
+void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo, bool isReset) {
   pInfo->pVloads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoad));
   if (pInfo->pVloads == NULL) return;
 
@@ -30,6 +30,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
     SVnodeObj *pVnode = *ppVnode;
     SVnodeLoad vload = {0};
     vnodeGetLoad(pVnode->pImpl, &vload);
+    if (isReset) vnodeResetLoad(pVnode->pImpl, &vload);
     taosArrayPush(pInfo->pVloads, &vload);
     pIter = taosHashIterate(pMgmt->hash, pIter);
   }
@@ -39,7 +40,7 @@ void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo) {
 
 void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) {
   SMonVloadInfo vloads = {0};
-  vmGetVnodeLoads(pMgmt, &vloads);
+  vmGetVnodeLoads(pMgmt, &vloads, true);
 
   SArray *pVloads = vloads.pVloads;
   if (pVloads == NULL) return;
@@ -66,10 +67,10 @@ void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) {
   pInfo->vstat.totalVnodes = totalVnodes;
   pInfo->vstat.masterNum = masterNum;
   pInfo->vstat.numOfSelectReqs = numOfSelectReqs - pMgmt->state.numOfSelectReqs;
-  pInfo->vstat.numOfInsertReqs = numOfInsertReqs - pMgmt->state.numOfInsertReqs;
+  pInfo->vstat.numOfInsertReqs = numOfInsertReqs;                          // delta
-  pInfo->vstat.numOfInsertSuccessReqs = numOfInsertSuccessReqs - pMgmt->state.numOfInsertSuccessReqs;
+  pInfo->vstat.numOfInsertSuccessReqs = numOfInsertSuccessReqs;            // delta
-  pInfo->vstat.numOfBatchInsertReqs = numOfBatchInsertReqs - pMgmt->state.numOfBatchInsertReqs;
+  pInfo->vstat.numOfBatchInsertReqs = numOfBatchInsertReqs;                // delta
-  pInfo->vstat.numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs - pMgmt->state.numOfBatchInsertSuccessReqs;
+  pInfo->vstat.numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs;  // delta
   pMgmt->state.totalVnodes = totalVnodes;
   pMgmt->state.masterNum = masterNum;
   pMgmt->state.numOfSelectReqs = numOfSelectReqs;
@@ -109,7 +110,7 @@ int32_t vmProcessGetMonitorInfoReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
 int32_t vmProcessGetLoadsReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
   SMonVloadInfo vloads = {0};
-  vmGetVnodeLoads(pMgmt, &vloads);
+  vmGetVnodeLoads(pMgmt, &vloads, false);
 
   int32_t rspLen = tSerializeSMonVloadInfo(NULL, 0, &vloads);
   if (rspLen < 0) {
@@ -212,38 +213,47 @@ static int32_t vmTsmaProcessCreate(SVnode *pVnode, SCreateVnodeReq *pReq) {
 }
 
 int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
-  SCreateVnodeReq createReq = {0};
+  SCreateVnodeReq req = {0};
   SVnodeCfg       vnodeCfg = {0};
   SWrapperCfg     wrapperCfg = {0};
   int32_t         code = -1;
   char            path[TSDB_FILENAME_LEN] = {0};
 
-  if (tDeserializeSCreateVnodeReq(pMsg->pCont, pMsg->contLen, &createReq) != 0) {
+  if (tDeserializeSCreateVnodeReq(pMsg->pCont, pMsg->contLen, &req) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return -1;
   }
 
   dInfo(
-      "vgId:%d, start to create vnode, tsma:%d standby:%d cacheLast:%d cacheLastSize:%d sstTrigger:%d "
+      "vgId:%d, start to create vnode, page:%d pageSize:%d buffer:%d szPage:%d szBuf:%" PRIu64
-      "tsdbPageSize:%d",
+      " cacheLast:%d cacheLastSize:%d sstTrigger:%d tsdbPageSize:%d %d dbname:%s dbId:%" PRId64
-      createReq.vgId, createReq.isTsma, createReq.standby, createReq.cacheLast, createReq.cacheLastSize,
+      "days:%d keep0:%d keep1:%d keep2:%d tsma:%d precision:%d compression:%d minRows:%d maxRows:%d, wal "
-      createReq.sstTrigger, createReq.tsdbPageSize);
+      "fsync:%d level:%d retentionPeriod:%d retentionSize:%d rollPeriod:%d segSize:%d, hash method:%d begin:%u end:%u "
-  dInfo("vgId:%d, hashMethod:%d begin:%u end:%u prefix:%d surfix:%d", createReq.vgId, createReq.hashMethod,
+      "prefix:%d surfix:%d replica:%d selfIndex:%d strict:%d",
-        createReq.hashBegin, createReq.hashEnd, createReq.hashPrefix, createReq.hashSuffix);
+      req.vgId, req.pages, req.pageSize, req.buffer, req.pageSize * 1024, (uint64_t)req.buffer * 1024 * 1024,
-  vmGenerateVnodeCfg(&createReq, &vnodeCfg);
+      req.cacheLast, req.cacheLastSize, req.sstTrigger, req.tsdbPageSize, req.tsdbPageSize * 1024, req.db, req.dbUid,
+      req.daysPerFile, req.daysToKeep0, req.daysToKeep1, req.daysToKeep2, req.isTsma, req.precision, req.compression,
+      req.minRows, req.maxRows, req.walFsyncPeriod, req.walLevel, req.walRetentionPeriod, req.walRetentionSize,
+      req.walRollPeriod, req.walSegmentSize, req.hashMethod, req.hashBegin, req.hashEnd, req.hashPrefix, req.hashSuffix,
+      req.replica, req.selfIndex, req.strict);
+  for (int32_t i = 0; i < req.replica; ++i) {
+    dInfo("vgId:%d, replica:%d fqdn:%s port:%u", req.vgId, req.replicas[i].id, req.replicas[i].fqdn,
+          req.replicas[i].port);
+  }
+  vmGenerateVnodeCfg(&req, &vnodeCfg);
 
-  if (vmTsmaAdjustDays(&vnodeCfg, &createReq) < 0) {
+  if (vmTsmaAdjustDays(&vnodeCfg, &req) < 0) {
-    dError("vgId:%d, failed to adjust tsma days since %s", createReq.vgId, terrstr());
+    dError("vgId:%d, failed to adjust tsma days since %s", req.vgId, terrstr());
     code = terrno;
     goto _OVER;
   }
 
-  vmGenerateWrapperCfg(pMgmt, &createReq, &wrapperCfg);
+  vmGenerateWrapperCfg(pMgmt, &req, &wrapperCfg);
 
-  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, createReq.vgId);
+  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
   if (pVnode != NULL) {
-    dDebug("vgId:%d, already exist", createReq.vgId);
+    dDebug("vgId:%d, already exist", req.vgId);
-    tFreeSCreateVnodeReq(&createReq);
+    tFreeSCreateVnodeReq(&req);
     vmReleaseVnode(pMgmt, pVnode);
     terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
     code = terrno;
@@ -252,36 +262,36 @@ int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
 
   snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
   if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
-    tFreeSCreateVnodeReq(&createReq);
+    tFreeSCreateVnodeReq(&req);
-    dError("vgId:%d, failed to create vnode since %s", createReq.vgId, terrstr());
+    dError("vgId:%d, failed to create vnode since %s", req.vgId, terrstr());
     code = terrno;
     goto _OVER;
   }
 
   SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
   if (pImpl == NULL) {
-    dError("vgId:%d, failed to open vnode since %s", createReq.vgId, terrstr());
+    dError("vgId:%d, failed to open vnode since %s", req.vgId, terrstr());
     code = terrno;
     goto _OVER;
   }
 
   code = vmOpenVnode(pMgmt, &wrapperCfg, pImpl);
   if (code != 0) {
-    dError("vgId:%d, failed to open vnode since %s", createReq.vgId, terrstr());
+    dError("vgId:%d, failed to open vnode since %s", req.vgId, terrstr());
     code = terrno;
     goto _OVER;
   }
 
-  code = vmTsmaProcessCreate(pImpl, &createReq);
+  code = vmTsmaProcessCreate(pImpl, &req);
   if (code != 0) {
-    dError("vgId:%d, failed to create tsma since %s", createReq.vgId, terrstr());
+    dError("vgId:%d, failed to create tsma since %s", req.vgId, terrstr());
     code = terrno;
     goto _OVER;
   }
 
   code = vnodeStart(pImpl);
   if (code != 0) {
-    dError("vgId:%d, failed to start sync since %s", createReq.vgId, terrstr());
+    dError("vgId:%d, failed to start sync since %s", req.vgId, terrstr());
     goto _OVER;
   }
 
@@ -296,10 +306,10 @@ _OVER:
     vnodeClose(pImpl);
     vnodeDestroy(path, pMgmt->pTfs);
   } else {
-    dInfo("vgId:%d, vnode is created", createReq.vgId);
+    dInfo("vgId:%d, vnode is created", req.vgId);
   }
 
-  tFreeSCreateVnodeReq(&createReq);
+  tFreeSCreateVnodeReq(&req);
   terrno = code;
   return code;
 }

source/dnode/mgmt/node_mgmt/inc/dmNodes.h

@@ -35,7 +35,7 @@ void qmGetMonitorInfo(void *pMgmt, SMonQmInfo *pInfo);
 void smGetMonitorInfo(void *pMgmt, SMonSmInfo *pInfo);
 void bmGetMonitorInfo(void *pMgmt, SMonBmInfo *pInfo);
 
-void vmGetVnodeLoads(void *pMgmt, SMonVloadInfo *pInfo);
+void vmGetVnodeLoads(void *pMgmt, SMonVloadInfo *pInfo, bool isReset);
 void mmGetMnodeLoads(void *pMgmt, SMonMloadInfo *pInfo);
 void qmGetQnodeLoads(void *pMgmt, SQnodeLoad *pInfo);
 

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

@@ -152,7 +152,7 @@ void dmGetVnodeLoads(SMonVloadInfo *pInfo) {
     if (tsMultiProcess) {
       dmSendLocalRecv(pDnode, TDMT_MON_VM_LOAD, tDeserializeSMonVloadInfo, pInfo);
     } else if (pWrapper->pMgmt != NULL) {
-      vmGetVnodeLoads(pWrapper->pMgmt, pInfo);
+      vmGetVnodeLoads(pWrapper->pMgmt, pInfo, false);
     }
     dmReleaseWrapper(pWrapper);
   }

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

@@ -44,7 +44,8 @@ int32_t mndAddAlterVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pV
 int32_t mndAddDropVnodeAction(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid, bool isRedo);
 int32_t mndSetMoveVgroupInfoToTrans(SMnode *, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, int32_t vn, SArray *pArray);
 int32_t mndSetMoveVgroupsInfoToTrans(SMnode *, STrans *pTrans, int32_t dropDnodeId);
-int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SArray *pArray);
+int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb, SDbObj *pNewDb, SVgObj *pVgroup,
+                                  SArray *pArray);
 
 void *mndBuildCreateVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *cntlen, bool standby);
 void *mndBuildDropVnodeReq(SMnode *, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);

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

@@ -746,7 +746,7 @@ static int32_t mndSetAlterDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *p
   return 0;
 }
 
-static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
+static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb, SDbObj *pNewDb) {
   SSdb   *pSdb = pMnode->pSdb;
   void   *pIter = NULL;
   SArray *pArray = mndBuildDnodesArray(pMnode, 0);
@@ -756,8 +756,8 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
     pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
     if (pIter == NULL) break;
 
-    if (mndVgroupInDb(pVgroup, pNew->uid)) {
+    if (mndVgroupInDb(pVgroup, pNewDb->uid)) {
-      if (mndBuildAlterVgroupAction(pMnode, pTrans, pNew, pVgroup, pArray) != 0) {
+      if (mndBuildAlterVgroupAction(pMnode, pTrans, pOldDb, pNewDb, pVgroup, pArray) != 0) {
         sdbCancelFetch(pSdb, pIter);
         sdbRelease(pSdb, pVgroup);
         taosArrayDestroy(pArray);

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

@@ -117,9 +117,13 @@ int32_t tDecodeSStreamObj(SDecoder *pDecoder, SStreamObj *pObj) {
       SArray *pArray = taosArrayInit(innerSz, sizeof(void *));
       for (int32_t j = 0; j < innerSz; j++) {
         SStreamTask *pTask = taosMemoryCalloc(1, sizeof(SStreamTask));
-        if (pTask == NULL) return -1;
+        if (pTask == NULL) {
+          taosArrayDestroy(pArray);
+          return -1;
+        }
         if (tDecodeSStreamTask(pDecoder, pTask) < 0) {
           taosMemoryFree(pTask);
+          taosArrayDestroy(pArray);
           return -1;
         }
         taosArrayPush(pArray, &pTask);

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

@@ -87,7 +87,7 @@ static int32_t mndCreateDefaultMnode(SMnode *pMnode) {
 
   SSdbRaw *pRaw = mndMnodeActionEncode(&mnodeObj);
   if (pRaw == NULL) return -1;
-  sdbSetRawStatus(pRaw, SDB_STATUS_READY);
+  (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
 
   mInfo("mnode:%d, will be created when deploying, raw:%p", mnodeObj.id, pRaw);
 

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

@@ -1530,45 +1530,81 @@ _OVER:
 #endif
 }
 
-int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SArray *pArray) {
+static int32_t mndCheckDnodeMemory(SMnode *pMnode, SDbObj *pOldDb, SDbObj *pNewDb, SVgObj *pOldVgroup,
-  if (pVgroup->replica <= 0 || pVgroup->replica == pDb->cfg.replications) {
+                                   SVgObj *pNewVgroup, SArray *pArray) {
-    if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup, TDMT_VND_ALTER_CONFIG) != 0) {
+  for (int32_t i = 0; i < (int32_t)taosArrayGetSize(pArray); ++i) {
-      return -1;
+    SDnodeObj *pDnode = taosArrayGet(pArray, i);
+    bool       inVgroup = false;
+    for (int32_t j = 0; j < pOldVgroup->replica; ++j) {
+      SVnodeGid *pVgId = &pOldVgroup->vnodeGid[i];
+      if (pDnode->id == pVgId->dnodeId) {
+        pDnode->memUsed -= mndGetVgroupMemory(pMnode, pOldDb, pOldVgroup);
+        inVgroup = true;
       }
+    }
+    for (int32_t j = 0; j < pNewVgroup->replica; ++j) {
+      SVnodeGid *pVgId = &pNewVgroup->vnodeGid[i];
+      if (pDnode->id == pVgId->dnodeId) {
+        pDnode->memUsed += mndGetVgroupMemory(pMnode, pNewDb, pNewVgroup);
+        inVgroup = true;
+      }
+    }
+    if (pDnode->memAvail - pDnode->memUsed <= 0) {
+      mError("db:%s, vgId:%d, no enough memory in dnode:%d, avail:%" PRId64 " used:%" PRId64, pNewVgroup->dbName,
+             pNewVgroup->vgId, pDnode->id, pDnode->memAvail, pDnode->memUsed);
+      terrno = TSDB_CODE_MND_NO_ENOUGH_MEM_IN_DNODE;
+      return -1;
+    } else if (inVgroup) {
+      mInfo("db:%s, vgId:%d, memory in dnode:%d, avail:%" PRId64 " used:%" PRId64, pNewVgroup->dbName,
+             pNewVgroup->vgId, pDnode->id, pDnode->memAvail, pDnode->memUsed);
     } else {
+    }
+  }
+  return 0;
+}
+
+int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pOldDb, SDbObj *pNewDb, SVgObj *pVgroup,
+                                  SArray *pArray) {
   SVgObj newVgroup = {0};
   memcpy(&newVgroup, pVgroup, sizeof(SVgObj));
+
+  if (pVgroup->replica <= 0 || pVgroup->replica == pNewDb->cfg.replications) {
+    if (mndAddAlterVnodeAction(pMnode, pTrans, pNewDb, pVgroup, TDMT_VND_ALTER_CONFIG) != 0) return -1;
+    if (mndCheckDnodeMemory(pMnode, pOldDb, pNewDb, &newVgroup, pVgroup, pArray) != 0) return -1;
+    return 0;
+  }
+
   mndTransSetSerial(pTrans);
 
-    if (newVgroup.replica < pDb->cfg.replications) {
+  if (newVgroup.replica < pNewDb->cfg.replications) {
     mInfo("db:%s, vgId:%d, vn:0 dnode:%d, will add 2 vnodes", pVgroup->dbName, pVgroup->vgId,
           pVgroup->vnodeGid[0].dnodeId);
 
     if (mndAddVnodeToVgroup(pMnode, &newVgroup, pArray) != 0) return -1;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[1], true) != 0) return -1;
+    if (mndAddCreateVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, &newVgroup.vnodeGid[1], true) != 0) return -1;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
+    if (mndAddAlterVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
-      if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
+    if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pNewDb, &newVgroup) != 0) return -1;
 
     if (mndAddVnodeToVgroup(pMnode, &newVgroup, pArray) != 0) return -1;
-      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[2], true) != 0) return -1;
+    if (mndAddCreateVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, &newVgroup.vnodeGid[2], true) != 0) return -1;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
+    if (mndAddAlterVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
-      if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
+    if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pNewDb, &newVgroup) != 0) return -1;
-    } else if (newVgroup.replica > pDb->cfg.replications) {
+  } else if (newVgroup.replica > pNewDb->cfg.replications) {
     mInfo("db:%s, vgId:%d, will remove 2 vnodes", pVgroup->dbName, pVgroup->vgId);
 
     SVnodeGid del1 = {0};
     if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del1) != 0) return -1;
-      if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del1, true) != 0) return -1;
+    if (mndAddSetVnodeStandByAction(pMnode, pTrans, pNewDb, pVgroup, &del1, true) != 0) return -1;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
+    if (mndAddAlterVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
-      if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del1, true) != 0) return -1;
+    if (mndAddDropVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, &del1, true) != 0) return -1;
-      if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
+    if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pNewDb, &newVgroup) != 0) return -1;
 
     SVnodeGid del2 = {0};
     if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del2) != 0) return -1;
-      if (mndAddSetVnodeStandByAction(pMnode, pTrans, pDb, pVgroup, &del2, true) != 0) return -1;
+    if (mndAddSetVnodeStandByAction(pMnode, pTrans, pNewDb, pVgroup, &del2, true) != 0) return -1;
-      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
+    if (mndAddAlterVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, TDMT_VND_ALTER_REPLICA) != 0) return -1;
-      if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del2, true) != 0) return -1;
+    if (mndAddDropVnodeAction(pMnode, pTrans, pNewDb, &newVgroup, &del2, true) != 0) return -1;
-      if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pDb, &newVgroup) != 0) return -1;
+    if (mndAddAlterVnodeConfirmAction(pMnode, pTrans, pNewDb, &newVgroup) != 0) return -1;
   } else {
   }
 
@@ -1591,7 +1627,6 @@ int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, S
     }
     (void)sdbSetRawStatus(pVgRaw, SDB_STATUS_READY);
   }
-  }
 
   return 0;
 }
@@ -1650,8 +1685,8 @@ static int32_t mndSplitVgroup(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SVgObj
   if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVg2, TDMT_VND_ALTER_HASHRANGE) != 0) goto _OVER;
 
   // adjust vgroup
-  if (mndBuildAlterVgroupAction(pMnode, pTrans, pDb, &newVg1, pArray) != 0) goto _OVER;
+  if (mndBuildAlterVgroupAction(pMnode, pTrans, pDb, pDb, &newVg1, pArray) != 0) goto _OVER;
-  if (mndBuildAlterVgroupAction(pMnode, pTrans, pDb, &newVg2, pArray) != 0) goto _OVER;
+  if (mndBuildAlterVgroupAction(pMnode, pTrans, pDb, pDb, &newVg2, pArray) != 0) goto _OVER;
 
 _OVER:
   mndTransDrop(pTrans);

source/dnode/vnode/inc/vnode.h

@@ -73,6 +73,7 @@ int32_t vnodeGetCtbNum(SVnode *pVnode, int64_t suid, int64_t *num);
 int32_t vnodeGetTimeSeriesNum(SVnode *pVnode, int64_t *num);
 int32_t vnodeGetAllCtbNum(SVnode *pVnode, int64_t *num);
 
+void    vnodeResetLoad(SVnode *pVnode, SVnodeLoad *pLoad);
 int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad);
 int32_t vnodeValidateTableHash(SVnode *pVnode, char *tableFName);
 

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

@@ -56,6 +56,7 @@ typedef struct SSma               SSma;
 typedef struct STsdb              STsdb;
 typedef struct STQ                STQ;
 typedef struct SVState            SVState;
+typedef struct SVStatis           SVStatis;
 typedef struct SVBufPool          SVBufPool;
 typedef struct SQWorker           SQHandle;
 typedef struct STsdbKeepCfg       STsdbKeepCfg;
@@ -284,9 +285,17 @@ struct SVState {
   int64_t commitTerm;
 };
 
+struct SVStatis {
+  int64_t nInsert;              // delta
+  int64_t nInsertSuccess;       // delta
+  int64_t nBatchInsert;         // delta
+  int64_t nBatchInsertSuccess;  // delta
+};
+
 struct SVnodeInfo {
   SVnodeCfg config;
   SVState   state;
+  SVStatis  statis;
 };
 
 typedef enum {
@@ -309,6 +318,7 @@ struct SVnode {
   char*         path;
   SVnodeCfg     config;
   SVState       state;
+  SVStatis      statis;
   STfs*         pTfs;
   SMsgCb        msgCb;
   TdThreadMutex mutex;

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

@@ -180,6 +180,7 @@ int32_t tqMetaRestoreCheckInfo(STQ* pTq) {
       return -1;
     }
   }
+  tdbFree(pKey);
   tdbTbcClose(pCur);
   return 0;
 }

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

@@ -54,7 +54,7 @@ int32_t tqOffsetSnapRead(STqOffsetReader* pReader, uint8_t** ppData) {
 
   char*     fname = tqOffsetBuildFName(pReader->pTq->path, 0);
   TdFilePtr pFile = taosOpenFile(fname, TD_FILE_READ);
-  if (pFile != NULL) {
+  if (pFile == NULL) {
     taosMemoryFree(fname);
     return 0;
   }

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

@@ -15,6 +15,12 @@
 
 #include "vnd.h"
 
+#define VNODE_GET_LOAD_RESET_VALS(pVar, oVal, vType)                   \
+  do {                                                                 \
+    int##vType##_t newVal = atomic_sub_fetch_##vType(&(pVar), (oVal)); \
+    ASSERT(newVal >= 0);                                               \
+  } while (0)
+
 int vnodeQueryOpen(SVnode *pVnode) {
   return qWorkerInit(NODE_TYPE_VNODE, TD_VID(pVnode), (void **)&pVnode->pQuery, &pVnode->msgCb);
 }
@@ -375,13 +381,28 @@ int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
   pLoad->compStorage = (int64_t)2 * 1073741824;
   pLoad->pointsWritten = 100;
   pLoad->numOfSelectReqs = 1;
-  pLoad->numOfInsertReqs = 3;
+  pLoad->numOfInsertReqs = atomic_load_64(&pVnode->statis.nInsert);
-  pLoad->numOfInsertSuccessReqs = 2;
+  pLoad->numOfInsertSuccessReqs = atomic_load_64(&pVnode->statis.nInsertSuccess);
-  pLoad->numOfBatchInsertReqs = 5;
+  pLoad->numOfBatchInsertReqs = atomic_load_64(&pVnode->statis.nBatchInsert);
-  pLoad->numOfBatchInsertSuccessReqs = 4;
+  pLoad->numOfBatchInsertSuccessReqs = atomic_load_64(&pVnode->statis.nBatchInsertSuccess);
   return 0;
 }
 
+/**
+ * @brief Reset the statistics value by monitor interval
+ *
+ * @param pVnode
+ * @param pLoad
+ */
+void vnodeResetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
+  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nInsert, pLoad->numOfInsertReqs, 64);
+  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nInsertSuccess, pLoad->numOfInsertSuccessReqs, 64);
+  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nBatchInsert, pLoad->numOfBatchInsertReqs, 64);
+  VNODE_GET_LOAD_RESET_VALS(pVnode->statis.nBatchInsertSuccess, pLoad->numOfBatchInsertSuccessReqs, 64);
+
+
+}
+
 void vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId) {
   if (dbname) {
     *dbname = pVnode->config.dbname;

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

@@ -812,10 +812,10 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
   SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
   SSubmitRsp     submitRsp = {0};
   SSubmitMsgIter msgIter = {0};
-  SSubmitBlk    *pBlock;
+  SSubmitBlk    *pBlock = NULL;
   SVCreateTbReq  createTbReq = {0};
   SDecoder       decoder = {0};
-  int32_t        nRows;
+  int32_t        nRows = 0;
   int32_t        tsize, ret;
   SEncoder       encoder = {0};
   SArray        *newTbUids = NULL;
@@ -823,6 +823,7 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq
 
   pRsp->code = 0;
   pSubmitReq->version = version;
+  atomic_fetch_add_64(&pVnode->statis.nBatchInsert, 1);
 
 #ifdef TD_DEBUG_PRINT_ROW
   vnodeDebugPrintSubmitMsg(pVnode, pReq, __func__);
@@ -942,11 +943,15 @@ _exit:
 
   taosArrayDestroyEx(submitRsp.pArray, tFreeSSubmitBlkRsp);
 
+  atomic_fetch_add_64(&pVnode->statis.nInsert, submitRsp.numOfRows);
+  atomic_fetch_add_64(&pVnode->statis.nInsertSuccess, submitRsp.affectedRows);
+
   // TODO: the partial success scenario and the error case
   // => If partial success, extract the success submitted rows and reconstruct a new submit msg, and push to level
   // 1/level 2.
   // TODO: refactor
   if ((terrno == TSDB_CODE_SUCCESS) && (pRsp->code == TSDB_CODE_SUCCESS)) {
+    atomic_fetch_add_64(&pVnode->statis.nBatchInsertSuccess, 1);
     tdProcessRSmaSubmit(pVnode->pSma, pReq, STREAM_INPUT__DATA_SUBMIT);
   }
 
@@ -1024,71 +1029,75 @@ static int32_t vnodeProcessAlterHashRangeReq(SVnode *pVnode, int64_t version, vo
 }
 
 static int32_t vnodeProcessAlterConfigReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
-  SAlterVnodeReq alterReq = {0};
+  SAlterVnodeReq req = {0};
   bool           walChanged = false;
   bool           tsdbChanged = false;
 
-  if (tDeserializeSAlterVnodeReq(pReq, len, &alterReq) != 0) {
+  if (tDeserializeSAlterVnodeReq(pReq, len, &req) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
     return TSDB_CODE_INVALID_MSG;
   }
 
-  vInfo("vgId:%d, start to alter vnode config, cacheLast:%d cacheLastSize:%d", TD_VID(pVnode), alterReq.cacheLast,
+  vInfo("vgId:%d, start to alter vnode config, page:%d pageSize:%d buffer:%d szPage:%d szBuf:%" PRIu64
-        alterReq.cacheLastSize);
+        " cacheLast:%d cacheLastSize:%d days:%d keep0:%d keep1:%d keep2:%d fsync:%d level:%d strict:%d",
-  if (pVnode->config.cacheLastSize != alterReq.cacheLastSize) {
+        TD_VID(pVnode), req.pages, req.pageSize, req.buffer, req.pageSize * 1024, (uint64_t)req.buffer * 1024 * 1024,
-    pVnode->config.cacheLastSize = alterReq.cacheLastSize;
+        req.cacheLast, req.cacheLastSize, req.daysPerFile, req.daysToKeep0, req.daysToKeep1, req.daysToKeep2,
+        req.walFsyncPeriod, req.walLevel, req.strict);
+
+  if (pVnode->config.cacheLastSize != req.cacheLastSize) {
+    pVnode->config.cacheLastSize = req.cacheLastSize;
     tsdbCacheSetCapacity(pVnode, (size_t)pVnode->config.cacheLastSize * 1024 * 1024);
   }
 
-  if (pVnode->config.szBuf != alterReq.buffer * 1024LL * 1024LL) {
+  if (pVnode->config.szBuf != req.buffer * 1024LL * 1024LL) {
     vInfo("vgId:%d vnode buffer is changed from %" PRId64 " to %" PRId64, TD_VID(pVnode), pVnode->config.szBuf,
-          alterReq.buffer * 1024LL * 1024LL);
+          req.buffer * 1024LL * 1024LL);
-    pVnode->config.szBuf = alterReq.buffer * 1024LL * 1024LL;
+    pVnode->config.szBuf = req.buffer * 1024LL * 1024LL;
   }
 
-  if (pVnode->config.szCache != alterReq.pages) {
+  if (pVnode->config.szCache != req.pages) {
-    if (metaAlterCache(pVnode->pMeta, alterReq.pages) < 0) {
+    if (metaAlterCache(pVnode->pMeta, req.pages) < 0) {
       vError("vgId:%d failed to change vnode pages from %d to %d failed since %s", TD_VID(pVnode),
-             pVnode->config.szCache, alterReq.pages, tstrerror(errno));
+             pVnode->config.szCache, req.pages, tstrerror(errno));
       return errno;
     } else {
-      vInfo("vgId:%d vnode pages is changed from %d to %d", TD_VID(pVnode), pVnode->config.szCache, alterReq.pages);
+      vInfo("vgId:%d vnode pages is changed from %d to %d", TD_VID(pVnode), pVnode->config.szCache, req.pages);
-      pVnode->config.szCache = alterReq.pages;
+      pVnode->config.szCache = req.pages;
     }
   }
 
-  if (pVnode->config.cacheLast != alterReq.cacheLast) {
+  if (pVnode->config.cacheLast != req.cacheLast) {
-    pVnode->config.cacheLast = alterReq.cacheLast;
+    pVnode->config.cacheLast = req.cacheLast;
   }
 
-  if (pVnode->config.walCfg.fsyncPeriod != alterReq.walFsyncPeriod) {
+  if (pVnode->config.walCfg.fsyncPeriod != req.walFsyncPeriod) {
-    pVnode->config.walCfg.fsyncPeriod = alterReq.walFsyncPeriod;
+    pVnode->config.walCfg.fsyncPeriod = req.walFsyncPeriod;
 
     walChanged = true;
   }
 
-  if (pVnode->config.walCfg.level != alterReq.walLevel) {
+  if (pVnode->config.walCfg.level != req.walLevel) {
-    pVnode->config.walCfg.level = alterReq.walLevel;
+    pVnode->config.walCfg.level = req.walLevel;
 
     walChanged = true;
   }
 
-  if (pVnode->config.tsdbCfg.keep0 != alterReq.daysToKeep0) {
+  if (pVnode->config.tsdbCfg.keep0 != req.daysToKeep0) {
-    pVnode->config.tsdbCfg.keep0 = alterReq.daysToKeep0;
+    pVnode->config.tsdbCfg.keep0 = req.daysToKeep0;
     if (!VND_IS_RSMA(pVnode)) {
       tsdbChanged = true;
     }
   }
 
-  if (pVnode->config.tsdbCfg.keep1 != alterReq.daysToKeep1) {
+  if (pVnode->config.tsdbCfg.keep1 != req.daysToKeep1) {
-    pVnode->config.tsdbCfg.keep1 = alterReq.daysToKeep1;
+    pVnode->config.tsdbCfg.keep1 = req.daysToKeep1;
     if (!VND_IS_RSMA(pVnode)) {
       tsdbChanged = true;
     }
   }
 
-  if (pVnode->config.tsdbCfg.keep2 != alterReq.daysToKeep2) {
+  if (pVnode->config.tsdbCfg.keep2 != req.daysToKeep2) {
-    pVnode->config.tsdbCfg.keep2 = alterReq.daysToKeep2;
+    pVnode->config.tsdbCfg.keep2 = req.daysToKeep2;
     if (!VND_IS_RSMA(pVnode)) {
       tsdbChanged = true;
     }

source/libs/catalog/src/ctgUtil.c

@@ -1193,4 +1193,232 @@ SName* ctgGetFetchName(SArray* pNames, SCtgFetch* pFetch) {
   return (SName*)taosArrayGet(pReq->pTables, pFetch->tbIdx);
 }
 
+static void* ctgCloneDbVgroup(void* pSrc) {
+  return taosArrayDup((const SArray*)pSrc);
+}
 
+static void ctgFreeDbVgroup(void* p) {
+  taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneDbCfgInfo(void* pSrc) {
+  SDbCfgInfo* pDst = taosMemoryMalloc(sizeof(SDbCfgInfo));
+  if (NULL == pDst) {
+    return NULL;
+  }
+  memcpy(pDst, pSrc, sizeof(SDbCfgInfo));
+  return pDst;
+}
+
+static void ctgFreeDbCfgInfo(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneDbInfo(void* pSrc) {
+  SDbInfo* pDst = taosMemoryMalloc(sizeof(SDbInfo));
+  if (NULL == pDst) {
+    return NULL;
+  }
+  memcpy(pDst, pSrc, sizeof(SDbInfo));
+  return pDst;
+}
+
+static void ctgFreeDbInfo(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneTableMeta(void* pSrc) {
+  STableMeta* pMeta = pSrc;
+  int32_t size = sizeof(STableMeta) + (pMeta->tableInfo.numOfColumns + pMeta->tableInfo.numOfTags) * sizeof(SSchema);
+  STableMeta* pDst = taosMemoryMalloc(size);
+  if (NULL == pDst) {
+    return NULL;
+  }
+  memcpy(pDst, pSrc, size);
+  return pDst;
+}
+
+static void ctgFreeTableMeta(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneVgroupInfo(void* pSrc) {
+  SVgroupInfo* pDst = taosMemoryMalloc(sizeof(SVgroupInfo));
+  if (NULL == pDst) {
+    return NULL;
+  }
+  memcpy(pDst, pSrc, sizeof(SVgroupInfo));
+  return pDst;
+}
+
+static void ctgFreeVgroupInfo(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneTableIndices(void* pSrc) {
+  return taosArrayDup((const SArray*)pSrc);
+}
+
+static void ctgFreeTableIndices(void* p) {
+  taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneFuncInfo(void* pSrc) {
+  SFuncInfo* pDst = taosMemoryMalloc(sizeof(SFuncInfo));
+  if (NULL == pDst) {
+    return NULL;
+  }
+  memcpy(pDst, pSrc, sizeof(SFuncInfo));
+  return pDst;
+}
+
+static void ctgFreeFuncInfo(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneIndexInfo(void* pSrc) {
+  SIndexInfo* pDst = taosMemoryMalloc(sizeof(SIndexInfo));
+  if (NULL == pDst) {
+    return NULL;
+  }
+  memcpy(pDst, pSrc, sizeof(SIndexInfo));
+  return pDst;
+}
+
+static void ctgFreeIndexInfo(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneUserAuth(void* pSrc) {
+  bool* pDst = taosMemoryMalloc(sizeof(bool));
+  if (NULL == pDst) {
+    return NULL;
+  }
+  *pDst = *(bool*)pSrc;
+  return pDst;
+}
+
+static void ctgFreeUserAuth(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneQnodeList(void* pSrc) {
+  return taosArrayDup((const SArray*)pSrc);
+}
+
+static void ctgFreeQnodeList(void* p) {
+  taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneTableCfg(void* pSrc) {
+  STableCfg* pDst = taosMemoryMalloc(sizeof(STableCfg));
+  if (NULL == pDst) {
+    return NULL;
+  }
+  memcpy(pDst, pSrc, sizeof(STableCfg));
+  return pDst;
+}
+
+static void ctgFreeTableCfg(void* p) {
+  taosMemoryFree(((SMetaRes*)p)->pRes);
+}
+
+static void* ctgCloneDnodeList(void* pSrc) {
+  return taosArrayDup((const SArray*)pSrc);
+}
+
+static void ctgFreeDnodeList(void* p) {
+  taosArrayDestroy((SArray*)((SMetaRes*)p)->pRes);
+}
+
+static int32_t ctgCloneMetaDataArray(SArray* pSrc, FCopy copyFunc, SArray** pDst) {
+  if (NULL == pSrc) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  int32_t size = taosArrayGetSize(pSrc);
+  *pDst = taosArrayInit(size, sizeof(SMetaRes));
+  if (NULL == *pDst) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+  for (int32_t i = 0; i < size; ++i) {
+    SMetaRes* pRes = taosArrayGet(pSrc, i);
+    SMetaRes res = {.code = pRes->code, .pRes = copyFunc(pRes->pRes)};
+    if (NULL == res.pRes) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    taosArrayPush(*pDst, &res);
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+SMetaData* catalogCloneMetaData(SMetaData* pData) {
+  SMetaData* pRes = taosMemoryCalloc(1, sizeof(SMetaData));
+  if (NULL == pRes) {
+    return NULL;
+  }
+
+  int32_t code = ctgCloneMetaDataArray(pData->pDbVgroup, ctgCloneDbVgroup, &pRes->pDbVgroup);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pDbCfg, ctgCloneDbCfgInfo, &pRes->pDbCfg);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pDbInfo, ctgCloneDbInfo, &pRes->pDbInfo);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pTableMeta, ctgCloneTableMeta, &pRes->pTableMeta);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pTableHash, ctgCloneVgroupInfo, &pRes->pTableHash);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pTableIndex, ctgCloneTableIndices, &pRes->pTableIndex);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pUdfList, ctgCloneFuncInfo, &pRes->pUdfList);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pIndex, ctgCloneIndexInfo, &pRes->pIndex);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pUser, ctgCloneUserAuth, &pRes->pUser);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pQnodeList, ctgCloneQnodeList, &pRes->pQnodeList);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pTableCfg, ctgCloneTableCfg, &pRes->pTableCfg);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = ctgCloneMetaDataArray(pData->pDnodeList, ctgCloneDnodeList, &pRes->pDnodeList);
+  }
+
+  if (TSDB_CODE_SUCCESS != code) {
+    catalogFreeMetaData(pRes);
+    return NULL;
+  }
+
+  return pRes;
+}
+
+void catalogFreeMetaData(SMetaData* pData) {
+  if (NULL == pData) {
+    return;
+  }
+
+  taosArrayDestroyEx(pData->pDbVgroup, ctgFreeDbVgroup);
+  taosArrayDestroyEx(pData->pDbCfg, ctgFreeDbCfgInfo);
+  taosArrayDestroyEx(pData->pDbInfo, ctgFreeDbInfo);
+  taosArrayDestroyEx(pData->pTableMeta, ctgFreeTableMeta);
+  taosArrayDestroyEx(pData->pTableHash, ctgFreeVgroupInfo);
+  taosArrayDestroyEx(pData->pTableIndex, ctgFreeTableIndices);  
+  taosArrayDestroyEx(pData->pUdfList, ctgFreeFuncInfo); 
+  taosArrayDestroyEx(pData->pIndex, ctgFreeIndexInfo);
+  taosArrayDestroyEx(pData->pUser, ctgFreeUserAuth);
+  taosArrayDestroyEx(pData->pQnodeList, ctgFreeQnodeList);
+  taosArrayDestroyEx(pData->pTableCfg, ctgFreeTableCfg);
+  taosArrayDestroyEx(pData->pDnodeList, ctgFreeDnodeList);
+  taosMemoryFreeClear(pData->pSvrVer);
+  taosMemoryFree(pData);
+}

source/libs/executor/inc/executorimpl.h

@@ -767,6 +767,7 @@ typedef struct SStreamSessionAggOperatorInfo {
   SPhysiNode*         pPhyNode;   // create new child
   bool                isFinal;
   bool                ignoreExpiredData;
+  SHashObj*           pGroupIdTbNameMap;
 } SStreamSessionAggOperatorInfo;
 
 typedef struct SStreamPartitionOperatorInfo {
@@ -845,6 +846,7 @@ typedef struct SStreamStateAggOperatorInfo {
   void*               pDelIterator;
   SArray*             pChildren;  // cache for children's result;
   bool                ignoreExpiredData;
+  SHashObj*           pGroupIdTbNameMap;
 } SStreamStateAggOperatorInfo;
 
 typedef struct SSortOperatorInfo {
@@ -899,8 +901,12 @@ void    destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs);
 int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                     const char* pkey);
 void    initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows);
+
+void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
+                           SDiskbasedBuf* pBuf);
 void    doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                                SDiskbasedBuf* pBuf);
+
 int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf);
 bool    hasLimitOffsetInfo(SLimitInfo* pLimitInfo);
 void    initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLimitInfo);

source/libs/executor/src/executorimpl.c

@@ -1389,6 +1389,46 @@ int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprS
   return 0;
 }
 
+void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
+                           SDiskbasedBuf* pBuf) {
+  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
+  SSDataBlock*   pBlock = pbInfo->pRes;
+
+  // set output datablock version
+  pBlock->info.version = pTaskInfo->version;
+
+  blockDataCleanup(pBlock);
+  if (!hasRemainResults(pGroupResInfo)) {
+    return;
+  }
+
+  // clear the existed group id
+  pBlock->info.groupId = 0;
+  ASSERT(!pbInfo->mergeResultBlock);
+  doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo);
+  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
+    SStreamStateAggOperatorInfo* pInfo = pOperator->info;
+
+    char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
+    if (tbname != NULL) {
+      memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
+    } else {
+      pBlock->info.parTbName[0] = 0;
+    }
+  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
+             pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
+             pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
+    SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
+
+    char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
+    if (tbname != NULL) {
+      memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
+    } else {
+      pBlock->info.parTbName[0] = 0;
+    }
+  }
+}
+
 void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                             SDiskbasedBuf* pBuf) {
   SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

source/libs/executor/src/timewindowoperator.c

@@ -3728,6 +3728,9 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
   pInfo->pPhyNode = pPhyNode;
   pInfo->ignoreExpiredData = pSessionNode->window.igExpired;
 
+  pInfo->pGroupIdTbNameMap =
+      taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
+
   pOperator->name = "StreamSessionWindowAggOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
   pOperator->blocking = true;
@@ -4351,7 +4354,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
       printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
       return pInfo->pDelRes;
     }
-    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
+    doBuildStreamResBlock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
     if (pBInfo->pRes->info.rows == 0 || !hasRemainResults(&pInfo->groupResInfo)) {
       doSetOperatorCompleted(pOperator);
     }
@@ -4370,6 +4373,12 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "final session recv" : "single session recv");
 
+    if (pBlock->info.parTbName[0]) {
+      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
+                  TSDB_TABLE_NAME_LEN);
+      /*printf("\n\n put tbname %s\n\n", pBlock->info.parTbName);*/
+    }
+
     if (pBlock->info.type == STREAM_CLEAR) {
       SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
       doClearSessionWindows(&pInfo->streamAggSup, &pOperator->exprSupp, pBlock, START_TS_COLUMN_INDEX,
@@ -4451,7 +4460,7 @@ static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
     printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
     return pInfo->pDelRes;
   }
-  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
+  doBuildStreamResBlock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
   printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
   return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
 }
@@ -4482,7 +4491,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
   }
 
   {
-    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
+    doBuildStreamResBlock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
     if (pBInfo->pRes->info.rows > 0) {
       printDataBlock(pBInfo->pRes, "semi session");
       return pBInfo->pRes;
@@ -4523,6 +4532,12 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
     }
     printDataBlock(pBlock, "semi session recv");
 
+    if (pBlock->info.parTbName[0]) {
+      taosHashPut(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t), &pBlock->info.parTbName,
+                  TSDB_TABLE_NAME_LEN);
+      /*printf("\n\n put tbname %s\n\n", pBlock->info.parTbName);*/
+    }
+
     if (pBlock->info.type == STREAM_CLEAR) {
       SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
       doClearSessionWindows(&pInfo->streamAggSup, pSup, pBlock, START_TS_COLUMN_INDEX, pSup->numOfExprs, 0, pWins);
@@ -4566,7 +4581,7 @@ static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
   initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
   blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
 
-  doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
+  doBuildStreamResBlock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
   if (pBInfo->pRes->info.rows > 0) {
     printDataBlock(pBInfo->pRes, "semi session");
     return pBInfo->pRes;
@@ -4610,6 +4625,10 @@ SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream
         createOperatorFpSet(operatorDummyOpenFn, doStreamSessionSemiAgg, NULL, NULL,
                             destroyStreamSessionAggOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
   }
+
+  pInfo->pGroupIdTbNameMap =
+      taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
+
   pOperator->operatorType = pPhyNode->type;
   if (numOfChild > 0) {
     pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
@@ -4910,7 +4929,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
       printDataBlock(pInfo->pDelRes, "single state");
       return pInfo->pDelRes;
     }
-    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
+    doBuildStreamResBlock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
     if (pBInfo->pRes->info.rows == 0 || !hasRemainResults(&pInfo->groupResInfo)) {
       doSetOperatorCompleted(pOperator);
     }
@@ -4972,7 +4991,7 @@ static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
     printDataBlock(pInfo->pDelRes, "single state");
     return pInfo->pDelRes;
   }
-  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
+  doBuildStreamResBlock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
   printDataBlock(pBInfo->pRes, "single state");
   return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
 }
@@ -5045,6 +5064,9 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
   pInfo->pChildren = NULL;
   pInfo->ignoreExpiredData = pStateNode->window.igExpired;
 
+  pInfo->pGroupIdTbNameMap =
+      taosHashInit(1024, taosGetDefaultHashFunction(TSDB_DATA_TYPE_UBIGINT), false, HASH_NO_LOCK);
+
   pOperator->name = "StreamStateAggOperator";
   pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
   pOperator->blocking = true;

source/libs/parser/src/parInsert.c

@@ -266,7 +266,7 @@ static int32_t getTableVgroup(SInsertParseContext* pCxt, int32_t tbNo, SName* pT
   return catalogGetTableHashVgroup(pBasicCtx->pCatalog, &conn, pTbName, pVg);
 }
 
-static int32_t getTableMetaImpl(SInsertParseContext* pCxt, int32_t tbNo, SName* name, char* dbFname, bool isStb) {
+static int32_t getTableMetaImpl(SInsertParseContext* pCxt, int32_t tbNo, SName* name, bool isStb) {
   CHECK_CODE(getTableSchema(pCxt, tbNo, name, isStb, &pCxt->pTableMeta));
   if (!isStb) {
     SVgroupInfo vg;
@@ -276,12 +276,12 @@ static int32_t getTableMetaImpl(SInsertParseContext* pCxt, int32_t tbNo, SName*
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t getTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name, char* dbFname) {
+static int32_t getTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name) {
-  return getTableMetaImpl(pCxt, tbNo, name, dbFname, false);
+  return getTableMetaImpl(pCxt, tbNo, name, false);
 }
 
-static int32_t getSTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name, char* dbFname) {
+static int32_t getSTableMeta(SInsertParseContext* pCxt, int32_t tbNo, SName* name) {
-  return getTableMetaImpl(pCxt, tbNo, name, dbFname, true);
+  return getTableMetaImpl(pCxt, tbNo, name, true);
 }
 
 static int32_t getDBCfg(SInsertParseContext* pCxt, const char* pDbFName, SDbCfgInfo* pInfo) {
@@ -1178,7 +1178,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, int32_t tbNo, SName*
   tNameGetFullDbName(&sname, dbFName);
   strcpy(pCxt->sTableName, sname.tname);
 
-  CHECK_CODE(getSTableMeta(pCxt, tbNo, &sname, dbFName));
+  CHECK_CODE(getSTableMeta(pCxt, tbNo, &sname));
   if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) {
     return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed");
   }
@@ -1385,6 +1385,10 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, TdFilePtr fp, STableDataB
       (*numOfRows)++;
     }
     pCxt->pSql = pRawSql;
+
+    if (pDataBlock->nAllocSize > tsMaxMemUsedByInsert * 1024 * 1024) {
+      break;
+    }
   }
 
   if (0 == (*numOfRows) && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
@@ -1393,23 +1397,13 @@ static int32_t parseCsvFile(SInsertParseContext* pCxt, TdFilePtr fp, STableDataB
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t parseDataFromFile(SInsertParseContext* pCxt, SToken filePath, STableDataBlocks* dataBuf) {
+static int32_t parseDataFromFileAgain(SInsertParseContext* pCxt, int16_t tableNo, const SName* pTableName,
-  char filePathStr[TSDB_FILENAME_LEN] = {0};
+                                      STableDataBlocks* dataBuf) {
-  if (TK_NK_STRING == filePath.type) {
-    trimString(filePath.z, filePath.n, filePathStr, sizeof(filePathStr));
-  } else {
-    strncpy(filePathStr, filePath.z, filePath.n);
-  }
-  TdFilePtr fp = taosOpenFile(filePathStr, TD_FILE_READ | TD_FILE_STREAM);
-  if (NULL == fp) {
-    return TAOS_SYSTEM_ERROR(errno);
-  }
-
   int32_t maxNumOfRows;
   CHECK_CODE(allocateMemIfNeed(dataBuf, getExtendedRowSize(dataBuf), &maxNumOfRows));
 
   int32_t numOfRows = 0;
-  CHECK_CODE(parseCsvFile(pCxt, fp, dataBuf, maxNumOfRows, &numOfRows));
+  CHECK_CODE(parseCsvFile(pCxt, pCxt->pComCxt->csvCxt.fp, dataBuf, maxNumOfRows, &numOfRows));
 
   SSubmitBlk* pBlocks = (SSubmitBlk*)(dataBuf->pData);
   if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, dataBuf, numOfRows)) {
@@ -1417,12 +1411,38 @@ static int32_t parseDataFromFile(SInsertParseContext* pCxt, SToken filePath, STa
                                     "too many rows in sql, total number of rows should be less than INT32_MAX");
   }
 
+  if (!taosEOFFile(pCxt->pComCxt->csvCxt.fp)) {
+    pCxt->pComCxt->needMultiParse = true;
+    pCxt->pComCxt->csvCxt.tableNo = tableNo;
+    memcpy(&pCxt->pComCxt->csvCxt.tableName, pTableName, sizeof(SName));
+    pCxt->pComCxt->csvCxt.pLastSqlPos = pCxt->pSql;
+  }
+
   dataBuf->numOfTables = 1;
   pCxt->totalNum += numOfRows;
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t parseDataFromFile(SInsertParseContext* pCxt, int16_t tableNo, const SName* pTableName, SToken filePath,
+                                 STableDataBlocks* dataBuf) {
+  char filePathStr[TSDB_FILENAME_LEN] = {0};
+  if (TK_NK_STRING == filePath.type) {
+    trimString(filePath.z, filePath.n, filePathStr, sizeof(filePathStr));
+  } else {
+    strncpy(filePathStr, filePath.z, filePath.n);
+  }
+  pCxt->pComCxt->csvCxt.fp = taosOpenFile(filePathStr, TD_FILE_READ | TD_FILE_STREAM);
+  if (NULL == pCxt->pComCxt->csvCxt.fp) {
+    return TAOS_SYSTEM_ERROR(errno);
+  }
+
+  return parseDataFromFileAgain(pCxt, tableNo, pTableName, dataBuf);
+}
+
 static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) {
+  if (!pCxt->pComCxt->needMultiParse) {
+    taosCloseFile(&pCxt->pComCxt->csvCxt.fp);
+  }
   taosMemoryFreeClear(pCxt->pTableMeta);
   destroyBoundColumnInfo(&pCxt->tags);
   tdDestroySVCreateTbReq(&pCxt->createTblReq);
@@ -1481,7 +1501,8 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
         return buildSyntaxErrMsg(&pCxt->msg, "invalid charactor in SQL", sToken.z);
       }
 
-      if (0 == pCxt->totalNum && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
+      if (0 == pCxt->totalNum && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) &&
+          !pCxt->pComCxt->needMultiParse) {
         return buildInvalidOperationMsg(&pCxt->msg, "no data in sql");
       }
       break;
@@ -1536,7 +1557,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
       NEXT_TOKEN(pCxt->pSql, sToken);
       autoCreateTbl = true;
     } else if (!existedUsing) {
-      CHECK_CODE(getTableMeta(pCxt, tbNum, &name, dbFName));
+      CHECK_CODE(getTableMeta(pCxt, tbNum, &name));
       if (TSDB_SUPER_TABLE == pCxt->pTableMeta->tableType) {
         return buildInvalidOperationMsg(&pCxt->msg, "insert data into super table is not supported");
       }
@@ -1577,17 +1598,22 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
       if (0 == sToken.n || (TK_NK_STRING != sToken.type && TK_NK_ID != sToken.type)) {
         return buildSyntaxErrMsg(&pCxt->msg, "file path is required following keyword FILE", sToken.z);
       }
-      CHECK_CODE(parseDataFromFile(pCxt, sToken, dataBuf));
+      CHECK_CODE(parseDataFromFile(pCxt, tbNum, &name, sToken, dataBuf));
       pCxt->pOutput->insertType = TSDB_QUERY_TYPE_FILE_INSERT;
 
       tbNum++;
+      if (!pCxt->pComCxt->needMultiParse) {
         continue;
+      } else {
+        parserInfo("0x%" PRIx64 " insert from csv. File is too large, do it in batches.", pCxt->pComCxt->requestId);
+        break;
+      }
     }
 
     return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is expected", sToken.z);
   }
 
-  qDebug("0x%" PRIx64 " insert input rows: %d", pCxt->pComCxt->requestId, pCxt->totalNum);
+  parserInfo("0x%" PRIx64 " insert input rows: %d", pCxt->pComCxt->requestId, pCxt->totalNum);
 
   if (TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) {
     SParsedDataColInfo* tags = taosMemoryMalloc(sizeof(pCxt->tags));
@@ -1612,6 +1638,26 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
   return buildOutput(pCxt);
 }
 
+static int32_t parseInsertBodyAgain(SInsertParseContext* pCxt) {
+  STableDataBlocks* dataBuf = NULL;
+  CHECK_CODE(getTableMeta(pCxt, pCxt->pComCxt->csvCxt.tableNo, &pCxt->pComCxt->csvCxt.tableName));
+  CHECK_CODE(getDataBlockFromList(pCxt->pTableBlockHashObj, &pCxt->pTableMeta->uid, sizeof(pCxt->pTableMeta->uid),
+                                  TSDB_DEFAULT_PAYLOAD_SIZE, sizeof(SSubmitBlk), getTableInfo(pCxt->pTableMeta).rowSize,
+                                  pCxt->pTableMeta, &dataBuf, NULL, &pCxt->createTblReq));
+  CHECK_CODE(parseDataFromFileAgain(pCxt, pCxt->pComCxt->csvCxt.tableNo, &pCxt->pComCxt->csvCxt.tableName, dataBuf));
+  if (taosEOFFile(pCxt->pComCxt->csvCxt.fp)) {
+    CHECK_CODE(parseInsertBody(pCxt));
+    pCxt->pComCxt->needMultiParse = false;
+    return TSDB_CODE_SUCCESS;
+  }
+  parserInfo("0x%" PRIx64 " insert again input rows: %d", pCxt->pComCxt->requestId, pCxt->totalNum);
+  // merge according to vgId
+  if (taosHashGetSize(pCxt->pTableBlockHashObj) > 0) {
+    CHECK_CODE(mergeTableDataBlocks(pCxt->pTableBlockHashObj, pCxt->pOutput->payloadType, &pCxt->pVgDataBlocks));
+  }
+  return buildOutput(pCxt);
+}
+
 // INSERT INTO
 //   tb_name
 //       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
@@ -1621,7 +1667,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
 int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache) {
   SInsertParseContext context = {
       .pComCxt = pContext,
-      .pSql = (char*)pContext->pSql,
+      .pSql = pContext->needMultiParse ? (char*)pContext->csvCxt.pLastSqlPos : (char*)pContext->pSql,
       .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
       .pTableMeta = NULL,
       .createTblReq = {0},
@@ -1691,10 +1737,16 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache
 
   context.pOutput->payloadType = PAYLOAD_TYPE_KV;
 
-  int32_t code = skipInsertInto(&context.pSql, &context.msg);
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (!context.pComCxt->needMultiParse) {
+    code = skipInsertInto(&context.pSql, &context.msg);
     if (TSDB_CODE_SUCCESS == code) {
       code = parseInsertBody(&context);
     }
+  } else {
+    code = parseInsertBodyAgain(&context);
+  }
+
   if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) {
     SName* pTable = taosHashIterate(context.pTableNameHashObj, NULL);
     while (NULL != pTable) {

source/libs/parser/src/parser.c

@@ -214,6 +214,16 @@ int32_t qAnalyseSqlSemantic(SParseContext* pCxt, const struct SCatalogReq* pCata
   return code;
 }
 
+void qDestroyParseContext(SParseContext* pCxt) {
+  if (NULL == pCxt) {
+    return;
+  }
+
+  taosArrayDestroy(pCxt->pTableMetaPos);
+  taosArrayDestroy(pCxt->pTableVgroupPos);
+  taosMemoryFree(pCxt);
+}
+
 void qDestroyQuery(SQuery* pQueryNode) { nodesDestroyNode((SNode*)pQueryNode); }
 
 int32_t qExtractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema) {

source/libs/parser/test/mockCatalog.cpp

@@ -130,7 +130,7 @@ void generatePerformanceSchema(MockCatalogService* mcs) {
  *          c5         |       column       |       DOUBLE       |    8     |
  */
 void generateTestTables(MockCatalogService* mcs, const std::string& db) {
-  ITableBuilder& builder = mcs->createTableBuilder(db, "t1", TSDB_NORMAL_TABLE, 6)
+  mcs->createTableBuilder(db, "t1", TSDB_NORMAL_TABLE, 6)
       .setPrecision(TSDB_TIME_PRECISION_MILLI)
       .setVgid(1)
       .addColumn("ts", TSDB_DATA_TYPE_TIMESTAMP)
@@ -138,8 +138,8 @@ void generateTestTables(MockCatalogService* mcs, const std::string& db) {
       .addColumn("c2", TSDB_DATA_TYPE_BINARY, 20)
       .addColumn("c3", TSDB_DATA_TYPE_BIGINT)
       .addColumn("c4", TSDB_DATA_TYPE_DOUBLE)
-                               .addColumn("c5", TSDB_DATA_TYPE_DOUBLE);
+      .addColumn("c5", TSDB_DATA_TYPE_DOUBLE)
-  builder.done();
+      .done();
 }
 
 /*

source/libs/parser/test/parTestUtil.cpp

@@ -343,7 +343,6 @@ class ParserTestBaseImpl {
 
       unique_ptr<SQuery*, void (*)(SQuery**)> query((SQuery**)taosMemoryCalloc(1, sizeof(SQuery*)), destroyQuery);
       doParseSql(&cxt, query.get());
-      SQuery* pQuery = *(query.get());
 
       if (g_dump) {
         dump();

source/libs/scalar/src/scalar.c

@@ -363,7 +363,11 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
       }
 
       SSDataBlock *block = *(SSDataBlock **)taosArrayGet(ctx->pBlockList, index);
-      if (NULL == block || ref->slotId >= taosArrayGetSize(block->pDataBlock)) {
+      if (NULL == block) {
+        SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
+      }
+
+      if (ref->slotId >= taosArrayGetSize(block->pDataBlock)) {
         sclError("column slotId is too big, slodId:%d, dataBlockSize:%d", ref->slotId, (int32_t)taosArrayGetSize(block->pDataBlock));
         SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
       }

source/libs/scalar/src/sclfunc.c

@@ -2002,7 +2002,7 @@ int32_t avgScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *
     }
   }
 
-  if (hasNull) {
+  if (hasNull || (count == 0)) {
     colDataAppendNULL(pOutputData, 0);
   } else {
     if (IS_SIGNED_NUMERIC_TYPE(type)) {

source/libs/tdb/src/db/tdbBtree.c

@@ -1485,7 +1485,7 @@ static int tdbBtreeCellSize(const SPage *pPage, SCell *pCell, int dropOfp, TXN *
     if (dropOfp) {
       int    ret = 0;
       SPgno  pgno = *(SPgno *)(pCell + nLocal - sizeof(SPgno));
-      int    nLeft = nPayload - nLocal + sizeof(SPgno);
+      int    nLeft = nPayload - nLocal + sizeof(SPgno) + nHeader;
       SPage *ofp;
       int    bytes;
 

source/libs/tdb/test/tdbExOVFLTest.cpp

@@ -238,8 +238,8 @@ TEST(TdbOVFLPagesTest, TbGetTest) {
   }
 }
 
-TEST(TdbOVFLPagesTest, DISABLED_TbDeleteTest) {
+// TEST(TdbOVFLPagesTest, DISABLED_TbDeleteTest) {
-  // TEST(TdbOVFLPagesTest, TbDeleteTest) {
+TEST(TdbOVFLPagesTest, TbDeleteTest) {
   int ret = 0;
 
   taosRemoveDir("tdb");
@@ -267,7 +267,8 @@ TEST(TdbOVFLPagesTest, DISABLED_TbDeleteTest) {
   tdbBegin(pEnv, &txn);
 
   // generate value payload
-  char val[((4083 - 4 - 3 - 2) + 1) * 100];  // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
+  // char val[((4083 - 4 - 3 - 2) + 1) * 100];  // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
+  char val[((4083 - 4 - 3 - 2) + 1) * 2];  // pSize(4096) - amSize(1) - pageHdr(8) - footerSize(4)
   int  valLen = sizeof(val) / sizeof(val[0]);
   generateBigVal(val, valLen);
 
@@ -340,8 +341,8 @@ tdbBegin(pEnv, &txn);
   tdbTxnClose(&txn);
 }
 
-TEST(tdb_test, DISABLED_simple_insert1) {
+// TEST(tdb_test, DISABLED_simple_insert1) {
-  // TEST(tdb_test, simple_insert1) {
+TEST(tdb_test, simple_insert1) {
   int           ret;
   TDB          *pEnv;
   TTB          *pDb;

source/libs/wal/src/walMeta.c

@@ -150,7 +150,6 @@ int walCheckAndRepairMeta(SWal* pWal) {
   const char* idxPattern = "^[0-9]+.idx$";
   regex_t     logRegPattern;
   regex_t     idxRegPattern;
-  SArray*     actualLog = taosArrayInit(8, sizeof(SWalFileInfo));
 
   regcomp(&logRegPattern, logPattern, REG_EXTENDED);
   regcomp(&idxRegPattern, idxPattern, REG_EXTENDED);
@@ -163,6 +162,8 @@ int walCheckAndRepairMeta(SWal* pWal) {
     return -1;
   }
 
+  SArray* actualLog = taosArrayInit(8, sizeof(SWalFileInfo));
+
   // scan log files and build new meta
   TdDirEntryPtr pDirEntry;
   while ((pDirEntry = taosReadDir(pDir)) != NULL) {

source/os/CMakeLists.txt

@@ -52,3 +52,7 @@ else()
         os PUBLIC dl m rt
     )
 endif()
+
+IF (JEMALLOC_ENABLED)
+    target_link_libraries(os PUBLIC -ljemalloc)
+ENDIF ()

source/os/src/osSocket.c

@@ -267,21 +267,29 @@ int32_t taosSetSockOpt(TdSocketPtr pSocket, int32_t level, int32_t optname, void
     return -1;
   }
 #ifdef WINDOWS
+#ifdef TCP_KEEPCNT
   if (level == SOL_SOCKET && optname == TCP_KEEPCNT) {
     return 0;
   }
+#endif
 
+#ifdef TCP_KEEPIDLE
   if (level == SOL_TCP && optname == TCP_KEEPIDLE) {
     return 0;
   }
+#endif
 
+#ifdef TCP_KEEPINTVL
   if (level == SOL_TCP && optname == TCP_KEEPINTVL) {
     return 0;
   }
+#endif
 
+#ifdef TCP_KEEPCNT
   if (level == SOL_TCP && optname == TCP_KEEPCNT) {
     return 0;
   }
+#endif
 
   return setsockopt(pSocket->fd, level, optname, optval, optlen);
 #else
@@ -601,26 +609,32 @@ int32_t taosKeepTcpAlive(TdSocketPtr pSocket) {
 
 #ifndef __APPLE__
   // all fails on macosx
+#ifdef TCP_KEEPCNT
   int32_t probes = 3;
   if (taosSetSockOpt(pSocket, SOL_TCP, TCP_KEEPCNT, (void *)&probes, sizeof(probes)) < 0) {
     // printf("fd:%d setsockopt SO_KEEPCNT failed: %d (%s)", sockFd, errno, strerror(errno));
     taosCloseSocket(&pSocket);
     return -1;
   }
+#endif
 
+#ifdef TCP_KEEPIDLE
   int32_t alivetime = 10;
   if (taosSetSockOpt(pSocket, SOL_TCP, TCP_KEEPIDLE, (void *)&alivetime, sizeof(alivetime)) < 0) {
     // printf("fd:%d setsockopt SO_KEEPIDLE failed: %d (%s)", sockFd, errno, strerror(errno));
     taosCloseSocket(&pSocket);
     return -1;
   }
+#endif
 
+#ifdef TCP_KEEPINTVL
   int32_t interval = 3;
   if (taosSetSockOpt(pSocket, SOL_TCP, TCP_KEEPINTVL, (void *)&interval, sizeof(interval)) < 0) {
     // printf("fd:%d setsockopt SO_KEEPINTVL failed: %d (%s)", sockFd, errno, strerror(errno));
     taosCloseSocket(&pSocket);
     return -1;
   }
+#endif
 #endif  // __APPLE__
 
   int32_t nodelay = 1;

tests/system-test/6-cluster/5dnode3mnodeSep1VnodeStopMnodeCreateStb.py

@@ -132,7 +132,7 @@ class TDTestCase:
         tdSql.error("create mnode on dnode 2")
         tdSql.query("select * from information_schema.ins_dnodes;")
         print(tdSql.queryResult)
-        clusterComCheck.checkDnodes(dnodeNumbers)
+        clusterComCheck.checkDnodes(dnodeNumbers, 60)
 
         # create database and stable
         clusterComCreate.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], paraDict["vgroups"],paraDict['replica'])

tests/system-test/6-cluster/clusterCommonCheck.py

@@ -37,10 +37,10 @@ class ClusterComCheck:
         tdSql.init(conn.cursor())
         # tdSql.init(conn.cursor(), logSql)  # output sql.txt file
 
-    def checkDnodes(self,dnodeNumbers):
+    def checkDnodes(self,dnodeNumbers, timeout=30):
         count=0
         # print(tdSql)
-        while count < 30:
+        while count < timeout:
             tdSql.query("select * from information_schema.ins_dnodes")
             # tdLog.debug(tdSql.queryResult)
             status=0
@@ -50,14 +50,14 @@ class ClusterComCheck:
             tdLog.info(status)
 
             if status == dnodeNumbers:
-                tdLog.success("it find cluster with %d dnodes and check that all cluster dnodes are ready within 30s! " %dnodeNumbers)
+                tdLog.success("it find cluster with %d dnodes and check that all cluster dnodes are ready within %ds! " % (dnodeNumbers, count))
                 return True
             count+=1
             time.sleep(1)
         else:
             tdSql.query("select * from information_schema.ins_dnodes")
             tdLog.debug(tdSql.queryResult)
-            tdLog.exit("it find cluster with %d dnodes but  check that there dnodes are not ready within 30s ! "%dnodeNumbers)
+            tdLog.exit("it find cluster with %d dnodes but  check that there dnodes are not ready within %ds ! "% (dnodeNumbers, timeout))
 
     def checkDbRows(self,dbNumbers):
         dbNumbers=int(dbNumbers)

tools/shell/CMakeLists.txt

@@ -28,7 +28,4 @@ target_include_directories(
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 
-IF (TD_LINUX_64 AND JEMALLOC_ENABLED)
-    add_dependencies(taosd jemalloc)
-ENDIF ()
 SET_TARGET_PROPERTIES(shell PROPERTIES OUTPUT_NAME taos)

tools/shell/src/shellAuto.c

@@ -1133,6 +1133,10 @@ bool firstMatchCommand(TAOS* con, SShellCmd* cmd) {
 
   // print to screen
   printScreen(con, cmd, match);
+#ifdef WINDOWS
+  printf("\r");
+  shellShowOnScreen(cmd);
+#endif
   freeCommand(input);
   taosMemoryFree(input);
   return true;
@@ -1195,6 +1199,10 @@ bool nextMatchCommand(TAOS* con, SShellCmd* cmd, SWords* firstMatch) {
 
   // print to screen
   printScreen(con, cmd, match);
+#ifdef WINDOWS
+  printf("\r");
+  shellShowOnScreen(cmd);
+#endif
 
   // free
   if (input->source) {

tools/shell/src/shellCommand.c

@@ -417,8 +417,9 @@ char taosGetConsoleChar() {
   static char mbStr[5];
   static unsigned long bufLen = 0;
   static uint16_t bufIndex = 0, mbStrIndex = 0, mbStrLen = 0;
+  CONSOLE_READCONSOLE_CONTROL inputControl={ sizeof(CONSOLE_READCONSOLE_CONTROL), 0, 1<<TAB_KEY, 0 };
   while (bufLen == 0) {
-    ReadConsoleW(console, buf, SHELL_INPUT_MAX_COMMAND_SIZE, &bufLen, NULL);
+    ReadConsoleW(console, buf, SHELL_INPUT_MAX_COMMAND_SIZE, &bufLen, &inputControl);
     if (bufLen > 0 && buf[0] == 0) bufLen = 0;
     bufIndex = 0;
   }

tools/shell/src/shellEngine.c

@@ -1112,7 +1112,9 @@ int32_t shellExecute() {
 #ifdef WEBSOCKET
   if (!shell.args.restful && !shell.args.cloud) {
 #endif
+#ifndef WINDOWS
   printfIntroduction();
+#endif  
 	shellGetGrantInfo();
 #ifdef WEBSOCKET
   }