Merge branch '3.0' into test/jcy

CMakeLists.txt

@@ -7,7 +7,7 @@ project(
 )
 
 if (NOT DEFINED TD_SOURCE_DIR)
-  set( TD_SOURCE_DIR ${CMAKE_SOURCE_DIR} )
+  set( TD_SOURCE_DIR ${PROJECT_SOURCE_DIR} )
 endif()
 
 set(TD_SUPPORT_DIR "${TD_SOURCE_DIR}/cmake")

example/src/tmq.c

@@ -60,7 +60,6 @@ int32_t init_env() {
   pRes =
       taos_query(pConn, "create stable if not exists st1 (ts timestamp, c1 int, c2 float, c3 binary(10)) tags(t1 int)");
   if (taos_errno(pRes) != 0) {
-    assert(0);
     printf("failed to create super table st1, reason:%s\n", taos_errstr(pRes));
     return -1;
   }
@@ -104,8 +103,8 @@ int32_t create_topic() {
   }
   taos_free_result(pRes);
 
-  /*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
-  pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");
+  pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");
+  /*pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from ct1");*/
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
     return -1;
@@ -163,9 +162,10 @@ tmq_t* build_consumer() {
   tmq_conf_set(conf, "group.id", "tg2");
   tmq_conf_set(conf, "td.connect.user", "root");
   tmq_conf_set(conf, "td.connect.pass", "taosdata");
-  tmq_conf_set(conf, "td.connect.db", "abc1");
+  /*tmq_conf_set(conf, "td.connect.db", "abc1");*/
   tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print);
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
+  assert(tmq);
   return tmq;
 }
 

examples/c/CMakeLists.txt

 PROJECT(TDengine)
 
 IF (TD_LINUX)
-  INCLUDE_DIRECTORIES(. ${TD_COMMUNITY_DIR}/src/inc ${TD_COMMUNITY_DIR}/src/client/inc  ${TD_COMMUNITY_DIR}/inc)
+  INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc  ${TD_SOURCE_DIR}/inc)
   AUX_SOURCE_DIRECTORY(. SRC)
   ADD_EXECUTABLE(demo apitest.c)
   TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread )
@@ -13,7 +13,7 @@ IF (TD_LINUX)
   TARGET_LINK_LIBRARIES(epoll taos_static trpc tutil pthread lua)
 ENDIF ()
 IF (TD_DARWIN)
-  INCLUDE_DIRECTORIES(. ${TD_COMMUNITY_DIR}/src/inc ${TD_COMMUNITY_DIR}/src/client/inc  ${TD_COMMUNITY_DIR}/inc)
+  INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc  ${TD_SOURCE_DIR}/inc)
   AUX_SOURCE_DIRECTORY(. SRC)
   ADD_EXECUTABLE(demo demo.c)
   TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread lua)

include/common/tglobal.h

@@ -123,9 +123,9 @@ extern SDiskCfg tsDiskCfg[];
 
 #define NEEDTO_COMPRESSS_MSG(size) (tsCompressMsgSize != -1 && (size) > tsCompressMsgSize)
 
-int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char *envFile,
-                      const char *apolloUrl, SArray *pArgs, bool tsc);
-int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloUrl, SArray *pArgs, bool tsc);
+int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, const char *envFile,
+                      char *apolloUrl, SArray *pArgs, bool tsc);
+int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc);
 void    taosCleanupCfg();
 void    taosCfgDynamicOptions(const char *option, const char *value);
 void    taosAddDataDir(int32_t index, char *v1, int32_t level, int32_t primary);

include/common/tmsg.h

@@ -229,6 +229,14 @@ typedef struct {
 typedef struct {
   int32_t totalLen;
   int32_t len;
+  // head of SSubmitBlk
+  // int64_t uid;        // table unique id
+  // int64_t suid;       // stable id
+  // int32_t sversion;   // data schema version
+  // int32_t dataLen;    // data part length, not including the SSubmitBlk head
+  // int32_t schemaLen;  // schema length, if length is 0, no schema exists
+  // int16_t numOfRows;  // total number of rows in current submit block
+  // head of SSubmitBlk
   const void* pMsg;
 } SSubmitMsgIter;
 
@@ -237,6 +245,15 @@ int32_t tGetSubmitMsgNext(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
 int32_t tInitSubmitBlkIter(SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
 STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
 
+// TODO: KEEP one suite of iterator API finally.
+// 1) use tInitSubmitMsgIterEx firstly as not decrease the merge conflicts
+// 2) replace tInitSubmitMsgIterEx with tInitSubmitMsgIter later
+// 3) finally, rename tInitSubmitMsgIterEx to tInitSubmitMsgIter
+// int32_t tInitSubmitMsgIterEx(const SSubmitReq* pMsg, SSubmitMsgIter* pIter);
+// int32_t tGetSubmitMsgNextEx(SSubmitMsgIter* pIter, SSubmitBlk** pPBlock);
+// int32_t tInitSubmitBlkIterEx(SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkIter* pIter);
+// STSRow* tGetSubmitBlkNextEx(SSubmitBlkIter* pIter);
+
 typedef struct {
   int32_t index;  // index of failed block in submit blocks
   int32_t vnode;  // vnode index of failed block
@@ -1496,8 +1513,8 @@ typedef struct {
   int32_t    qmsg1Len;
   int32_t    qmsg2Len;
   func_id_t* pFuncIds;
-  char*      qmsg1;  // not null: pAst1:qmsg1:SRetention1 => trigger aggr task1
-  char*      qmsg2;  // not null: pAst2:qmsg2:SRetention2 => trigger aggr task2
+  char*      qmsg1;  // pAst1:qmsg1:SRetention1 => trigger aggr task1
+  char*      qmsg2;  // pAst2:qmsg2:SRetention2 => trigger aggr task2
   int8_t     nFuncIds;
 } SRSmaParam;
 
@@ -2052,80 +2069,6 @@ static FORCE_INLINE void* tDecodeSMqSetCVgReq(void* buf, SMqSetCVgReq* pReq) {
   return buf;
 }
 
-typedef struct {
-  int64_t leftForVer;
-  int32_t vgId;
-  int32_t epoch;
-  int64_t consumerId;
-  char    topicName[TSDB_TOPIC_FNAME_LEN];
-} SMqCancelConnReq;
-
-static FORCE_INLINE int32_t tEncodeSMqCancelConnReq(void** buf, const SMqCancelConnReq* pReq) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
-  tlen += taosEncodeFixedI32(buf, pReq->vgId);
-  tlen += taosEncodeFixedI32(buf, pReq->epoch);
-  tlen += taosEncodeFixedI64(buf, pReq->consumerId);
-  tlen += taosEncodeString(buf, pReq->topicName);
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeSMqCancelConnReq(void* buf, SMqCancelConnReq* pReq) {
-  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
-  buf = taosDecodeFixedI32(buf, &pReq->vgId);
-  buf = taosDecodeFixedI32(buf, &pReq->epoch);
-  buf = taosDecodeFixedI64(buf, &pReq->consumerId);
-  buf = taosDecodeStringTo(buf, pReq->topicName);
-  return buf;
-}
-
-typedef struct {
-  int8_t reserved;
-} SMqCancelConnRsp;
-
-typedef struct {
-  int64_t leftForVer;
-  int32_t vgId;
-  int64_t oldConsumerId;
-  int64_t newConsumerId;
-  char*   topic;
-} SMqMVRebReq;
-
-static FORCE_INLINE int32_t tEncodeSMqMVRebReq(void** buf, const SMqMVRebReq* pReq) {
-  int32_t tlen = 0;
-  tlen += taosEncodeFixedI64(buf, pReq->leftForVer);
-  tlen += taosEncodeFixedI32(buf, pReq->vgId);
-  tlen += taosEncodeFixedI64(buf, pReq->oldConsumerId);
-  tlen += taosEncodeFixedI64(buf, pReq->newConsumerId);
-  tlen += taosEncodeString(buf, pReq->topic);
-  return tlen;
-}
-
-static FORCE_INLINE void* tDecodeSMqMVRebReq(void* buf, SMqMVRebReq* pReq) {
-  buf = taosDecodeFixedI64(buf, &pReq->leftForVer);
-  buf = taosDecodeFixedI32(buf, &pReq->vgId);
-  buf = taosDecodeFixedI64(buf, &pReq->oldConsumerId);
-  buf = taosDecodeFixedI64(buf, &pReq->newConsumerId);
-  buf = taosDecodeString(buf, &pReq->topic);
-  return buf;
-}
-
-typedef struct {
-  SMsgHead header;
-  int32_t  vgId;
-  int64_t  consumerId;
-  char     topicName[TSDB_TOPIC_FNAME_LEN];
-  char     cgroup[TSDB_CGROUP_LEN];
-} SMqSetCVgRsp;
-
-typedef struct {
-  SMsgHead header;
-  int32_t  vgId;
-  int64_t  consumerId;
-  char     topicName[TSDB_TOPIC_FNAME_LEN];
-  char     cgroup[TSDB_CGROUP_LEN];
-} SMqMVRebRsp;
-
 typedef struct {
   int32_t vgId;
   int64_t offset;
@@ -2152,6 +2095,24 @@ typedef struct {
   SSchema* pSchema;
 } SSchemaWrapper;
 
+static FORCE_INLINE SSchemaWrapper* tCloneSSchemaWrapper(const SSchemaWrapper* pSchemaWrapper) {
+  SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
+  if (pSW == NULL) return pSW;
+  pSW->nCols = pSchemaWrapper->nCols;
+  pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
+  if (pSW->pSchema == NULL) {
+    taosMemoryFree(pSW);
+    return NULL;
+  }
+  memcpy(pSW->pSchema, pSchemaWrapper->pSchema, pSW->nCols * sizeof(SSchema));
+  return pSW;
+}
+
+static FORCE_INLINE void tDeleteSSchemaWrapper(SSchemaWrapper* pSchemaWrapper) {
+  taosMemoryFree(pSchemaWrapper->pSchema);
+  taosMemoryFree(pSchemaWrapper);
+}
+
 static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
   int32_t tlen = 0;
   tlen += taosEncodeFixedI8(buf, pSchema->type);
@@ -2162,13 +2123,13 @@ static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema
   return tlen;
 }
 
-static FORCE_INLINE void* taosDecodeSSchema(void* buf, SSchema* pSchema) {
+static FORCE_INLINE void* taosDecodeSSchema(const void* buf, SSchema* pSchema) {
   buf = taosDecodeFixedI8(buf, &pSchema->type);
   buf = taosDecodeFixedI8(buf, &pSchema->flags);
   buf = taosDecodeFixedI32(buf, &pSchema->bytes);
   buf = taosDecodeFixedI16(buf, &pSchema->colId);
   buf = taosDecodeStringTo(buf, pSchema->name);
-  return buf;
+  return (void*)buf;
 }
 
 static FORCE_INLINE int32_t tEncodeSSchema(SCoder* pEncoder, const SSchema* pSchema) {
@@ -2198,7 +2159,7 @@ static FORCE_INLINE int32_t taosEncodeSSchemaWrapper(void** buf, const SSchemaWr
   return tlen;
 }
 
-static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pSW) {
+static FORCE_INLINE void* taosDecodeSSchemaWrapper(const void* buf, SSchemaWrapper* pSW) {
   buf = taosDecodeFixedU32(buf, &pSW->nCols);
   pSW->pSchema = (SSchema*)taosMemoryCalloc(pSW->nCols, sizeof(SSchema));
   if (pSW->pSchema == NULL) {
@@ -2208,7 +2169,7 @@ static FORCE_INLINE void* taosDecodeSSchemaWrapper(void* buf, SSchemaWrapper* pS
   for (int32_t i = 0; i < pSW->nCols; i++) {
     buf = taosDecodeSSchema(buf, &pSW->pSchema[i]);
   }
-  return buf;
+  return (void*)buf;
 }
 
 static FORCE_INLINE int32_t tEncodeSSchemaWrapper(SCoder* pEncoder, const SSchemaWrapper* pSW) {
@@ -2615,6 +2576,10 @@ static FORCE_INLINE int32_t tEncodeSMqDataBlkRsp(void** buf, const SMqDataBlkRsp
       void*   data = taosArrayGetP(pRsp->blockData, i);
       tlen += taosEncodeFixedI32(buf, bLen);
       tlen += taosEncodeBinary(buf, data, bLen);
+      if (pRsp->withSchema) {
+        SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(pRsp->blockSchema, i);
+        tlen += taosEncodeSSchemaWrapper(buf, pSW);
+      }
     }
   }
   return tlen;
@@ -2627,6 +2592,7 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p
   buf = taosDecodeFixedI32(buf, &pRsp->blockNum);
   pRsp->blockData = taosArrayInit(pRsp->blockNum, sizeof(void*));
   pRsp->blockDataLen = taosArrayInit(pRsp->blockNum, sizeof(void*));
+  pRsp->blockSchema = taosArrayInit(pRsp->blockNum, sizeof(void*));
   if (pRsp->blockNum != 0) {
     buf = taosDecodeFixedI8(buf, &pRsp->withTbName);
     buf = taosDecodeFixedI8(buf, &pRsp->withSchema);
@@ -2639,6 +2605,11 @@ static FORCE_INLINE void* tDecodeSMqDataBlkRsp(const void* buf, SMqDataBlkRsp* p
       buf = taosDecodeBinary(buf, &data, bLen);
       taosArrayPush(pRsp->blockDataLen, &bLen);
       taosArrayPush(pRsp->blockData, &data);
+      if (pRsp->withSchema) {
+        SSchemaWrapper* pSW = (SSchemaWrapper*)taosMemoryMalloc(sizeof(SSchemaWrapper));
+        buf = taosDecodeSSchemaWrapper(buf, pSW);
+        taosArrayPush(pRsp->blockSchema, &pSW);
+      }
     }
   }
   return (void*)buf;

include/util/tconfig.h

@@ -30,6 +30,7 @@ typedef enum {
   CFG_STYPE_CFG_FILE,
   CFG_STYPE_ENV_FILE,
   CFG_STYPE_ENV_VAR,
+  CFG_STYPE_ENV_CMD,
   CFG_STYPE_APOLLO_URL,
   CFG_STYPE_ARG_LIST,
   CFG_STYPE_TAOS_OPTIONS
@@ -82,7 +83,7 @@ typedef struct SConfig {
 } SConfig;
 
 SConfig *cfgInit();
-int32_t  cfgLoad(SConfig *pCfg, ECfgSrcType cfgType, const char *sourceStr);
+int32_t  cfgLoad(SConfig *pCfg, ECfgSrcType cfgType, const void *sourceStr);
 int32_t  cfgLoadFromArray(SConfig *pCfg, SArray *pArgs);  // SConfigPair
 void     cfgCleanup(SConfig *pCfg);
 
@@ -105,6 +106,8 @@ const char *cfgDtypeStr(ECfgDataType type);
 
 void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump);
 
+int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char* apolloUrl);
+
 #ifdef __cplusplus
 }
 #endif

include/util/tenv.h

+
+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_ENV_H_
+#define _TD_ENV_H_
+
+#include "os.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int32_t taosEnvNameToCfgName(const char *envNameStr, char *cfgNameStr, int32_t cfgNameMaxLen);
+int32_t taosEnvToCfg(const char *envStr, char *cfgStr);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_ENV_H_*/

include/util/tjson.h

@@ -49,6 +49,8 @@ int32_t tjsonAddItemToObject(SJson* pJson, const char* pName, SJson* pItem);
 int32_t tjsonAddItemToArray(SJson* pJson, SJson* pItem);
 
 SJson*  tjsonGetObjectItem(const SJson* pJson, const char* pName);
+int32_t tjsonGetObjectName(const SJson* pJson, char** pName);
+int32_t tjsonGetObjectValueString(const SJson* pJson, char** pStringValue);
 int32_t tjsonGetStringValue(const SJson* pJson, const char* pName, char* pVal);
 int32_t tjsonDupStringValue(const SJson* pJson, const char* pName, char** pVal);
 int32_t tjsonGetBigIntValue(const SJson* pJson, const char* pName, int64_t* pVal);
@@ -81,6 +83,7 @@ char* tjsonToUnformattedString(const SJson* pJson);
 
 SJson* tjsonParse(const char* pStr);
 bool   tjsonValidateJson(const char* pJson);
+const char* tjsonGetError();
 
 #ifdef __cplusplus
 }

source/client/inc/clientInt.h

@@ -231,6 +231,10 @@ static FORCE_INLINE SReqResultInfo* tmqGetNextResInfo(TAOS_RES* res, bool conver
   msg->resIter++;
   if (msg->resIter < msg->rsp.blockNum) {
     SRetrieveTableRsp* pRetrieve = (SRetrieveTableRsp*)taosArrayGetP(msg->rsp.blockData, msg->resIter);
+    if (msg->rsp.withSchema) {
+      SSchemaWrapper* pSW = (SSchemaWrapper*)taosArrayGetP(msg->rsp.blockSchema, msg->resIter);
+      setResSchemaInfo(&msg->resInfo, pSW->pSchema, pSW->nCols);
+    }
     setQueryResultFromRsp(&msg->resInfo, pRetrieve, convertUcs4);
     return &msg->resInfo;
   }

source/client/src/clientEnv.c

@@ -254,12 +254,12 @@ void taos_init_imp(void) {
 
   deltaToUtcInitOnce();
 
-  if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, 1) != 0) {
+  if (taosCreateLog("taoslog", 10, configDir, NULL, NULL, NULL, NULL, 1) != 0) {
     tscInitRes = -1;
     return;
   }
 
-  if (taosInitCfg(configDir, NULL, NULL, NULL, 1) != 0) {
+  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1) != 0) {
     tscInitRes = -1;
     return;
   }

source/client/src/clientHb.c

@@ -14,9 +14,9 @@
  */
 
 #include "catalog.h"
-#include "scheduler.h"
 #include "clientInt.h"
 #include "clientLog.h"
+#include "scheduler.h"
 #include "trpc.h"
 
 static SClientHbMgr clientHbMgr = {0};
@@ -110,7 +110,8 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
 static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
   SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey));
   if (NULL == info) {
-    tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid, pRsp->connKey.connType);
+    tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid,
+            pRsp->connKey.connType);
     return TSDB_CODE_SUCCESS;
   }
 
@@ -471,10 +472,10 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
     pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
   }
 
-//  if (code) {
-//    taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
-//    taosMemoryFreeClear(pBatchReq);
-//  }
+  //  if (code) {
+  //    taosArrayDestroyEx(pBatchReq->reqs, hbFreeReq);
+  //    taosMemoryFreeClear(pBatchReq);
+  //  }
 
   return pBatchReq;
 }
@@ -641,7 +642,6 @@ void appHbMgrCleanup(void) {
     taosHashCleanup(pTarget->activeInfo);
     pTarget->activeInfo = NULL;
 
-
     pIter = taosHashIterate(pTarget->connInfo, NULL);
     while (pIter != NULL) {
       SHbConnInfo *info = pIter;
@@ -668,13 +668,13 @@ int hbMgrInit() {
   hbMgrInitHandle();
 
   // init backgroud thread
-  //hbCreateThread();
+  /*hbCreateThread();*/
 
   return 0;
 }
 
 void hbMgrCleanUp() {
-  //hbStopThread();
+  // hbStopThread();
 
   // destroy all appHbMgr
   int8_t old = atomic_val_compare_exchange_8(&clientHbMgr.inited, 1, 0);

source/client/src/clientImpl.c

@@ -226,16 +226,14 @@ int32_t execDdlQuery(SRequestObj* pRequest, SQuery* pQuery) {
 
 int32_t getPlan(SRequestObj* pRequest, SQuery* pQuery, SQueryPlan** pPlan, SArray* pNodeList) {
   pRequest->type = pQuery->msgType;
-  SPlanContext cxt = {
-    .queryId = pRequest->requestId,
+  SPlanContext cxt = {.queryId = pRequest->requestId,
                       .acctId = pRequest->pTscObj->acctId,
                       .mgmtEpSet = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp),
                       .pAstRoot = pQuery->pRoot,
                       .showRewrite = pQuery->showRewrite,
                       .pTransporter = pRequest->pTscObj->pAppInfo->pTransporter,
                       .pMsg = pRequest->msgBuf,
-    .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE
-  };
+                      .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE};
   int32_t      code = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
   if (TSDB_CODE_SUCCESS == code) {
     code = qCreateQueryPlan(&cxt, pPlan, pNodeList);
@@ -247,6 +245,7 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t
   ASSERT(pSchema != NULL && numOfCols > 0);
 
   pResInfo->numOfCols = numOfCols;
+  // TODO handle memory leak
   pResInfo->fields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
   pResInfo->userFields = taosMemoryCalloc(numOfCols, sizeof(TAOS_FIELD));
 
@@ -840,12 +839,12 @@ int32_t setResultDataPtr(SReqResultInfo* pResultInfo, TAOS_FIELD* pFields, int32
     return code;
   }
 
-  char* p = (char*) pResultInfo->pData;
+  char* p = (char*)pResultInfo->pData;
 
-  int32_t dataLen = *(int32_t*) p;
+  int32_t dataLen = *(int32_t*)p;
   p += sizeof(int32_t);
 
-  uint64_t groupId = *(uint64_t*) p;
+  uint64_t groupId = *(uint64_t*)p;
   p += sizeof(uint64_t);
 
   int32_t* colLength = (int32_t*)p;

source/client/src/tmq.c

@@ -376,7 +376,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
 
   ASSERT(user);
   ASSERT(pass);
-  ASSERT(conf->db);
+  /*ASSERT(conf->db);*/
   ASSERT(conf->groupId[0]);
 
   pTmq->pTscObj = taos_connect_internal(conf->ip, user, pass, NULL, conf->db, conf->port, CONN_TYPE__TMQ);
@@ -1118,7 +1118,9 @@ SMqRspObj* tmqBuildRspFromWrapper(SMqPollRspWrapper* pWrapper) {
 
   pRspObj->resInfo.totalRows = 0;
   pRspObj->resInfo.precision = TSDB_TIME_PRECISION_MILLI;
+  if (!pWrapper->msg.withSchema) {
     setResSchemaInfo(&pRspObj->resInfo, pWrapper->topicHandle->schema.pSchema, pWrapper->topicHandle->schema.nCols);
+  }
 
   taosFreeQitem(pWrapper);
   return pRspObj;

source/common/src/tglobal.c

@@ -220,7 +220,7 @@ struct SConfig *taosGetCfg() {
   return tsCfg;
 }
 
-static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *envFile, const char *apolloUrl) {
+static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *inputCfgDir, const char *envFile, char *apolloUrl) {
   char cfgDir[PATH_MAX] = {0};
   char cfgFile[PATH_MAX + 100] = {0};
 
@@ -231,6 +231,8 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *e
     tstrncpy(cfgFile, cfgDir, sizeof(cfgDir));
   }
 
+  if (apolloUrl == NULL || apolloUrl[0] == '\0') cfgGetApollUrl(envCmd, envFile, apolloUrl);
+
   if (cfgLoad(pCfg, CFG_STYPE_APOLLO_URL, apolloUrl) != 0) {
     uError("failed to load from apollo url:%s since %s", apolloUrl, terrstr());
     return -1;
@@ -251,6 +253,11 @@ static int32_t taosLoadCfg(SConfig *pCfg, const char *inputCfgDir, const char *e
     return -1;
   }
 
+  if (cfgLoad(pCfg, CFG_STYPE_ENV_CMD, envCmd) != 0) {
+    uError("failed to load from cmd env variables since %s", terrstr());
+    return -1;
+  }
+
   return 0;
 }
 
@@ -325,8 +332,8 @@ static int32_t taosAddSystemCfg(SConfig *pCfg) {
   if (cfgAddFloat(pCfg, "numOfCores", tsNumOfCores, 0, 100000, 1) != 0) return -1;
   if (cfgAddInt64(pCfg, "openMax", tsOpenMax, 0, INT64_MAX, 1) != 0) return -1;
   if (cfgAddInt64(pCfg, "streamMax", tsStreamMax, 0, INT64_MAX, 1) != 0) return -1;
-  if (cfgAddInt32(pCfg, "pageSize(KB)", tsPageSizeKB, 0, INT64_MAX, 1) != 0) return -1;
-  if (cfgAddInt64(pCfg, "totalMemory(KB)", tsTotalMemoryKB, 0, INT64_MAX, 1) != 0) return -1;
+  if (cfgAddInt32(pCfg, "pageSizeKB", tsPageSizeKB, 0, INT64_MAX, 1) != 0) return -1;
+  if (cfgAddInt64(pCfg, "totalMemoryKB", tsTotalMemoryKB, 0, INT64_MAX, 1) != 0) return -1;
   if (cfgAddString(pCfg, "os sysname", info.sysname, 1) != 0) return -1;
   if (cfgAddString(pCfg, "os nodename", info.nodename, 1) != 0) return -1;
   if (cfgAddString(pCfg, "os release", info.release, 1) != 0) return -1;
@@ -572,8 +579,8 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   return 0;
 }
 
-int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char *envFile,
-                      const char *apolloUrl, SArray *pArgs, bool tsc) {
+int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDir, const char **envCmd, const char *envFile,
+                      char *apolloUrl, SArray *pArgs, bool tsc) {
   osDefaultInit();
 
   SConfig *pCfg = cfgInit();
@@ -588,7 +595,7 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
     if (taosAddServerLogCfg(pCfg) != 0) return -1;
   }
 
-  if (taosLoadCfg(pCfg, cfgDir, envFile, apolloUrl) != 0) {
+  if (taosLoadCfg(pCfg, envCmd, cfgDir, envFile, apolloUrl) != 0) {
     uError("failed to load cfg since %s", terrstr());
     cfgCleanup(pCfg);
     return -1;
@@ -625,7 +632,7 @@ int32_t taosCreateLog(const char *logname, int32_t logFileNum, const char *cfgDi
   return 0;
 }
 
-int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloUrl, SArray *pArgs, bool tsc) {
+int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile, char *apolloUrl, SArray *pArgs, bool tsc) {
   if (tsCfg != NULL) return 0;
   tsCfg = cfgInit();
 
@@ -640,7 +647,7 @@ int32_t taosInitCfg(const char *cfgDir, const char *envFile, const char *apolloU
   }
   taosAddSystemCfg(tsCfg);
 
-  if (taosLoadCfg(tsCfg, cfgDir, envFile, apolloUrl) != 0) {
+  if (taosLoadCfg(tsCfg, envCmd, cfgDir, envFile, apolloUrl) != 0) {
     uError("failed to load cfg since %s", terrstr());
     cfgCleanup(tsCfg);
     tsCfg = NULL;

source/common/src/tmsg.c

@@ -93,7 +93,87 @@ STSRow *tGetSubmitBlkNext(SSubmitBlkIter *pIter) {
     return row;
   }
 }
+#if 0
+// TODO: KEEP one suite of iterator API finally.
+// 1) use tInitSubmitMsgIterEx firstly as not decrease the merge conflicts
+// 2) replace tInitSubmitMsgIterEx with tInitSubmitMsgIter later
+// 3) finally, rename tInitSubmitMsgIterEx to tInitSubmitMsgIter
 
+int32_t tInitSubmitMsgIterEx(const SSubmitReq *pMsg, SSubmitMsgIter *pIter) {
+  if (pMsg == NULL) {
+    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
+    return -1;
+  }
+
+  pIter->totalLen = htonl(pMsg->length);
+  ASSERT(pIter->totalLen > 0);
+  pIter->len = 0;
+  pIter->pMsg = pMsg;
+  if (pIter->totalLen <= sizeof(SSubmitReq)) {
+    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
+    return -1;
+  }
+
+  return 0;
+}
+
+int32_t tGetSubmitMsgNextEx(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) {
+  ASSERT(pIter->len >= 0);
+
+  if (pIter->len == 0) {
+    pIter->len += sizeof(SSubmitReq);
+  } else {
+    if (pIter->len >= pIter->totalLen) {
+      ASSERT(0);
+    }
+
+    SSubmitBlk *pSubmitBlk = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
+    pIter->len += (sizeof(SSubmitBlk) + pIter->dataLen + pIter->schemaLen);
+    ASSERT(pIter->len > 0);
+  }
+
+  if (pIter->len > pIter->totalLen) {
+    terrno = TSDB_CODE_TDB_SUBMIT_MSG_MSSED_UP;
+    *pPBlock = NULL;
+    return -1;
+  }
+
+  if (pIter->len == pIter->totalLen) {
+    *pPBlock = NULL;
+  } else {
+    *pPBlock = (SSubmitBlk *)POINTER_SHIFT(pIter->pMsg, pIter->len);
+    pIter->uid = htobe64((*pPBlock)->uid);
+    pIter->suid = htobe64((*pPBlock)->suid);
+    pIter->sversion = htonl((*pPBlock)->sversion);
+    pIter->dataLen = htonl((*pPBlock)->dataLen);
+    pIter->schemaLen = htonl((*pPBlock)->schemaLen);
+    pIter->numOfRows = htons((*pPBlock)->numOfRows);
+  }
+  return 0;
+}
+
+int32_t tInitSubmitBlkIterEx(SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkIter *pIter) {
+  if (pMsgIter->dataLen <= 0) return -1;
+  pIter->totalLen = pMsgIter->dataLen;
+  pIter->len = 0;
+  pIter->row = (STSRow *)(pBlock->data + pMsgIter->schemaLen);
+  return 0;
+}
+
+STSRow *tGetSubmitBlkNextEx(SSubmitBlkIter *pIter) {
+  STSRow *row = pIter->row;
+
+  if (pIter->len >= pIter->totalLen) {
+    return NULL;
+  } else {
+    pIter->len += TD_ROW_LEN(row);
+    if (pIter->len < pIter->totalLen) {
+      pIter->row = POINTER_SHIFT(row, TD_ROW_LEN(row));
+    }
+    return row;
+  }
+}
+#endif
 int32_t tEncodeSEpSet(SCoder *pEncoder, const SEpSet *pEp) {
   if (tEncodeI8(pEncoder, pEp->inUse) < 0) return -1;
   if (tEncodeI8(pEncoder, pEp->numOfEps) < 0) return -1;

source/dnode/mgmt/exe/dmMain.c

@@ -24,6 +24,7 @@ static struct {
   bool         printVersion;
   char         envFile[PATH_MAX];
   char         apolloUrl[PATH_MAX];
+  const char **envCmd;
   SArray      *pArgs;  // SConfigPair
   SDnode      *pDnode;
   EDndNodeType ntype;
@@ -56,6 +57,9 @@ static void dmSetSignalHandle() {
 }
 
 static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
+  int32_t cmdEnvIndex = 0;
+  global.envCmd = taosMemoryMalloc(argc-1);
+  memset(global.envCmd, 0, argc-1);
   for (int32_t i = 1; i < argc; ++i) {
     if (strcmp(argv[i], "-c") == 0) {
       if (i < argc - 1) {
@@ -70,7 +74,7 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
       }
     } else if (strcmp(argv[i], "-a") == 0) {
       tstrncpy(global.apolloUrl, argv[++i], PATH_MAX);
-    } else if (strcmp(argv[i], "-e") == 0) {
+    } else if (strcmp(argv[i], "-E") == 0) {
       tstrncpy(global.envFile, argv[++i], PATH_MAX);
     } else if (strcmp(argv[i], "-n") == 0) {
       global.ntype = atoi(argv[++i]);
@@ -84,6 +88,9 @@ static int32_t dmParseArgs(int32_t argc, char const *argv[]) {
       global.dumpConfig = true;
     } else if (strcmp(argv[i], "-V") == 0) {
       global.printVersion = true;
+    } else if (strcmp(argv[i], "-e") == 0) {
+      global.envCmd[cmdEnvIndex] = argv[++i];
+      cmdEnvIndex++;
     } else {
     }
   }
@@ -129,7 +136,7 @@ static SDnodeOpt dmGetOpt() {
 static int32_t dmInitLog() {
   char logName[12] = {0};
   snprintf(logName, sizeof(logName), "%slog", dmLogName(global.ntype));
-  return taosCreateLog(logName, 1, configDir, global.envFile, global.apolloUrl, global.pArgs, 0);
+  return taosCreateLog(logName, 1, configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0);
 }
 
 static void dmSetProcInfo(int32_t argc, char **argv) {
@@ -168,6 +175,10 @@ static int32_t dmRunDnode() {
   return code;
 }
 
+static void taosCleanupArgs() {
+  if (global.envCmd != NULL) taosMemoryFree(global.envCmd);
+}
+
 int main(int argc, char const *argv[]) {
   if (!taosCheckSystemIsSmallEnd()) {
     printf("failed to start since on non-small-end machines\n");
@@ -176,26 +187,31 @@ int main(int argc, char const *argv[]) {
 
   if (dmParseArgs(argc, argv) != 0) {
     printf("failed to start since parse args error\n");
+    taosCleanupArgs();
     return -1;
   }
 
   if (global.generateGrant) {
     dmGenerateGrant();
+    taosCleanupArgs();
     return 0;
   }
 
   if (global.printVersion) {
     dmPrintVersion();
+    taosCleanupArgs();
     return 0;
   }
 
   if (dmInitLog() != 0) {
-    dError("failed to start since init log error");
+    printf("failed to start since init log error");
+    taosCleanupArgs();
     return -1;
   }
 
-  if (taosInitCfg(configDir, global.envFile, global.apolloUrl, global.pArgs, 0) != 0) {
+  if (taosInitCfg(configDir, global.envCmd, global.envFile, global.apolloUrl, global.pArgs, 0) != 0) {
     dError("failed to start since read config error");
+    taosCleanupArgs();
     return -1;
   }
 
@@ -203,9 +219,11 @@ int main(int argc, char const *argv[]) {
     dmDumpCfg();
     taosCleanupCfg();
     taosCloseLog();
+    taosCleanupArgs();
     return 0;
   }
 
   dmSetProcInfo(argc, (char **)argv);
+  taosCleanupArgs();
   return dmRunDnode();
 }

source/dnode/mgmt/interface/src/dmFile.c

@@ -124,7 +124,7 @@ TdFilePtr dmCheckRunning(const char *dataDir) {
   TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
   if (pFile == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
-    dError("failed to lock file:%s since %s", filepath, terrstr());
+    dError("failed to open file:%s since %s", filepath, terrstr());
     return NULL;
   }
 

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

@@ -476,7 +476,10 @@ int32_t mndScheduleStream(SMnode* pMnode, STrans* pTrans, SStreamObj* pStream) {
 int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscribeObj* pSub) {
   SSdb*       pSdb = pMnode->pSdb;
   SVgObj*     pVgroup = NULL;
-  SQueryPlan* pPlan = qStringToQueryPlan(pTopic->physicalPlan);
+  SQueryPlan* pPlan = NULL;
+  SSubplan*   plan = NULL;
+  if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
+    pPlan = qStringToQueryPlan(pTopic->physicalPlan);
     if (pPlan == NULL) {
       terrno = TSDB_CODE_QRY_INVALID_INPUT;
       return -1;
@@ -501,7 +504,8 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
       terrno = TSDB_CODE_MND_UNSUPPORTED_TOPIC;
       return -1;
     }
-  SSubplan* plan = nodesListGetNode(inner->pNodeList, 0);
+    plan = nodesListGetNode(inner->pNodeList, 0);
+  }
 
   int64_t             unexistKey = -1;
   SMqConsumerEpInSub* pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
@@ -519,38 +523,35 @@ int32_t mndSchedInitSubEp(SMnode* pMnode, const SMqTopicObj* pTopic, SMqSubscrib
     }
 
     pSub->vgNum++;
-    plan->execNode.nodeId = pVgroup->vgId;
-    plan->execNode.epSet = mndGetVgroupEpset(pMnode, pVgroup);
 
     SMqVgEp* pVgEp = taosMemoryMalloc(sizeof(SMqVgEp));
-    pVgEp->epSet = plan->execNode.epSet;
-    pVgEp->vgId = plan->execNode.nodeId;
-
-#if 0
-    SMqConsumerEp consumerEp = {0};
-    consumerEp.status = 0;
-    consumerEp.consumerId = -1;
-    consumerEp.epSet = plan->execNode.epSet;
-    consumerEp.vgId = plan->execNode.nodeId;
-#endif
+    pVgEp->epSet = mndGetVgroupEpset(pMnode, pVgroup);
+    pVgEp->vgId = pVgroup->vgId;
+    taosArrayPush(pEpInSub->vgs, &pVgEp);
 
     mDebug("init subscribption %s, assign vg: %d", pSub->key, pVgEp->vgId);
 
+    if (pTopic->subType == TOPIC_SUB_TYPE__TABLE) {
       int32_t msgLen;
+
+      plan->execNode.epSet = pVgEp->epSet;
+      plan->execNode.nodeId = pVgEp->vgId;
+
       if (qSubPlanToString(plan, &pVgEp->qmsg, &msgLen) < 0) {
         sdbRelease(pSdb, pVgroup);
         qDestroyQueryPlan(pPlan);
         terrno = TSDB_CODE_QRY_INVALID_INPUT;
         return -1;
       }
-    taosArrayPush(pEpInSub->vgs, &pVgEp);
+    } else {
+      pVgEp->qmsg = strdup("");
+    }
 
     ASSERT(taosHashGetSize(pSub->consumerHash) == 1);
 
     /*taosArrayPush(pSub->unassignedVg, &consumerEp);*/
   }
 
-  ASSERT(pEpInSub->vgs->size > 0);
   pEpInSub = taosHashGet(pSub->consumerHash, &unexistKey, sizeof(int64_t));
 
   ASSERT(pEpInSub->vgs->size > 0);

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

@@ -453,7 +453,6 @@ static void *mndBuildVCreateStbReq(SMnode *pMnode, SVgObj *pVgroup, SStbObj *pSt
       }
     }
     if (pStb->ast2Len > 0) {
-      int32_t qmsgLen2 = 0;
       if (mndConvertRSmaTask(pStb->pAst2, 0, 0, &pRSmaParam->qmsg2, &pRSmaParam->qmsg2Len) != TSDB_CODE_SUCCESS) {
         taosMemoryFreeClear(pRSmaParam->pFuncIds);
         taosMemoryFreeClear(pRSmaParam->qmsg1);

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

@@ -282,10 +282,10 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
   topicObj.version = 1;
   topicObj.sql = strdup(pCreate->sql);
   topicObj.sqlLen = strlen(pCreate->sql) + 1;
-  topicObj.ast = strdup(pCreate->ast);
-  topicObj.astLen = strlen(pCreate->ast) + 1;
 
   if (pCreate->ast && pCreate->ast[0]) {
+    topicObj.ast = strdup(pCreate->ast);
+    topicObj.astLen = strlen(pCreate->ast) + 1;
     topicObj.subType = TOPIC_SUB_TYPE__TABLE;
     topicObj.withTbName = 0;
     topicObj.withSchema = 0;
@@ -314,6 +314,9 @@ static int32_t mndCreateTopic(SMnode *pMnode, SNodeMsg *pReq, SCMCreateTopicReq
       return -1;
     }
   } else {
+    topicObj.ast = strdup("");
+    topicObj.astLen = 1;
+    topicObj.physicalPlan = strdup("");
     topicObj.subType = TOPIC_SUB_TYPE__DB;
     topicObj.withTbName = 1;
     topicObj.withSchema = 1;

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

@@ -51,7 +51,7 @@ static FORCE_INLINE tb_uid_t metaGenerateUid(SMeta* pMeta) { return tGenIdPI64()
 #define META_CHILD_TABLE  TD_CHILD_TABLE
 #define META_NORMAL_TABLE TD_NORMAL_TABLE
 
-int             metaCreateTable(SMeta* pMeta, STbCfg* pTbCfg);
+int             metaCreateTable(SMeta* pMeta, STbCfg* pTbCfg, STbDdlH* pHandle);
 int             metaDropTable(SMeta* pMeta, tb_uid_t uid);
 int             metaCommit(SMeta* pMeta);
 int32_t         metaCreateTSma(SMeta* pMeta, SSmaCfg* pCfg);
@@ -74,7 +74,7 @@ tb_uid_t        metaCtbCursorNext(SMCtbCursor* pCtbCur);
 // SMetaDB
 int  metaOpenDB(SMeta* pMeta);
 void metaCloseDB(SMeta* pMeta);
-int  metaSaveTableToDB(SMeta* pMeta, STbCfg* pTbCfg);
+int  metaSaveTableToDB(SMeta* pMeta, STbCfg* pTbCfg, STbDdlH* pHandle);
 int  metaRemoveTableFromDb(SMeta* pMeta, tb_uid_t uid);
 int  metaSaveSmaToDB(SMeta* pMeta, STSma* pTbCfg);
 int  metaRemoveSmaFromDb(SMeta* pMeta, int64_t indexUid);

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

@@ -57,7 +57,6 @@ int32_t tsdbDropTSmaData(STsdb *pTsdb, int64_t indexUid);
 int32_t tsdbInsertRSmaData(STsdb *pTsdb, char *msg);
 void    tsdbCleanupReadHandle(tsdbReaderT queryHandle);
 int32_t tdScanAndConvertSubmitMsg(SSubmitReq *pMsg);
-
 typedef enum {
   TSDB_FILE_HEAD = 0,  // .head
   TSDB_FILE_DATA,      // .data

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

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef _TD_VNODE_TSDB_SMA_H_
+#define _TD_VNODE_TSDB_SMA_H_
+
+#include "os.h"
+#include "thash.h"
+#include "tmsg.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef int32_t (*__tb_ddl_fn_t)(void *ahandle, void **result, void *p1, void *p2);
+
+struct STbDdlH {
+  void         *ahandle;
+  void         *result;
+  __tb_ddl_fn_t fp;
+};
+
+typedef struct {
+  tb_uid_t  suid;
+  SArray   *tbUids;
+  SHashObj *uidHash;
+} STbUidStore;
+
+static FORCE_INLINE int32_t tsdbUidStoreInit(STbUidStore **pStore) {
+  ASSERT(*pStore == NULL);
+  *pStore = taosMemoryCalloc(1, sizeof(STbUidStore));
+  if (*pStore == NULL) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    return TSDB_CODE_FAILED;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t tsdbUidStorePut(STbUidStore *pStore, tb_uid_t suid, tb_uid_t *uid);
+void    tsdbUidStoreDestory(STbUidStore *pStore);
+void   *tsdbUidStoreFree(STbUidStore *pStore);
+
+int32_t tsdbRegisterRSma(STsdb *pTsdb, SMeta *pMeta, SVCreateTbReq *pReq);
+int32_t tsdbFetchTbUidList(void *pTsdb, void **result, void *suid, void *uid);
+int32_t tsdbUpdateTbUidList(STsdb *pTsdb, STbUidStore *pUidStore);
+int32_t tsdbTriggerRSma(STsdb *pTsdb, SMeta *pMeta, void *pMsg, int32_t inputType);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_VNODE_TSDB_SMA_H_*/
\ No newline at end of file

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

@@ -103,6 +103,8 @@ struct SVnode {
 
 #define TD_VID(PVNODE) (PVNODE)->config.vgId
 
+typedef struct STbDdlH STbDdlH;
+
 // sma
 void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data);
 
@@ -116,6 +118,8 @@ void smaHandleRes(void *pVnode, int64_t smaId, const SArray *data);
 
 #include "vnodeSync.h"
 
+#include "tsdbSma.h"
+
 #ifdef __cplusplus
 }
 #endif

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

@@ -250,7 +250,7 @@ void metaCloseDB(SMeta *pMeta) {
   }
 }
 
-int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
+int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
   tb_uid_t       uid;
   SMetaDB       *pMetaDb;
   void          *pKey;
@@ -349,6 +349,12 @@ int metaSaveTableToDB(SMeta *pMeta, STbCfg *pTbCfg) {
     if (ret < 0) {
       return -1;
     }
+    // child table handle for rsma
+    if (pHandle && pHandle->fp) {
+      if (((*pHandle->fp)(pHandle->ahandle, &pHandle->result, &ctbIdxKey.suid, &uid)) < 0) {
+        return -1;
+      };
+    }
   } else if (pTbCfg->type == META_NORMAL_TABLE) {
     pKey = &uid;
     kLen = sizeof(uid);

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

@@ -15,7 +15,7 @@
 
 #include "vnodeInt.h"
 
-int metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg) {
+int metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg, STbDdlH *pHandle) {
   // Validate the tbOptions
   // if (metaValidateTbCfg(pMeta, pTbCfg) < 0) {
   //   // TODO: handle error
@@ -24,7 +24,7 @@ int metaCreateTable(SMeta *pMeta, STbCfg *pTbCfg) {
 
   // TODO: add atomicity
 
-  if (metaSaveTableToDB(pMeta, pTbCfg) < 0) {
+  if (metaSaveTableToDB(pMeta, pTbCfg, pHandle) < 0) {
     // TODO: handle error
     return -1;
   }

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

@@ -411,8 +411,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
 
   SMqDataBlkRsp rsp = {0};
   rsp.reqOffset = pReq->currentOffset;
+  rsp.withSchema = pExec->withSchema;
   rsp.blockData = taosArrayInit(0, sizeof(void*));
   rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
+  rsp.blockSchema = taosArrayInit(0, sizeof(void*));
 
   while (1) {
     consumerEpoch = atomic_load_32(&pExec->epoch);
@@ -512,6 +514,12 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           ASSERT(actualLen <= dataStrLen);
           taosArrayPush(rsp.blockDataLen, &actualLen);
           taosArrayPush(rsp.blockData, &buf);
+
+          if (pExec->withSchema) {
+            SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
+            taosArrayPush(rsp.blockSchema, &pSW);
+          }
+
           rsp.blockNum++;
         }
         // db subscribe
@@ -540,6 +548,10 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           ASSERT(actualLen <= dataStrLen);
           taosArrayPush(rsp.blockDataLen, &actualLen);
           taosArrayPush(rsp.blockData, &buf);
+
+          SSchemaWrapper* pSW = tCloneSSchemaWrapper(pExec->pExecReader[workerId]->pSchemaWrapper);
+          taosArrayPush(rsp.blockSchema, &pSW);
+
           rsp.blockNum++;
         }
       } else {
@@ -586,6 +598,8 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
   // TODO destroy
   taosArrayDestroy(rsp.blockData);
   taosArrayDestroy(rsp.blockDataLen);
+  taosArrayDestroyP(rsp.blockSchema, (FDelete)tDeleteSSchemaWrapper);
+
   return 0;
 }
 
@@ -827,12 +841,16 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) {
     pExec->pWalReader = walOpenReadHandle(pTq->pVnode->pWal);
     for (int32_t i = 0; i < 5; i++) {
       pExec->pExecReader[i] = tqInitSubmitMsgScanner(pTq->pVnode->pMeta);
+      if (pExec->subType == TOPIC_SUB_TYPE__TABLE) {
         SReadHandle handle = {
             .reader = pExec->pExecReader[i],
             .meta = pTq->pVnode->pMeta,
         };
         pExec->task[i] = qCreateStreamExecTaskInfo(pExec->qmsg, &handle);
         ASSERT(pExec->task[i]);
+      } else {
+        pExec->task[i] = NULL;
+      }
     }
     taosHashPut(pTq->execs, req.subKey, strlen(req.subKey), pExec, sizeof(STqExec));
     return 0;

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

@@ -65,7 +65,9 @@ bool tqNextDataBlock(STqReadHandle* pHandle) {
 
     /*pHandle->pBlock->uid = htobe64(pHandle->pBlock->uid);*/
     /*if (pHandle->tbUid == pHandle->pBlock->uid) {*/
-    ASSERT(pHandle->tbIdHash);
+    if (pHandle->tbIdHash == NULL) {
+      return true;
+    }
     void* ret = taosHashGet(pHandle->tbIdHash, &pHandle->pBlock->uid, sizeof(int64_t));
     if (ret != NULL) {
       /*printf("retrieve one tb %ld\n", pHandle->pBlock->uid);*/
@@ -107,6 +109,27 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
   *pNumOfRows = pHandle->pBlock->numOfRows;
   int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
 
+  if (colNumNeed == 0) {
+    *ppCols = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData));
+    if (*ppCols == NULL) {
+      return -1;
+    }
+
+    int32_t colMeta = 0;
+    while (colMeta < pSchemaWrapper->nCols) {
+      SSchema*        pColSchema = &pSchemaWrapper->pSchema[colMeta];
+      SColumnInfoData colInfo = {0};
+      colInfo.info.bytes = pColSchema->bytes;
+      colInfo.info.colId = pColSchema->colId;
+      colInfo.info.type = pColSchema->type;
+
+      if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
+        goto FAIL;
+      }
+      taosArrayPush(*ppCols, &colInfo);
+      colMeta++;
+    }
+  } else {
     if (colNumNeed > pSchemaWrapper->nCols) {
       colNumNeed = pSchemaWrapper->nCols;
     }
@@ -140,6 +163,7 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
         colNeed++;
       }
     }
+  }
 
   int32_t colActual = taosArrayGetSize(*ppCols);
   *pNumOfCols = colActual;

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

@@ -81,8 +81,8 @@ static STsdb *tsdbNew(const char *path, SVnode *pVnode, const STsdbCfg *pTsdbCfg
 
 static void tsdbFree(STsdb *pTsdb) {
   if (pTsdb) {
-    // tsdbFreeSmaEnv(REPO_TSMA_ENV(pTsdb));
-    // tsdbFreeSmaEnv(REPO_RSMA_ENV(pTsdb));
+    tsdbFreeSmaEnv(REPO_TSMA_ENV(pTsdb));
+    tsdbFreeSmaEnv(REPO_RSMA_ENV(pTsdb));
     tsdbFreeFS(pTsdb->fs);
     taosMemoryFreeClear(pTsdb->path);
     taosMemoryFree(pTsdb);

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

@@ -3636,6 +3636,8 @@ int32_t tsdbQuerySTableByTagCond(void* pMeta, uint64_t uid, TSKEY skey, const ch
     tsdbError("%p failed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
     terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
     goto _error;
+  } else {
+    tsdbDebug("%p succeed to get stable, uid:%" PRIu64 ", TID:0x%" PRIx64 " QID:0x%" PRIx64, pMeta, uid, taskId, reqId);
   }
 
   if (pTbCfg->type != META_SUPER_TABLE) {

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

@@ -83,6 +83,7 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
     case TDMT_VND_SUBMIT:
       pRsp->msgType = TDMT_VND_SUBMIT_RSP;
       vnodeProcessSubmitReq(pVnode, ptr, pRsp);
+      tsdbTriggerRSma(pVnode->pTsdb, pVnode->pMeta, ptr, STREAM_DATA_TYPE_SUBMIT_BLOCK);
       break;
     case TDMT_VND_MQ_VG_CHANGE:
       if (tqProcessVgChangeReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)),
@@ -101,7 +102,6 @@ int vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg
       }
     } break;
     case TDMT_VND_CREATE_SMA: {  // timeRangeSMA
-
       if (tsdbCreateTSma(pVnode->pTsdb, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead))) < 0) {
         // TODO
       }
@@ -277,19 +277,12 @@ int vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
 static int vnodeProcessCreateStbReq(SVnode *pVnode, void *pReq) {
   SVCreateTbReq vCreateTbReq = {0};
   tDeserializeSVCreateTbReq(pReq, &vCreateTbReq);
-  if (metaCreateTable(pVnode->pMeta, &(vCreateTbReq)) < 0) {
+  if (metaCreateTable(pVnode->pMeta, &(vCreateTbReq), NULL) < 0) {
     // TODO
     return -1;
   }
 
-  // TODO: remove the debug log
-  SRSmaParam *param = vCreateTbReq.stbCfg.pRSmaParam;
-  if (param) {
-    printf("qmsg1 len = %d, body = %s\n", param->qmsg1 ? (int32_t)strlen(param->qmsg1) : 0,
-           param->qmsg1 ? param->qmsg1 : "");
-    printf("qmsg1 len = %d, body = %s\n", param->qmsg2 ? (int32_t)strlen(param->qmsg2) : 0,
-           param->qmsg2 ? param->qmsg2 : "");
-  }
+  tsdbRegisterRSma(pVnode->pTsdb, pVnode->pMeta, &vCreateTbReq);
 
   taosMemoryFree(vCreateTbReq.stbCfg.pSchema);
   taosMemoryFree(vCreateTbReq.stbCfg.pTagSchema);
@@ -309,6 +302,13 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, SRpcMsg *pMsg, void *pReq, SR
   SVCreateTbBatchRsp vCreateTbBatchRsp = {0};
   tDeserializeSVCreateTbBatchReq(pReq, &vCreateTbBatchReq);
   int reqNum = taosArrayGetSize(vCreateTbBatchReq.pArray);
+
+  STbDdlH ddlHandle = {
+      .ahandle = pVnode->pTsdb,
+      .result = NULL,
+      .fp = tsdbFetchTbUidList,
+  };
+
   for (int i = 0; i < reqNum; i++) {
     SVCreateTbReq *pCreateTbReq = taosArrayGet(vCreateTbBatchReq.pArray, i);
 
@@ -324,7 +324,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, SRpcMsg *pMsg, void *pReq, SR
       taosArrayPush(vCreateTbBatchRsp.rspList, &rsp);
     }
 
-    if (metaCreateTable(pVnode->pMeta, pCreateTbReq) < 0) {
+    if (metaCreateTable(pVnode->pMeta, pCreateTbReq, &ddlHandle) < 0) {
       // TODO: handle error
       vError("vgId:%d, failed to create table: %s", TD_VID(pVnode), pCreateTbReq->name);
     }
@@ -348,6 +348,8 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, SRpcMsg *pMsg, void *pReq, SR
     }
   }
 
+  tsdbUpdateTbUidList(pVnode->pTsdb, ddlHandle.result);
+
   vTrace("vgId:%d process create %" PRIzu " tables", TD_VID(pVnode), taosArrayGetSize(vCreateTbBatchReq.pArray));
   taosArrayDestroy(vCreateTbBatchReq.pArray);
   if (vCreateTbBatchRsp.rspList) {

source/dnode/vnode/test/tsdbSmaTest.cpp

@@ -407,7 +407,7 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
     }
   }
 
-  EXPECT_EQ(tdScanAndConvertSubmitMsg(pMsg), TSDB_CODE_SUCCESS);
+  // EXPECT_EQ(tdScanAndConvertSubmitMsg(pMsg), TSDB_CODE_SUCCESS);
 
   EXPECT_EQ(tsdbUpdateSmaWindow(pTsdb, pMsg, 0), 0);
 

source/libs/executor/inc/executorimpl.h

@@ -399,8 +399,8 @@ typedef struct SOptrBasicInfo {
 // TODO move the resultrowsiz together with SOptrBasicInfo:rowCellInfoOffset
 typedef struct SAggSupporter {
   SHashObj*      pResultRowHashTable;  // quick locate the window object for each result
-  SHashObj*      pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
-  SArray*        pResultRowArrayList;  // The array list that contains the Result rows
+//  SHashObj*      pResultRowListSet;    // used to check if current ResultRowInfo has ResultRow object or not
+//  SArray*        pResultRowArrayList;  // The array list that contains the Result rows
   char*          keyBuf;               // window key buffer
   SDiskbasedBuf* pResultBuf;           // query result buffer based on blocked-wised disk file
   int32_t        resultRowSize;        // the result buffer size for each result row, with the meta data size for each row

source/libs/executor/src/executorimpl.c

@@ -268,7 +268,6 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
   int32_t numOfCols = LIST_LENGTH(pNode->pSlots);
 
   SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
-  pBlock->info.numOfCols = numOfCols;
   pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
 
   pBlock->info.blockId = pNode->dataBlockId;
@@ -294,6 +293,7 @@ SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
     taosArrayPush(pBlock->pDataBlock, &idata);
   }
 
+  pBlock->info.numOfCols = taosArrayGetSize(pBlock->pDataBlock);
   return pBlock;
 }
 
@@ -324,11 +324,16 @@ static void prepareResultListBuffer(SResultRowInfo* pResultRowInfo, jmp_buf env)
     newCapacity = (int64_t)(pResultRowInfo->capacity * 1.5);
   }
 
-  if (newCapacity == pResultRowInfo->capacity) {
+  if (newCapacity <= pResultRowInfo->capacity) {
     newCapacity += 4;
   }
 
-  pResultRowInfo->pPosition = taosMemoryRealloc(pResultRowInfo->pPosition, newCapacity * sizeof(SResultRowPosition));
+  char* p = taosMemoryRealloc(pResultRowInfo->pPosition, newCapacity * sizeof(SResultRowPosition));
+  if (p == NULL) {
+    longjmp(env, TSDB_CODE_OUT_OF_MEMORY);
+  }
+
+  pResultRowInfo->pPosition = (SResultRowPosition*)p;
 
   int32_t inc = (int32_t)newCapacity - pResultRowInfo->capacity;
   memset(&pResultRowInfo->pPosition[pResultRowInfo->capacity], 0, sizeof(SResultRowPosition) * inc);
@@ -419,55 +424,33 @@ SResultRow* getNewResultRow_rv(SDiskbasedBuf* pResultBuf, int64_t tableGroupId,
  * | 8 bytes  | actual length |
  * +----------+---------------+
  */
-static SResultRow* doSetResultOutBufByKey_rv(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, int64_t uid,
+static SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, int64_t uid,
                                              char* pData, int16_t bytes, bool masterscan, uint64_t groupId,
                                              SExecTaskInfo* pTaskInfo, bool isIntervalQuery, SAggSupporter* pSup) {
-  bool existInCurrentResusltRowInfo = false;
   SET_RES_WINDOW_KEY(pSup->keyBuf, pData, bytes, groupId);
 
   SResultRowPosition* p1 =
       (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
 
+  SResultRow* pResult = NULL;
+
   // in case of repeat scan/reverse scan, no new time window added.
   if (isIntervalQuery) {
-    if (!masterscan) {  // the *p1 may be NULL in case of sliding+offset exists.
-      if (p1 != NULL) {
-        return getResultRowByPos(pResultBuf, p1);
-      } else {
-        return NULL;
-      }
-    }
-
-    if (p1 != NULL) {
-      if (pResultRowInfo->size == 0) {
-        existInCurrentResusltRowInfo =
-            false;  // this time window created by other timestamp that does not belongs to current table.
-      } else if (pResultRowInfo->size == 1) {
-        SResultRowPosition* p = &pResultRowInfo->pPosition[0];
-        existInCurrentResusltRowInfo = (p->pageId == p1->pageId && p->offset == p1->offset);
-      } else {  // check if current pResultRowInfo contains the existInCurrentResusltRowInfo pResultRow
-        SET_RES_EXT_WINDOW_KEY(pSup->keyBuf, pData, bytes, uid, pResultRowInfo);
-        int64_t* index = taosHashGet(pSup->pResultRowListSet, pSup->keyBuf, GET_RES_EXT_WINDOW_KEY_LEN(bytes));
-        if (index != NULL) {
-          // TODO check the scan order for current opened time window
-          existInCurrentResusltRowInfo = true;
-        } else {
-          existInCurrentResusltRowInfo = false;
-        }
-      }
+    if (masterscan && p1 != NULL) {  // the *p1 may be NULL in case of sliding+offset exists.
+      pResult = getResultRowByPos(pResultBuf, p1);
     }
   } else {
     // In case of group by column query, the required SResultRow object must be existInCurrentResusltRowInfo in the
     // pResultRowInfo object.
     if (p1 != NULL) {
-      return getResultRowByPos(pResultBuf, p1);
+      pResult = getResultRowByPos(pResultBuf, p1);
     }
   }
 
-  SResultRow* pResult = NULL;
-  if (!existInCurrentResusltRowInfo) {
     // 1. close current opened time window
-    if (pResultRowInfo->cur.pageId != -1) {  // todo extract function
+  if (pResultRowInfo->cur.pageId != -1 && ((pResult == NULL) || (pResult->pageId != pResultRowInfo->cur.pageId &&
+                                            pResult->offset != pResultRowInfo->cur.offset))) {
+    // todo extract function
     SResultRowPosition pos = pResultRowInfo->cur;
     SFilePage*         pPage = getBufPage(pResultBuf, pos.pageId);
     SResultRow*        pRow = (SResultRow*)((char*)pPage + pos.offset);
@@ -475,31 +458,21 @@ static SResultRow* doSetResultOutBufByKey_rv(SDiskbasedBuf* pResultBuf, SResultR
     releaseBufPage(pResultBuf, pPage);
   }
 
+  // allocate a new buffer page
   prepareResultListBuffer(pResultRowInfo, pTaskInfo->env);
-    if (p1 == NULL) {
+  if (pResult == NULL) {
     pResult = getNewResultRow_rv(pResultBuf, groupId, pSup->resultRowSize);
     initResultRow(pResult);
 
     // add a new result set for a new group
     SResultRowPosition pos = {.pageId = pResult->pageId, .offset = pResult->offset};
-      taosHashPut(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes), &pos,
-                  sizeof(SResultRowPosition));
-      SResultRowCell cell = {.groupId = groupId, .pos = pos};
-      taosArrayPush(pSup->pResultRowArrayList, &cell);
-    } else {
-      pResult = getResultRowByPos(pResultBuf, p1);
+    taosHashPut(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes), &pos, sizeof(SResultRowPosition));
   }
 
   // 2. set the new time window to be the new active time window
   pResultRowInfo->pPosition[pResultRowInfo->size++] =
       (SResultRowPosition){.pageId = pResult->pageId, .offset = pResult->offset};
   pResultRowInfo->cur = (SResultRowPosition){.pageId = pResult->pageId, .offset = pResult->offset};
-    SET_RES_EXT_WINDOW_KEY(pSup->keyBuf, pData, bytes, uid, pResultRowInfo);
-    taosHashPut(pSup->pResultRowListSet, pSup->keyBuf, GET_RES_EXT_WINDOW_KEY_LEN(bytes), &pResultRowInfo->cur,
-                POINTER_BYTES);
-  } else {
-    pResult = getResultRowByPos(pResultBuf, p1);
-  }
 
   // too many time window in query
   if (pResultRowInfo->size > MAX_INTERVAL_TIME_WINDOW) {
@@ -656,7 +629,7 @@ static int32_t setResultOutputBufByKey_rv(SResultRowInfo* pResultRowInfo, int64_
                                           int32_t numOfOutput, int32_t* rowCellInfoOffset, SAggSupporter* pAggSup,
                                           SExecTaskInfo* pTaskInfo) {
   assert(win->skey <= win->ekey);
-  SResultRow* pResultRow = doSetResultOutBufByKey_rv(pAggSup->pResultBuf, pResultRowInfo, id, (char*)&win->skey,
+  SResultRow* pResultRow = doSetResultOutBufByKey(pAggSup->pResultBuf, pResultRowInfo, id, (char*)&win->skey,
                                                      TSDB_KEYSIZE, masterscan, tableGroupId, pTaskInfo, true, pAggSup);
 
   if (pResultRow == NULL) {
@@ -1059,6 +1032,8 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc
     pColInfo->info.bytes = tDataTypes[type].bytes;
 
     pInput->pData[paramIndex] = pColInfo;
+  } else {
+    pColInfo = pInput->pData[paramIndex];
   }
 
   ASSERT(!IS_VAR_DATA_TYPE(type));
@@ -1091,6 +1066,7 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt
 
     SInputColumnInfoData* pInput = &pCtx[i].input;
     pInput->uid = pBlock->info.uid;
+    pInput->colDataAggIsSet = false;
 
     SExprInfo* pOneExpr = &pOperator->pExpr[i];
     for (int32_t j = 0; j < pOneExpr->base.numOfParams; ++j) {
@@ -1707,7 +1683,7 @@ int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char*
   SResultRowInfo* pResultRowInfo = &binfo->resultRowInfo;
   SqlFunctionCtx* pCtx = binfo->pCtx;
 
-  SResultRow* pResultRow = doSetResultOutBufByKey_rv(pBuf, pResultRowInfo, groupId, (char*)pData, bytes, true, groupId,
+  SResultRow* pResultRow = doSetResultOutBufByKey(pBuf, pResultRowInfo, groupId, (char*)pData, bytes, true, groupId,
                                                      pTaskInfo, false, pAggSup);
   assert(pResultRow != NULL);
 
@@ -2713,7 +2689,7 @@ void setFunctionResultOutput(SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t
 
   int64_t     tid = 0;
   int64_t     groupId = 0;
-  SResultRow* pRow = doSetResultOutBufByKey_rv(pSup->pResultBuf, pResultRowInfo, tid, (char*)&tid, sizeof(tid), true,
+  SResultRow* pRow = doSetResultOutBufByKey(pSup->pResultBuf, pResultRowInfo, tid, (char*)&tid, sizeof(tid), true,
                                                groupId, pTaskInfo, false, pSup);
 
   for (int32_t i = 0; i < pDataBlock->info.numOfCols; ++i) {
@@ -3001,7 +2977,7 @@ void doSetTableGroupOutputBuf(SAggOperatorInfo* pAggInfo, int32_t numOfOutput, u
   int32_t*        rowCellInfoOffset = pAggInfo->binfo.rowCellInfoOffset;
 
   SResultRow* pResultRow =
-      doSetResultOutBufByKey_rv(pAggInfo->aggSup.pResultBuf, pResultRowInfo, uid, (char*)&groupId, sizeof(groupId),
+      doSetResultOutBufByKey(pAggInfo->aggSup.pResultBuf, pResultRowInfo, uid, (char*)&groupId, sizeof(groupId),
                                 true, groupId, pTaskInfo, false, &pAggInfo->aggSup);
   assert(pResultRow != NULL);
 
@@ -3775,6 +3751,9 @@ int32_t setSDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLoadI
       }
 
       memcpy(pColInfoData->pData, pStart, colLen[i]);
+      //TODO setting this flag to true temporarily so aggregate function on stable will
+      //examine NULL value for non-primary key column
+      pColInfoData->hasNull = true;
       pStart += colLen[i];
     }
   } else {  // extract data according to pColList
@@ -4895,8 +4874,8 @@ bool aggDecodeResultRow(SOperatorInfo* pOperator, SAggSupporter* pSup, SOptrBasi
     initResultRow(resultRow);
     prepareResultListBuffer(&pInfo->resultRowInfo, pOperator->pTaskInfo->env);
     //    pInfo->resultRowInfo.cur = pInfo->resultRowInfo.size;
-    pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] =
-        (SResultRowPosition){.pageId = resultRow->pageId, .offset = resultRow->offset};
+//    pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] =
+//        (SResultRowPosition){.pageId = resultRow->pageId, .offset = resultRow->offset};
     pInfo->resultRowInfo.cur = (SResultRowPosition){.pageId = resultRow->pageId, .offset = resultRow->offset};
   }
 
@@ -5609,10 +5588,10 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n
   pAggSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
   pAggSup->keyBuf = taosMemoryCalloc(1, keyBufSize + POINTER_BYTES + sizeof(int64_t));
   pAggSup->pResultRowHashTable = taosHashInit(10, hashFn, true, HASH_NO_LOCK);
-  pAggSup->pResultRowListSet = taosHashInit(100, hashFn, false, HASH_NO_LOCK);
-  pAggSup->pResultRowArrayList = taosArrayInit(10, sizeof(SResultRowCell));
+//  pAggSup->pResultRowListSet = taosHashInit(100, hashFn, false, HASH_NO_LOCK);
+//  pAggSup->pResultRowArrayList = taosArrayInit(10, sizeof(SResultRowCell));
 
-  if (pAggSup->keyBuf == NULL || pAggSup->pResultRowArrayList == NULL || pAggSup->pResultRowListSet == NULL ||
+  if (pAggSup->keyBuf == NULL /*|| pAggSup->pResultRowArrayList == NULL || pAggSup->pResultRowListSet == NULL*/ ||
       pAggSup->pResultRowHashTable == NULL) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
@@ -5628,8 +5607,8 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n
 static void cleanupAggSup(SAggSupporter* pAggSup) {
   taosMemoryFreeClear(pAggSup->keyBuf);
   taosHashCleanup(pAggSup->pResultRowHashTable);
-  taosHashCleanup(pAggSup->pResultRowListSet);
-  taosArrayDestroy(pAggSup->pResultRowArrayList);
+//  taosHashCleanup(pAggSup->pResultRowListSet);
+//  taosArrayDestroy(pAggSup->pResultRowArrayList);
   destroyDiskbasedBuf(pAggSup->pResultBuf);
 }
 
@@ -6352,7 +6331,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
 
       int32_t     numOfCols = 0;
       tsdbReaderT pDataReader = doCreateDataReader(pTableScanNode, pHandle, pTableGroupInfo, (uint64_t)queryId, taskId);
-      if (pDataReader == NULL) {
+      if (pDataReader == NULL && terrno != 0) {
         return NULL;
       }
 

source/libs/function/src/builtins.c

@@ -355,7 +355,7 @@ static int32_t translateToIso8601(SFunctionNode* pFunc, char* pErrBuf, int32_t l
     return invaildFuncParaTypeErrMsg(pErrBuf, len, pFunc->functionName);
   }
 
-  pFunc->node.resType = (SDataType) { .bytes = 24, .type = TSDB_DATA_TYPE_BINARY};
+  pFunc->node.resType = (SDataType) { .bytes = 64, .type = TSDB_DATA_TYPE_BINARY};
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/function/src/udfd.c

@@ -548,7 +548,7 @@ static int32_t udfdParseArgs(int32_t argc, char *argv[]) {
 static int32_t udfdInitLog() {
   char logName[12] = {0};
   snprintf(logName, sizeof(logName), "%slog", "udfd");
-  return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, 0);
+  return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, NULL, 0);
 }
 
 void udfdCtrlAllocBufCb(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
@@ -656,7 +656,7 @@ int main(int argc, char *argv[]) {
     return -1;
   }
 
-  if (taosInitCfg(configDir, NULL, NULL, NULL, 0) != 0) {
+  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) {
     fnError("failed to start since read config error");
     return -1;
   }

source/libs/function/test/runUdf.c

@@ -31,13 +31,13 @@ static int32_t parseArgs(int32_t argc, char *argv[]) {
 static int32_t initLog() {
   char logName[12] = {0};
   snprintf(logName, sizeof(logName), "%slog", "udfc");
-  return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, 0);
+  return taosCreateLog(logName, 1, configDir, NULL, NULL, NULL, NULL, 0);
 }
 
 int main(int argc, char *argv[]) {
   parseArgs(argc, argv);
   initLog();
-  if (taosInitCfg(configDir, NULL, NULL, NULL, 0) != 0) {
+  if (taosInitCfg(configDir, NULL, NULL, NULL, NULL, 0) != 0) {
     fnError("failed to start since read config error");
     return -1;
   }

source/libs/scalar/src/scalar.c

@@ -517,8 +517,8 @@ EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
     res->node.resType = node->node.resType;
     int32_t type = output.columnData->info.type;
     if (IS_VAR_DATA_TYPE(type)) {
-      res->datum.p = output.columnData->pData;
-      output.columnData->pData = NULL;
+      res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1);
+      memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData));
     } else {
       memcpy(nodesGetValueFromNode(res), output.columnData->pData, tDataTypes[type].bytes);
     }

source/libs/sync/CMakeLists.txt

@@ -11,7 +11,7 @@ target_link_libraries(
 
 target_include_directories(
     sync
-    PUBLIC "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    PUBLIC "${TD_SOURCE_DIR}/include/libs/sync"
     PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/inc"
 )
 

source/libs/sync/test/CMakeLists.txt

@@ -199,197 +199,197 @@ target_sources(syncApplyMsgTest
 
 target_include_directories(syncTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncEnvTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncPingTimerTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncIOTickQTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncIOTickPingTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncIOSendMsgTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncIOClientTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncIOServerTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncRaftStoreTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncEnqTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncIndexTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncInitTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncUtilTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncVotesGrantedTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncVotesRespondTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncIndexMgrTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncLogStoreTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncEntryTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncRequestVoteTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncRequestVoteReplyTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncAppendEntriesTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncAppendEntriesReplyTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncClientRequestTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncTimeoutTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncPingTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncPingReplyTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncRpcMsgTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncPingTimerTest2
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncPingSelfTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncElectTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncEncodeTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncWriteTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncReplicateTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncRefTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncLogStoreCheck
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncRaftCfgTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncRespMgrTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncSnapshotTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 target_include_directories(syncApplyMsgTest
     PUBLIC
-    "${CMAKE_SOURCE_DIR}/include/libs/sync"
+    "${TD_SOURCE_DIR}/include/libs/sync"
     "${CMAKE_CURRENT_SOURCE_DIR}/../inc"
 )
 

source/util/src/tconfig.c

@@ -18,6 +18,9 @@
 #include "taoserror.h"
 #include "tlog.h"
 #include "tutil.h"
+#include "tenv.h"
+#include "cJSON.h"
+#include "tjson.h"
 
 #define CFG_NAME_PRINT_LEN 24
 #define CFG_SRC_PRINT_LEN  12
@@ -25,6 +28,7 @@
 int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath);
 int32_t cfgLoadFromEnvFile(SConfig *pConfig, const char *filepath);
 int32_t cfgLoadFromEnvVar(SConfig *pConfig);
+int32_t cfgLoadFromEnvCmd(SConfig *pConfig, const char **envCmd);
 int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url);
 int32_t cfgSetItem(SConfig *pConfig, const char *name, const char *value, ECfgSrcType stype);
 
@@ -45,7 +49,7 @@ SConfig *cfgInit() {
   return pCfg;
 }
 
-int32_t cfgLoad(SConfig *pCfg, ECfgSrcType cfgType, const char *sourceStr) {
+int32_t cfgLoad(SConfig *pCfg, ECfgSrcType cfgType, const void *sourceStr) {
   switch (cfgType) {
     case CFG_STYPE_CFG_FILE:
       return cfgLoadFromCfgFile(pCfg, sourceStr);
@@ -55,6 +59,8 @@ int32_t cfgLoad(SConfig *pCfg, ECfgSrcType cfgType, const char *sourceStr) {
       return cfgLoadFromEnvVar(pCfg);
     case CFG_STYPE_APOLLO_URL:
       return cfgLoadFromApollUrl(pCfg, sourceStr);
+    case CFG_STYPE_ENV_CMD:
+      return cfgLoadFromEnvCmd(pCfg, (const char **)sourceStr);
     default:
       return -1;
   }
@@ -463,6 +469,8 @@ const char *cfgStypeStr(ECfgSrcType type) {
       return "arg_list";
     case CFG_STYPE_TAOS_OPTIONS:
       return "taos_options";
+    case CFG_STYPE_ENV_CMD:
+      return "env_cmd";
     default:
       return "invalid";
   }
@@ -584,12 +592,154 @@ void cfgDumpCfg(SConfig *pCfg, bool tsc, bool dump) {
 }
 
 int32_t cfgLoadFromEnvVar(SConfig *pConfig) {
-  uDebug("load from env variables not implemented yet");
+  char   *line = NULL, *name, *value, *value2, *value3;
+  int32_t olen, vlen, vlen2, vlen3;
+  ssize_t _bytes = 0;
+  TdCmdPtr pCmd = taosOpenCmd("set");
+  if (pCmd == NULL) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    return -1;
+  }
+  while (!taosEOFCmd(pCmd)) {
+    name = value = value2 = value3 = NULL;
+    olen = vlen = vlen2 = vlen3 = 0;
+
+    _bytes = taosGetLineCmd(pCmd, &line);
+    if (_bytes < 0) {
+      break;
+    }
+    if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
+    taosEnvToCfg(line, line);
+
+    paGetToken(line, &name, &olen);
+    if (olen == 0) continue;
+    name[olen] = 0;
+
+    paGetToken(name + olen + 1, &value, &vlen);
+    if (vlen == 0) continue;
+    value[vlen] = 0;
+
+    paGetToken(value + vlen + 1, &value2, &vlen2);
+    if (vlen2 != 0) {
+      value2[vlen2] = 0;
+      paGetToken(value2 + vlen2 + 1, &value3, &vlen3);
+      if (vlen3 != 0) value3[vlen3] = 0;
+    }
+
+    cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_VAR);
+    if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
+      cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_VAR);
+    }
+  }
+
+  taosCloseCmd(&pCmd);
+  if (line != NULL) taosMemoryFreeClear(line);
+
+  uInfo("load from env variables cfg success");
+  return 0;
+}
+
+int32_t cfgLoadFromEnvCmd(SConfig *pConfig, const char **envCmd) {
+  char   *buf, *name, *value, *value2, *value3;
+  int32_t olen, vlen, vlen2, vlen3;
+  int32_t index = 0;
+  if (envCmd == NULL) return 0;
+  while (envCmd[index]!=NULL) {
+    buf = taosMemoryMalloc(strlen(envCmd[index]));
+    taosEnvToCfg(envCmd[index], buf);
+    index++;
+    
+    name = value = value2 = value3 = NULL;
+    olen = vlen = vlen2 = vlen3 = 0;
+
+    paGetToken(buf, &name, &olen);
+    if (olen == 0) continue;
+    name[olen] = 0;
+
+    paGetToken(name + olen + 1, &value, &vlen);
+    if (vlen == 0) continue;
+    value[vlen] = 0;
+
+    paGetToken(value + vlen + 1, &value2, &vlen2);
+    if (vlen2 != 0) {
+      value2[vlen2] = 0;
+      paGetToken(value2 + vlen2 + 1, &value3, &vlen3);
+      if (vlen3 != 0) value3[vlen3] = 0;
+    }
+
+    cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_CMD);
+    if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
+      cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_CMD);
+    }
+
+    taosMemoryFree(buf);
+  }
+
+  uInfo("load from env cmd cfg success");
   return 0;
 }
 
-int32_t cfgLoadFromEnvFile(SConfig *pConfig, const char *filepath) {
-  uDebug("load from env file not implemented yet");
+int32_t cfgLoadFromEnvFile(SConfig *pConfig, const char *envFile) {
+  char   *line = NULL, *name, *value, *value2, *value3;
+  int32_t olen, vlen, vlen2, vlen3;
+  ssize_t _bytes = 0;
+
+  const char *filepath = ".env";
+  if (envFile != NULL && strlen(envFile)>0) {
+    if (!taosCheckExistFile(envFile)) {
+      uError("fial to load env file: %s", envFile);
+      return -1;
+    }
+    filepath = envFile;
+  }else {
+    if (!taosCheckExistFile(filepath)) {
+      uInfo("fial to load env file: %s", filepath);
+      return 0;
+    }
+  }
+
+  TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ | TD_FILE_STREAM);
+  if (pFile == NULL) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    return -1;
+  }
+
+  while (!taosEOFFile(pFile)) {
+    name = value = value2 = value3 = NULL;
+    olen = vlen = vlen2 = vlen3 = 0;
+
+    _bytes = taosGetLineFile(pFile, &line);
+    if (_bytes <= 0) {
+      break;
+    }
+    if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
+    taosEnvToCfg(line, line);
+
+    paGetToken(line, &name, &olen);
+    if (olen == 0) continue;
+    name[olen] = 0;
+
+    paGetToken(name + olen + 1, &value, &vlen);
+    if (vlen == 0) continue;
+    value[vlen] = 0;
+
+    paGetToken(value + vlen + 1, &value2, &vlen2);
+    if (vlen2 != 0) {
+      value2[vlen2] = 0;
+      paGetToken(value2 + vlen2 + 1, &value3, &vlen3);
+      if (vlen3 != 0) value3[vlen3] = 0;
+    }
+
+    cfgSetItem(pConfig, name, value, CFG_STYPE_ENV_FILE);
+    if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
+      cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_ENV_FILE);
+    }
+  }
+
+  taosCloseFile(&pFile);
+  if (line != NULL) taosMemoryFreeClear(line);
+
+  uInfo("load from env cfg file %s success", filepath);
   return 0;
 }
 
@@ -617,11 +767,11 @@ int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath) {
     olen = vlen = vlen2 = vlen3 = 0;
 
     _bytes = taosGetLineFile(pFile, &line);
-    if (_bytes < 0) {
+    if (_bytes <= 0) {
       break;
     }
 
-    line[_bytes - 1] = 0;
+    if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
 
     paGetToken(line, &name, &olen);
     if (olen == 0) continue;
@@ -659,6 +809,192 @@ int32_t cfgLoadFromCfgFile(SConfig *pConfig, const char *filepath) {
 }
 
 int32_t cfgLoadFromApollUrl(SConfig *pConfig, const char *url) {
-  uDebug("load from apoll url not implemented yet");
+  char   *cfgLineBuf = NULL, *name, *value, *value2, *value3;
+  int32_t olen, vlen, vlen2, vlen3;
+  if (url == NULL || strlen(url) == 0) {
+    uInfo("fail to load apoll url");
+    return 0;
+  }
+  
+  char *p = strchr(url, ':');
+  if (p == NULL) {
+    uError("fail to load apoll url: %s, unknown format", url);
+    return -1;
+  }
+  p++;
+
+  if (bcmp(url, "jsonFile", 8) == 0) {
+    char *filepath = p;
+    if (!taosCheckExistFile(filepath)) {
+      uError("fial to load json file: %s", filepath);
+      return -1;
+    }
+
+    TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ);
+    if (pFile == NULL) {
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      return -1;
+    }
+    size_t fileSize = taosLSeekFile(pFile, 0, SEEK_END);
+    char *buf = taosMemoryMalloc(fileSize);
+    taosLSeekFile(pFile, 0, SEEK_SET);
+    if(taosReadFile(pFile, buf, fileSize) <= 0) {
+      taosCloseFile(&pFile);
+      uError("load json file error: %s", filepath);
+      return -1;
+    }
+    taosCloseFile(&pFile);
+    SJson* pJson = tjsonParse(buf);
+    if (NULL == pJson) {
+      const char *jsonParseError = tjsonGetError();
+      if (jsonParseError != NULL) {
+        uError("load json file parse error: %s", jsonParseError);
+      }
+      return -1;
+    }
+    taosMemoryFreeClear(buf);
+
+    int32_t jsonArraySize = tjsonGetArraySize(pJson);
+    for(int32_t i = 0; i < jsonArraySize; i++) {
+      cJSON* item = tjsonGetArrayItem(pJson, i);
+      if (item == NULL) break;
+      char *itemName = NULL, *itemValueString = NULL;
+      tjsonGetObjectName(item, &itemName);
+      tjsonGetObjectName(item, &itemName);
+      tjsonGetObjectValueString(item, &itemValueString);
+      if (itemValueString != NULL && itemName != NULL) {
+        size_t itemNameLen = strlen(itemName);
+        size_t itemValueStringLen = strlen(itemValueString);
+        cfgLineBuf = taosMemoryMalloc(itemNameLen + itemValueStringLen + 2);
+        memcpy(cfgLineBuf, itemName, itemNameLen);
+        cfgLineBuf[itemNameLen] = ' ';
+        memcpy(&cfgLineBuf[itemNameLen+1], itemValueString, itemValueStringLen);
+        cfgLineBuf[itemNameLen + itemValueStringLen + 1] = '\0';
+
+        paGetToken(cfgLineBuf, &name, &olen);
+        if (olen == 0) continue;
+        name[olen] = 0;
+
+        paGetToken(name + olen + 1, &value, &vlen);
+        if (vlen == 0) continue;
+        value[vlen] = 0;
+
+        paGetToken(value + vlen + 1, &value2, &vlen2);
+        if (vlen2 != 0) {
+          value2[vlen2] = 0;
+          paGetToken(value2 + vlen2 + 1, &value3, &vlen3);
+          if (vlen3 != 0) value3[vlen3] = 0;
+        }
+        cfgSetItem(pConfig, name, value, CFG_STYPE_APOLLO_URL);
+        if (value2 != NULL && value3 != NULL && value2[0] != 0 && value3[0] != 0 && strcasecmp(name, "dataDir") == 0) {
+          cfgSetTfsItem(pConfig, name, value, value2, value3, CFG_STYPE_APOLLO_URL);
+        }
+      }
+    }
+    tjsonDelete(pJson);
+
+  // } else if (bcmp(url, "jsonUrl", 7) == 0) {
+  // } else if (bcmp(url, "etcdUrl", 7) == 0) {
+  } else {
+    uError("Unsupported url: %s", url);
+    return -1;
+  }
+
+  uInfo("load from apoll url not implemented yet");
+  return 0;
+}
+
+int32_t cfgGetApollUrl(const char **envCmd, const char *envFile, char* apolloUrl) {
+  int32_t index = 0;
+  if (envCmd == NULL) return 0;
+  while (envCmd[index]!=NULL) {
+    if (bcmp(envCmd[index], "TAOS_APOLLO_URL", 14) == 0) {
+      char *p = strchr(envCmd[index], '=');
+      if (p != NULL) {
+        p++;
+        if (*p == '\'') {
+          p++;
+          p[strlen(p)-1] = '\0';
+        }
+        memcpy(apolloUrl, p, TMIN(strlen(p)+1,PATH_MAX));
+        uInfo("get apollo url from env cmd success");
+        return 0;
+      }
+    }
+    index++;
+  }
+
+  char   *line = NULL;
+  ssize_t _bytes = 0;
+  TdCmdPtr pCmd = taosOpenCmd("set");
+  if (pCmd != NULL) {
+    while (!taosEOFCmd(pCmd)) {
+      _bytes = taosGetLineCmd(pCmd, &line);
+      if (_bytes < 0) {
+        break;
+      }
+      if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
+      if (bcmp(line, "TAOS_APOLLO_URL", 14) == 0) {
+        char *p = strchr(line, '=');
+        if (p != NULL) {
+          p++;
+          if (*p == '\'') {
+            p++;
+            p[strlen(p)-1] = '\0';
+          }
+          memcpy(apolloUrl, p, TMIN(strlen(p)+1,PATH_MAX));
+          uInfo("get apollo url from env variables success, apolloUrl=%s",apolloUrl);
+          taosCloseCmd(&pCmd);
+          if (line != NULL) taosMemoryFreeClear(line);
+          return 0;
+        }
+      }
+    }
+    taosCloseCmd(&pCmd);
+    if (line != NULL) taosMemoryFreeClear(line);
+  }
+
+  const char *filepath = ".env";
+  if (envFile != NULL && strlen(envFile)>0) {
+    if (!taosCheckExistFile(envFile)) {
+      uError("fial to load env file: %s", envFile);
+      return -1;
+    }
+    filepath = envFile;
+  }else {
+    if (!taosCheckExistFile(filepath)) {
+      uInfo("fial to load env file: %s", filepath);
+      return 0;
+    }
+  }
+  TdFilePtr pFile = taosOpenFile(filepath, TD_FILE_READ | TD_FILE_STREAM);
+  if (pFile != NULL) {
+    while (!taosEOFFile(pFile)) {
+      _bytes = taosGetLineFile(pFile, &line);
+      if (_bytes <= 0) {
+        break;
+      }
+      if(line[_bytes - 1] == '\n') line[_bytes - 1] = 0;
+      if (bcmp(line, "TAOS_APOLLO_URL", 14) == 0) {
+        char *p = strchr(line, '=');
+        if (p != NULL) {
+          p++;
+          if (*p == '\'') {
+            p++;
+            p[strlen(p)-1] = '\0';
+          }
+          memcpy(apolloUrl, p, TMIN(strlen(p)+1,PATH_MAX));
+          taosCloseFile(&pFile);
+          if (line != NULL) taosMemoryFreeClear(line);
+          uInfo("get apollo url from env file success");
           return 0;
+        }
+      }
+    }
+    taosCloseFile(&pFile);
+    if (line != NULL) taosMemoryFreeClear(line);
+  }
+
+  uInfo("fail get apollo url from cmd env file");
+  return -1;
 }
\ No newline at end of file

source/util/src/tenv.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#define _DEFAULT_SOURCE
+#include "tenv.h"
+#include "tconfig.h"
+
+static char toLowChar(char c) { return (c > 'Z' || c < 'A' ? c : (c - 'A' + 'a')); }
+
+int32_t taosEnvNameToCfgName(const char *envNameStr, char *cfgNameStr, int32_t cfgNameMaxLen) {
+  if (envNameStr == NULL || cfgNameStr == NULL) return -1;
+  char *p = cfgNameStr;
+  if (envNameStr[0] != 'T' || envNameStr[1] != 'A' || envNameStr[2] != 'O' || envNameStr[3] != 'S' ||
+      envNameStr[4] != '_') {
+    // if(p != envNameStr) strncpy(p, envNameStr, cfgNameMaxLen - 1);
+    // p[cfgNameMaxLen - 1] = '\0';
+    // return strlen(cfgNameStr);
+    cfgNameStr[0] = '\0';
+    return -1;
+  }
+  envNameStr += 5;
+  if (*envNameStr != '\0') {
+    *p = toLowChar(*envNameStr);
+    p++;
+    envNameStr++;
+  }
+
+  for (size_t i = 1; i < cfgNameMaxLen && *envNameStr != '\0'; i++) {
+    if (*envNameStr == '_') {
+      envNameStr++;
+      *p = *envNameStr;
+      if (*envNameStr == '\0') break;
+    } else {
+      *p = toLowChar(*envNameStr);
+    }
+    p++;
+    envNameStr++;
+  }
+
+  *p = '\0';
+  return strlen(cfgNameStr);
+}
+
+int32_t taosEnvToCfg(const char *envStr, char *cfgStr) {
+  if (envStr == NULL && cfgStr == NULL) {
+    return -1;
+  }
+  if (cfgStr != envStr) strcpy(cfgStr, envStr);
+  char *p = strchr(cfgStr, '=');
+  
+  if (p != NULL) {
+    char buf[CFG_NAME_MAX_LEN];
+    if (*(p+1) == '\'') {
+      *(p+1)= ' ';
+      char *pEnd = &cfgStr[strlen(cfgStr)-1];
+      if (*pEnd == '\'') *pEnd = '\0';
+    }
+    *p = '\0';
+    int32_t cfgNameLen = taosEnvNameToCfgName(cfgStr, buf, CFG_NAME_MAX_LEN);
+    if (cfgNameLen > 0) {
+      memcpy(cfgStr, buf, cfgNameLen);
+      memset(&cfgStr[cfgNameLen], ' ', p - cfgStr - cfgNameLen + 1);
+    }
+  }
+  return strlen(cfgStr);
+}
\ No newline at end of file

source/util/src/tjson.c

@@ -144,6 +144,22 @@ char* tjsonToUnformattedString(const SJson* pJson) { return cJSON_PrintUnformatt
 
 SJson* tjsonGetObjectItem(const SJson* pJson, const char* pName) { return cJSON_GetObjectItem(pJson, pName); }
 
+int32_t tjsonGetObjectName(const SJson* pJson, char** pName) {
+  *pName = ((cJSON*)pJson)->string;
+  if (NULL == *pName) {
+    return TSDB_CODE_FAILED;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t tjsonGetObjectValueString(const SJson* pJson, char** pValueString) {
+  *pValueString = ((cJSON*)pJson)->valuestring;
+  if (NULL == *pValueString) {
+    return TSDB_CODE_FAILED;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 int32_t tjsonGetStringValue(const SJson* pJson, const char* pName, char* pVal) {
   char* p = cJSON_GetStringValue(tjsonGetObjectItem((cJSON*)pJson, pName));
   if (NULL == p) {
@@ -310,3 +326,5 @@ bool tjsonValidateJson(const char *jIn) {
   }
   return true;
 }
+
+const char* tjsonGetError() { return cJSON_GetErrorPtr(); }
\ No newline at end of file

source/util/test/CMakeLists.txt

@@ -31,7 +31,7 @@ ENDIF()
 #    TARGET_LINK_LIBRARIES(trefTest util common)
 #ENDIF ()
 
-INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/util/inc)
+INCLUDE_DIRECTORIES(${TD_SOURCE_DIR}/src/util/inc)
 
 # freelistTest
 add_executable(freelistTest "")

source/util/test/cfgTest.cpp

@@ -32,6 +32,7 @@ TEST_F(CfgTest, 01_Str) {
   EXPECT_STREQ(cfgStypeStr(CFG_STYPE_CFG_FILE), "cfg_file");
   EXPECT_STREQ(cfgStypeStr(CFG_STYPE_ENV_FILE), "env_file");
   EXPECT_STREQ(cfgStypeStr(CFG_STYPE_ENV_VAR), "env_var");
+  EXPECT_STREQ(cfgStypeStr(CFG_STYPE_ENV_CMD), "env_cmd");
   EXPECT_STREQ(cfgStypeStr(CFG_STYPE_APOLLO_URL), "apollo_url");
   EXPECT_STREQ(cfgStypeStr(CFG_STYPE_ARG_LIST), "arg_list");
   EXPECT_STREQ(cfgStypeStr(ECfgSrcType(1024)), "invalid");

tests/script/sh/massiveTable/compileVersion.sh

@@ -58,6 +58,7 @@ function compileTDengineVersion() {
     cd ${debugDir}
     cmake ..
     make -j24   
+    make install
 }
 ########################################################################################
 ###############################  main process ##########################################
@@ -67,20 +68,5 @@ cd ${projectDir}
 gitPullBranchInfo $TDengineBrVer
 compileTDengineVersion
 
-taos_dir=${projectDir}/debug/build/bin
-taosd_dir=${projectDir}/debug/build/bin
-exec_process_dir=${projectDir}/debug/build/bin
-
-rm -f /usr/bin/taos
-rm -f /usr/bin/taosd
-rm -f /usr/bin/create_table
-rm -f /usr/bin/tmq_demo
-rm -f /usr/bin/tmq_sim
-
-ln -s $taos_dir/taos /usr/bin/taos
-ln -s $taosd_dir/taosd /usr/bin/taosd
-ln -s $exec_process_dir/create_table /usr/bin/create_table
-ln -s $exec_process_dir/tmq_demo /usr/bin/tmq_demo
-ln -s $exec_process_dir/tmq_sim /usr/bin/tmq_sim
 
 

tests/tsim/src/simSystem.c

@@ -28,8 +28,8 @@ char     simScriptDir[PATH_MAX] = {0};
 extern bool simExecSuccess;
 
 int32_t simInitCfg() {
-  taosCreateLog("simlog", 1, configDir, NULL, NULL, NULL, 1);
-  taosInitCfg(configDir, NULL, NULL, NULL, 1);
+  taosCreateLog("simlog", 1, configDir, NULL, NULL, NULL, NULL, 1);
+  taosInitCfg(configDir, NULL, NULL, NULL, NULL, 1);
 
   SConfig *pCfg = taosGetCfg();
   simDebugFlag = cfgGetItem(pCfg, "simDebugFlag")->i32;

tools/shell/CMakeLists.txt

@@ -3,7 +3,7 @@ aux_source_directory(src SHELL_SRC)
 add_executable(shell ${SHELL_SRC})
 target_link_libraries(
     shell
-    PUBLIC taos_static
+    PUBLIC taos
     PRIVATE os common transport util
 )
 target_include_directories(