Merge branch '3.0' of https://github.com/taosdata/TDengine into feature/data_format

include/common/tmsg.h

@@ -252,24 +252,31 @@ STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
 int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq* pReq, STSchema* pSchema);
 
 typedef struct {
-  int32_t index;  // index of failed block in submit blocks
-  int32_t vnode;  // vnode index of failed block
-  int32_t sid;    // table index of failed block
-  int32_t code;   // errorcode while write data to vnode, such as not created, dropped, no space, invalid table
-} SSubmitRspBlock;
-
-typedef struct {
-  int32_t         code;          // 0-success, > 0 error code
-  int32_t         numOfRows;     // number of records the client is trying to write
-  int32_t         affectedRows;  // number of records actually written
-  int32_t         failedRows;    // number of failed records (exclude duplicate records)
-  int32_t         numOfFailedBlocks;
-  SSubmitRspBlock failedBlocks[];
+  int8_t  hashMeta;
+  int64_t uid;
+  char*   tblFName;
+  int32_t numOfRows;
+  int32_t affectedRows;
+} SSubmitBlkRsp;
+
+typedef struct {
+  int32_t numOfRows;
+  int32_t affectedRows;
+  int32_t nBlocks;
+  union {
+    SArray*        pArray;
+    SSubmitBlkRsp* pBlocks;
+  };
 } SSubmitRsp;
 
+int32_t tEncodeSSubmitRsp(SEncoder* pEncoder, const SSubmitRsp* pRsp);
+int32_t tDecodeSSubmitRsp(SDecoder* pDecoder, SSubmitRsp* pRsp);
+void tFreeSSubmitRsp(SSubmitRsp *pRsp);
+
 #define COL_SMA_ON  ((int8_t)0x1)
 #define COL_IDX_ON  ((int8_t)0x2)
 #define COL_VAL_SET ((int8_t)0x4)
+
 typedef struct SSchema {
   int8_t   type;
   int8_t   flags;
@@ -476,6 +483,7 @@ typedef struct {
   int32_t  acctId;
   int64_t  clusterId;
   uint32_t connId;
+  int32_t  dnodeNum;
   int8_t   superUser;
   int8_t   connType;
   SEpSet   epSet;

include/common/tvariant.h

@@ -36,12 +36,11 @@ typedef struct SVariant {
   };
 } SVariant;
 
-int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value, bool *issigned);
+int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value);
+int32_t toUInteger(const char *z, int32_t n, int32_t base, uint64_t *value);
 
 bool taosVariantIsValid(SVariant *pVar);
 
-void taosVariantCreate(SVariant *pVar, const char *z, int32_t n, int32_t type);
-
 void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type);
 
 void taosVariantDestroy(SVariant *pV);
@@ -59,7 +58,7 @@ int32_t taosVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool
 #endif
 
 int32_t taosVariantTypeSetType(SVariant *pVariant, char type);
-char * taosVariantGet(SVariant *pVar, int32_t type);
+char   *taosVariantGet(SVariant *pVar, int32_t type);
 
 #ifdef __cplusplus
 }

include/libs/parser/parser.h

@@ -26,8 +26,7 @@ extern "C" {
 typedef struct SStmtCallback {
   TAOS_STMT* pStmt;
   int32_t (*getTbNameFn)(TAOS_STMT*, char**);
-  int32_t (*setBindInfoFn)(TAOS_STMT*, STableMeta*, void*);
-  int32_t (*setExecInfoFn)(TAOS_STMT*, SHashObj*, SHashObj*);
+  int32_t (*setInfoFn)(TAOS_STMT*, STableMeta*, void*, char*, bool, SHashObj*, SHashObj*);
   int32_t (*getExecInfoFn)(TAOS_STMT*, SHashObj**, SHashObj**);
 } SStmtCallback;
 
@@ -59,8 +58,9 @@ int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash
 int32_t qResetStmtDataBlock(void* block, bool keepBuf);
 int32_t qCloneStmtDataBlock(void** pDst, void* pSrc);
 void    qFreeStmtDataBlock(void* pDataBlock);
-int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc);
+int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t vgId);
 void    qDestroyStmtDataBlock(void* pBlock);
+STableMeta *qGetTableMetaInDataBlock(void* pDataBlock);
 
 int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
 int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,

include/libs/planner/planner.h

@@ -48,7 +48,8 @@ int32_t qCreateQueryPlan(SPlanContext* pCxt, SQueryPlan** pPlan, SArray* pExecNo
 // @pSource one execution location of this group of datasource subplans
 int32_t qSetSubplanExecutionNode(SSubplan* pSubplan, int32_t groupId, SDownstreamSourceNode* pSource);
 
-int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId);
+int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId,
+                       bool* pEmptyResult);
 
 // Convert to subplan to string for the scheduler to send to the executor
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen);

include/libs/scheduler/scheduler.h

@@ -56,6 +56,7 @@ typedef struct SQueryResult {
   uint64_t        numOfRows;
   int32_t         msgSize;
   char           *msg;
+  void           *res;
 } SQueryResult;
 
 typedef struct STaskInfo {
@@ -71,7 +72,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg);
  * @param nodeList  Qnode/Vnode address list, element is SQueryNodeAddr
  * @return
  */
-int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, int64_t *pJob, const char *sql, int64_t startTs, SQueryResult *pRes);
+int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, int64_t *pJob, const char *sql, int64_t startTs, bool needRes, SQueryResult *pRes);
 
 /**
  * Process the query job, generated according to the query physical plan.

source/client/inc/clientInt.h

@@ -307,9 +307,9 @@ int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* v
 // --- mq
 void hbMgrInitMqHbRspHandle();
 
-SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery);
+SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery, void** res);
 int32_t      getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList);
-int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
+int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList, void** res);
 int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest);
 
 #ifdef __cplusplus

source/client/inc/clientStmt.h

@@ -55,6 +55,7 @@ typedef struct SStmtQueryResInfo {
 
 typedef struct SStmtBindInfo {
   bool         needParse;
+  bool         inExecCache;
   uint64_t     tbUid;
   uint64_t     tbSuid;
   int32_t      sBindRowNum;
@@ -62,7 +63,9 @@ typedef struct SStmtBindInfo {
   int8_t       tbType;
   bool         tagsCached;
   void*        boundTags;  
-  char*        tbName;
+  char         tbName[TSDB_TABLE_FNAME_LEN];;
+  char         tbFName[TSDB_TABLE_FNAME_LEN];
+  char         stbFName[TSDB_TABLE_FNAME_LEN];
   SName        sname;
 } SStmtBindInfo;
 
@@ -71,12 +74,12 @@ typedef struct SStmtExecInfo {
   SRequestObj* pRequest;
   SHashObj*    pVgHash;
   SHashObj*    pBlockHash;
+  bool         autoCreateTbl;
 } SStmtExecInfo;
 
 typedef struct SStmtSQLInfo {
   STMT_TYPE         type;
   STMT_STATUS       status;
-  bool              autoCreate;
   uint64_t          runTimes;
   SHashObj*         pTableCache;   //SHash<SStmtTableCache>
   SQuery*           pQuery;
@@ -85,6 +88,7 @@ typedef struct SStmtSQLInfo {
   SArray*           nodeList;
   SQueryPlan*       pQueryPlan;
   SStmtQueryResInfo queryRes;
+  bool              autoCreateTbl;
 } SStmtSQLInfo;
 
 typedef struct STscStmt {

source/client/src/clientImpl.c

@@ -286,12 +286,12 @@ void setResPrecision(SReqResultInfo* pResInfo, int32_t precision) {
   pResInfo->precision = precision;
 }
 
-int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList) {
+int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList, void** pRes) {
   void* pTransporter = pRequest->pTscObj->pAppInfo->pTransporter;
 
   SQueryResult res = {.code = 0, .numOfRows = 0, .msgSize = ERROR_MSG_BUF_DEFAULT_SIZE, .msg = pRequest->msgBuf};
   int32_t      code = schedulerExecJob(pTransporter, pNodeList, pDag, &pRequest->body.queryJob, pRequest->sqlstr,
-                                       pRequest->metric.start, &res);
+                                       pRequest->metric.start, NULL != pRes, &res);
   if (code != TSDB_CODE_SUCCESS) {
     if (pRequest->body.queryJob != 0) {
       schedulerFreeJob(pRequest->body.queryJob);
@@ -310,6 +310,10 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
     }
   }
 
+  if (pRes) {
+    *pRes = res.res;
+  }
+
   pRequest->code = res.code;
   terrno = res.code;
   return pRequest->code;
@@ -320,7 +324,7 @@ int32_t getQueryPlan(SRequestObj* pRequest, SQuery* pQuery, SArray** pNodeList)
   return getPlan(pRequest, pQuery, &pRequest->body.pDag, *pNodeList);
 }
 
-SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery) {
+SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code, bool keepQuery, void** res) {
   if (TSDB_CODE_SUCCESS == code) {
     switch (pQuery->execMode) {
       case QUERY_EXEC_MODE_LOCAL:
@@ -333,7 +337,7 @@ SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, int32_t code
         SArray* pNodeList = taosArrayInit(4, sizeof(struct SQueryNodeAddr));
         code = getPlan(pRequest, pQuery, &pRequest->body.pDag, pNodeList);
         if (TSDB_CODE_SUCCESS == code) {
-          code = scheduleQuery(pRequest, pRequest->body.pDag, pNodeList);
+          code = scheduleQuery(pRequest, pRequest->body.pDag, pNodeList, res);
         }
         taosArrayDestroy(pNodeList);
         break;
@@ -373,7 +377,7 @@ SRequestObj* launchQuery(STscObj* pTscObj, const char* sql, int sqlLen) {
     return pRequest;
   }
 
-  return launchQueryImpl(pRequest, pQuery, code, false);
+  return launchQueryImpl(pRequest, pQuery, code, false, NULL);
 }
 
 int32_t refreshMeta(STscObj* pTscObj, SRequestObj* pRequest) {

source/client/src/clientMsgHandler.c

@@ -58,7 +58,7 @@ int32_t processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
     return code;
   }
 
-  if (!isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, &connectRsp.epSet)) {
+  if (connectRsp.dnodeNum > 1 && !isEpsetEqual(&pTscObj->pAppInfo->mgmtEp.epSet, &connectRsp.epSet)) {
     updateEpSet_s(&pTscObj->pAppInfo->mgmtEp, &connectRsp.epSet);
   }
 

source/client/src/clientSml.c

@@ -402,7 +402,7 @@ static int32_t smlModifyDBSchemas(SSmlHandle* info) {
     code = catalogGetSTableMeta(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, &pName, &pTableMeta);
 
     if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_INVALID_STB) {
-      SSchemaAction schemaAction = {.action = SCHEMA_ACTION_CREATE_STABLE};
+      SSchemaAction schemaAction = { SCHEMA_ACTION_CREATE_STABLE, 0};
       memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
       schemaAction.createSTable.tags = sTableData->tags;
       schemaAction.createSTable.fields = sTableData->cols;
@@ -1505,8 +1505,8 @@ static int32_t smlParseLine(SSmlHandle* info, const char* sql) {
 
     tinfo->sTableName = elements.measure;
     tinfo->sTableNameLen = elements.measureLen;
-    RandTableName rName = {.tags=tinfo->tags, .sTableName=tinfo->sTableName, .sTableNameLen=tinfo->sTableNameLen,
-                            .childTableName=tinfo->childTableName};
+    RandTableName rName = { tinfo->tags, tinfo->sTableName, tinfo->sTableNameLen,
+                            tinfo->childTableName, 0 };
     buildChildTableName(&rName);
     tinfo->uid = rName.uid;
 
@@ -1660,7 +1660,7 @@ static int32_t smlInsertData(SSmlHandle* info) {
   smlBuildOutput(info->exec, info->pVgHash);
   info->cost.insertRpcTime = taosGetTimestampUs();
 
-  launchQueryImpl(info->pRequest, info->pQuery, TSDB_CODE_SUCCESS, true);
+  launchQueryImpl(info->pRequest, info->pQuery, TSDB_CODE_SUCCESS, true, NULL);
 
   info->affectedRows = taos_affected_rows(info->pRequest);
   return info->pRequest->code;

source/common/src/tmsg.c

@@ -2764,6 +2764,7 @@ int32_t tSerializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
   if (tEncodeI32(&encoder, pRsp->acctId) < 0) return -1;
   if (tEncodeI64(&encoder, pRsp->clusterId) < 0) return -1;
   if (tEncodeU32(&encoder, pRsp->connId) < 0) return -1;
+  if (tEncodeI32(&encoder, pRsp->dnodeNum) < 0) return -1;
   if (tEncodeI8(&encoder, pRsp->superUser) < 0) return -1;
   if (tEncodeI8(&encoder, pRsp->connType) < 0) return -1;
   if (tEncodeSEpSet(&encoder, &pRsp->epSet) < 0) return -1;
@@ -2783,6 +2784,7 @@ int32_t tDeserializeSConnectRsp(void *buf, int32_t bufLen, SConnectRsp *pRsp) {
   if (tDecodeI32(&decoder, &pRsp->acctId) < 0) return -1;
   if (tDecodeI64(&decoder, &pRsp->clusterId) < 0) return -1;
   if (tDecodeU32(&decoder, &pRsp->connId) < 0) return -1;
+  if (tDecodeI32(&decoder, &pRsp->dnodeNum) < 0) return -1;
   if (tDecodeI8(&decoder, &pRsp->superUser) < 0) return -1;
   if (tDecodeI8(&decoder, &pRsp->connType) < 0) return -1;
   if (tDecodeSEpSet(&decoder, &pRsp->epSet) < 0) return -1;
@@ -4026,3 +4028,84 @@ int32_t tDecodeSVSubmitReq(SDecoder *pCoder, SVSubmitReq *pReq) {
   tEndDecode(pCoder);
   return 0;
 }
+
+static int32_t tEncodeSSubmitBlkRsp(SEncoder *pEncoder, const SSubmitBlkRsp *pBlock) {
+  if (tStartEncode(pEncoder) < 0) return -1;
+
+  if (tEncodeI8(pEncoder, pBlock->hashMeta) < 0) return -1;
+  if (pBlock->hashMeta) {
+    if (tEncodeI64(pEncoder, pBlock->uid) < 0) return -1;
+    if (tEncodeCStr(pEncoder, pBlock->tblFName) < 0) return -1;
+  }
+  if (tEncodeI32v(pEncoder, pBlock->numOfRows) < 0) return -1;
+  if (tEncodeI32v(pEncoder, pBlock->affectedRows) < 0) return -1;
+
+  tEndEncode(pEncoder);
+  return 0;
+}
+
+static int32_t tDecodeSSubmitBlkRsp(SDecoder *pDecoder, SSubmitBlkRsp *pBlock) {
+  if (tStartDecode(pDecoder) < 0) return -1;
+
+  if (tDecodeI8(pDecoder, &pBlock->hashMeta) < 0) return -1;
+  if (pBlock->hashMeta) {
+    if (tDecodeI64(pDecoder, &pBlock->uid) < 0) return -1;
+    pBlock->tblFName= taosMemoryCalloc(TSDB_TABLE_FNAME_LEN, 1);
+    if (NULL == pBlock->tblFName) return -1;
+    if (tDecodeCStrTo(pDecoder, pBlock->tblFName) < 0) return -1;
+  }
+  if (tDecodeI32v(pDecoder, &pBlock->numOfRows) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pBlock->affectedRows) < 0) return -1;
+
+  tEndDecode(pDecoder);
+  return 0;
+}
+
+int32_t tEncodeSSubmitRsp(SEncoder *pEncoder, const SSubmitRsp *pRsp) {
+  int32_t nBlocks = taosArrayGetSize(pRsp->pArray);
+
+  if (tStartEncode(pEncoder) < 0) return -1;
+
+  if (tEncodeI32v(pEncoder, pRsp->numOfRows) < 0) return -1;
+  if (tEncodeI32v(pEncoder, pRsp->affectedRows) < 0) return -1;
+  if (tEncodeI32v(pEncoder, nBlocks) < 0) return -1;
+  for (int32_t iBlock = 0; iBlock < nBlocks; iBlock++) {
+    if (tEncodeSSubmitBlkRsp(pEncoder, (SSubmitBlkRsp *)taosArrayGet(pRsp->pArray, iBlock)) < 0) return -1;
+  }
+
+  tEndEncode(pEncoder);
+  return 0;
+}
+
+int32_t tDecodeSSubmitRsp(SDecoder *pDecoder, SSubmitRsp *pRsp) {
+  if (tStartDecode(pDecoder) < 0) return -1;
+
+  if (tDecodeI32v(pDecoder, &pRsp->numOfRows) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pRsp->affectedRows) < 0) return -1;
+  if (tDecodeI32v(pDecoder, &pRsp->nBlocks) < 0) return -1;
+  pRsp->pBlocks = taosMemoryCalloc(pRsp->nBlocks, sizeof(*pRsp->pBlocks));
+  if (pRsp->pBlocks == NULL) return -1;
+  for (int32_t iBlock = 0; iBlock < pRsp->nBlocks; iBlock++) {
+    if (tDecodeSSubmitBlkRsp(pDecoder, pRsp->pBlocks + iBlock) < 0) return -1;
+  }
+
+  tEndDecode(pDecoder);  
+  tDecoderClear(pDecoder);
+  return 0;
+}
+
+void tFreeSSubmitRsp(SSubmitRsp *pRsp) {
+  if (NULL == pRsp) return;
+
+  if (pRsp->pBlocks) {
+    for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
+      SSubmitBlkRsp *sRsp = pRsp->pBlocks + i;
+      taosMemoryFree(sRsp->tblFName);
+    }
+
+    taosMemoryFree(pRsp->pBlocks);
+  }
+
+  taosMemoryFree(pRsp);
+}
+

source/common/src/trow.c

@@ -1159,7 +1159,7 @@ bool tdGetKvRowValOfColEx(STSRowIter *pIter, col_id_t colId, col_type_t colType,
 
 #ifdef TD_SUPPORT_BITMAP
   int16_t colIdx = -1;
-  if (pKvIdx) colIdx = POINTER_DISTANCE(TD_ROW_COL_IDX(pRow), pKvIdx) / sizeof(SKvRowIdx);
+  if (pKvIdx) colIdx = POINTER_DISTANCE(pKvIdx, TD_ROW_COL_IDX(pRow)) / sizeof(SKvRowIdx);
   if (tdGetBitmapValType(pIter->pBitmap, colIdx, &pVal->valType, 0) != TSDB_CODE_SUCCESS) {
     pVal->valType = TD_VTYPE_NONE;
   }
@@ -1226,7 +1226,7 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
                                                compareKvRowColId, TD_EQ);
 #ifdef TD_SUPPORT_BITMAP
     if (pIdx) {
-      colIdx = POINTER_DISTANCE(TD_ROW_COL_IDX(pRow), pIdx) / sizeof(SKvRowIdx);
+      colIdx = POINTER_DISTANCE(pIdx, TD_ROW_COL_IDX(pRow)) / sizeof(SKvRowIdx);
     }
 #endif
     tdGetKvRowValOfCol(pVal, pRow, pIter->pBitmap, pIdx ? pIdx->offset : -1, colIdx);

source/common/src/tvariant.c

@@ -35,104 +35,36 @@
     assert(0);                                                      \
   } while (0)
 
-int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value, bool *isSigned) {
+int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value) {
   errno = 0;
   char *endPtr = NULL;
 
-  int32_t index = 0;
-
-  bool specifiedSign = (z[0] == '+' || z[0] == '-');
-  if (specifiedSign) {
-    *isSigned = true;
-    index = 1;
-  }
-
-  uint64_t val = strtoull(&z[index], &endPtr, base);
-  if (errno == ERANGE || errno == EINVAL) {
+  *value = strtoll(z, &endPtr, base);
+  if (errno == ERANGE || errno == EINVAL || endPtr - z != n) {
     errno = 0;
     return -1;
   }
 
-  if (specifiedSign && val > INT64_MAX) {
-    return -1;
-  }
-
-  if (endPtr - &z[index] != n - index) {
-    return -1;
-  }
-
-  *isSigned = specifiedSign || (val <= INT64_MAX);
-  if (*isSigned) {
-    *value = (z[0] == '-') ? -val : val;
-  } else {
-    *(uint64_t *)value = val;
-  }
-
   return 0;
 }
 
-void taosVariantCreate(SVariant *pVar, const char *z, int32_t n, int32_t type) {
-  int32_t ret = 0;
-  memset(pVar, 0, sizeof(SVariant));
-
-  switch (type) {
-    case TSDB_DATA_TYPE_BOOL: {
-      if (strncasecmp(z, "true", 4) == 0) {
-        pVar->i = TSDB_TRUE;
-      } else if (strncasecmp(z, "false", 5) == 0) {
-        pVar->i = TSDB_FALSE;
-      } else {
-        return;
-      }
-      break;
-    }
-
-    case TSDB_DATA_TYPE_TINYINT:
-    case TSDB_DATA_TYPE_SMALLINT:
-    case TSDB_DATA_TYPE_BIGINT:
-    case TSDB_DATA_TYPE_INT: {
-      bool sign = true;
-
-      int32_t base = 10;
-      if (type == TK_NK_HEX) {
-        base = 16;
-      } else if (type == TK_NK_OCT) {
-        base = 8;
-      } else if (type == TK_NK_BIN) {
-        base = 2;
-      }
-
-      ret = toInteger(z, n, base, &pVar->i, &sign);
-      if (ret != 0) {
-        pVar->nType = -1;  // -1 means error type
-        return;
-      }
+int32_t toUInteger(const char *z, int32_t n, int32_t base, uint64_t *value) {
+  errno = 0;
+  char *endPtr = NULL;
 
-      pVar->nType = (sign) ? TSDB_DATA_TYPE_BIGINT : TSDB_DATA_TYPE_UBIGINT;
-      break;
-    }
-    case TSDB_DATA_TYPE_DOUBLE:
-    case TSDB_DATA_TYPE_FLOAT: {
-      pVar->d = strtod(z, NULL);
-      break;
-    }
-    case TSDB_DATA_TYPE_BINARY: {
-      pVar->pz = strndup(z, n);
-      //pVar->nLen = strRmquote(pVar->pz, n);
-      break;
-    }
-    case TSDB_DATA_TYPE_TIMESTAMP: {
-      assert(0);
-      pVar->i = taosGetTimestamp(TSDB_TIME_PRECISION_NANO);
-      break;
+  const char *p = z;
+  while (*p != 0 && *p == ' ') p++;
+  if (*p != 0 && *p == '-') {
+    return -1;
   }
 
-    default: {  // nType == 0 means the null value
-      type = TSDB_DATA_TYPE_NULL;
-    }
+  *value = strtoull(z, &endPtr, base);
+  if (errno == ERANGE || errno == EINVAL || endPtr - z != n) {
+    errno = 0;
+    return -1;
   }
 
-  pVar->nType = type;
+  return 0;
 }
 
 /**
@@ -483,7 +415,7 @@ static int32_t toNchar(SVariant *pVariant, char **pDest, int32_t *pDestSize) {
   } else {
     int32_t output = 0;
 
-    bool ret = taosMbsToUcs4(pDst, nLen, (TdUcs4*)*pDest, (nLen + 1) * TSDB_NCHAR_SIZE, &output);
+    bool ret = taosMbsToUcs4(pDst, nLen, (TdUcs4 *)*pDest, (nLen + 1) * TSDB_NCHAR_SIZE, &output);
     if (!ret) {
       return -1;
     }
@@ -518,9 +450,9 @@ static FORCE_INLINE int32_t convertToInteger(SVariant *pVariant, int64_t *result
   } else if (IS_UNSIGNED_NUMERIC_TYPE(pVariant->nType)) {
     *result = pVariant->u;
   } else if (IS_FLOAT_TYPE(pVariant->nType)) {
-    *result = (int64_t) pVariant->d;
+    *result = (int64_t)pVariant->d;
   } else {
-    //TODO: handling var types
+    // TODO: handling var types
   }
 #if 0
   errno = 0;
@@ -909,7 +841,7 @@ int32_t tVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool inc
               return -1;
             }
           } else {
-            tasoUcs4Copy((TdUcs4*)payload, pVariant->ucs4, pVariant->nLen);
+            tasoUcs4Copy((TdUcs4 *)payload, pVariant->ucs4, pVariant->nLen);
           }
         }
       } else {
@@ -1026,7 +958,7 @@ int32_t taosVariantTypeSetType(SVariant *pVariant, char type) {
   return 0;
 }
 
-char * taosVariantGet(SVariant *pVar, int32_t type) {
+char *taosVariantGet(SVariant *pVar, int32_t type) {
   switch (type) {
     case TSDB_DATA_TYPE_BOOL:
     case TSDB_DATA_TYPE_TINYINT:

source/common/test/commonTests.cpp

@@ -8,12 +8,11 @@
 #pragma GCC diagnostic ignored "-Wsign-compare"
 #include "os.h"
 
+#include "taos.h"
 #include "tcommon.h"
 #include "tdatablock.h"
-#include "tcommon.h"
-#include "taos.h"
-#include "tvariant.h"
 #include "tdef.h"
+#include "tvariant.h"
 
 namespace {
 //
@@ -29,72 +28,62 @@ TEST(testCase, toInteger_test) {
   uint32_t type = 0;
 
   int64_t val = 0;
-  bool sign = true;
 
-  int32_t ret = toInteger(s, strlen(s), 10, &val, &sign);
+  int32_t ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, 123);
-  ASSERT_EQ(sign, true);
 
   s = "9223372036854775807";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, 9223372036854775807);
-  ASSERT_EQ(sign, true);
 
   s = "9323372036854775807";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, 9323372036854775807u);
-  ASSERT_EQ(sign, false);
 
   s = "-9323372036854775807";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, -1);
 
   s = "-1";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, -1);
-  ASSERT_EQ(sign, true);
 
   s = "-9223372036854775807";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, -9223372036854775807);
-  ASSERT_EQ(sign, true);
 
   s = "1000u";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, -1);
 
   s = "0x10";
-  ret = toInteger(s, strlen(s), 16, &val, &sign);
+  ret = toInteger(s, strlen(s), 16, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, 16);
-  ASSERT_EQ(sign, true);
 
   s = "110";
-  ret = toInteger(s, strlen(s), 2, &val, &sign);
+  ret = toInteger(s, strlen(s), 2, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, 6);
-  ASSERT_EQ(sign, true);
 
   s = "110";
-  ret = toInteger(s, strlen(s), 8, &val, &sign);
+  ret = toInteger(s, strlen(s), 8, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, 72);
-  ASSERT_EQ(sign, true);
 
-  //18446744073709551615  UINT64_MAX
+  // 18446744073709551615  UINT64_MAX
   s = "18446744073709551615";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, 0);
   ASSERT_EQ(val, 18446744073709551615u);
-  ASSERT_EQ(sign, false);
 
   s = "18446744073709551616";
-  ret = toInteger(s, strlen(s), 10, &val, &sign);
+  ret = toInteger(s, strlen(s), 10, &val);
   ASSERT_EQ(ret, -1);
 }
 
@@ -108,8 +97,8 @@ TEST(testCase, Datablock_test) {
   infoData.info.type = TSDB_DATA_TYPE_INT;
   infoData.info.colId = 1;
 
-  infoData.pData = (char*) taosMemoryCalloc(40, infoData.info.bytes);
-  infoData.nullbitmap = (char*) taosMemoryCalloc(1, sizeof(char) * (40/8));
+  infoData.pData = (char*)taosMemoryCalloc(40, infoData.info.bytes);
+  infoData.nullbitmap = (char*)taosMemoryCalloc(1, sizeof(char) * (40 / 8));
   taosArrayPush(b->pDataBlock, &infoData);
 
   SColumnInfoData infoData1 = {0};
@@ -117,36 +106,36 @@ TEST(testCase, Datablock_test) {
   infoData1.info.type = TSDB_DATA_TYPE_BINARY;
   infoData1.info.colId = 2;
 
-  infoData1.varmeta.offset = (int32_t*) taosMemoryCalloc(40, sizeof(uint32_t));
+  infoData1.varmeta.offset = (int32_t*)taosMemoryCalloc(40, sizeof(uint32_t));
   taosArrayPush(b->pDataBlock, &infoData1);
 
   char* str = "the value of: %d";
   char  buf[128] = {0};
   char  varbuf[128] = {0};
 
-  for(int32_t i = 0; i < 40; ++i) {
-    SColumnInfoData* p0 = (SColumnInfoData *) taosArrayGet(b->pDataBlock, 0);
-    SColumnInfoData* p1 = (SColumnInfoData *) taosArrayGet(b->pDataBlock, 1);
+  for (int32_t i = 0; i < 40; ++i) {
+    SColumnInfoData* p0 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 0);
+    SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 1);
 
-    if (i&0x01) {
+    if (i & 0x01) {
       int32_t len = sprintf(buf, str, i);
       STR_TO_VARSTR(varbuf, buf)
-      colDataAppend(p0, i, (const char*) &i, false);
-      colDataAppend(p1, i, (const char*) varbuf, false);
+      colDataAppend(p0, i, (const char*)&i, false);
+      colDataAppend(p1, i, (const char*)varbuf, false);
 
       memset(varbuf, 0, sizeof(varbuf));
       memset(buf, 0, sizeof(buf));
     } else {
-      colDataAppend(p0, i, (const char*) &i, true);
-      colDataAppend(p1, i, (const char*) varbuf, true);
+      colDataAppend(p0, i, (const char*)&i, true);
+      colDataAppend(p1, i, (const char*)varbuf, true);
     }
 
     b->info.rows++;
   }
 
-  SColumnInfoData* p0 = (SColumnInfoData *) taosArrayGet(b->pDataBlock, 0);
-  SColumnInfoData* p1 = (SColumnInfoData *) taosArrayGet(b->pDataBlock, 1);
-  for(int32_t i = 0; i < 40; ++i) {
+  SColumnInfoData* p0 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 0);
+  SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 1);
+  for (int32_t i = 0; i < 40; ++i) {
     if (i & 0x01) {
       ASSERT_EQ(colDataIsNull_f(p0->nullbitmap, i), false);
       ASSERT_EQ(colDataIsNull(p1, b->info.rows, i, nullptr), false);
@@ -158,7 +147,7 @@ TEST(testCase, Datablock_test) {
     }
   }
 
-  printf("binary column length:%d\n", *(int32_t*) p1->pData);
+  printf("binary column length:%d\n", *(int32_t*)p1->pData);
 
   ASSERT_EQ(blockDataGetNumOfCols(b), 2);
   ASSERT_EQ(blockDataGetNumOfRows(b), 40);
@@ -167,7 +156,7 @@ TEST(testCase, Datablock_test) {
   printf("the second row of binary:%s, length:%d\n", (char*)varDataVal(pData), varDataLen(pData));
 
   SArray*         pOrderInfo = taosArrayInit(3, sizeof(SBlockOrderInfo));
-  SBlockOrderInfo order = { true,  TSDB_ORDER_ASC, 0, NULL };
+  SBlockOrderInfo order = {true, TSDB_ORDER_ASC, 0, NULL};
   taosArrayPush(pOrderInfo, &order);
 
   blockDataSort(b, pOrderInfo);
@@ -244,8 +233,8 @@ TEST(testCase, var_dataBlock_split_test) {
   infoData.info.type = TSDB_DATA_TYPE_INT;
   infoData.info.colId = 1;
 
-  infoData.pData = (char*) taosMemoryCalloc(numOfRows, infoData.info.bytes);
-  infoData.nullbitmap = (char*) taosMemoryCalloc(1, sizeof(char) * (numOfRows/8));
+  infoData.pData = (char*)taosMemoryCalloc(numOfRows, infoData.info.bytes);
+  infoData.nullbitmap = (char*)taosMemoryCalloc(1, sizeof(char) * (numOfRows / 8));
   taosArrayPush(b->pDataBlock, &infoData);
 
   SColumnInfoData infoData1 = {0};
@@ -253,13 +242,13 @@ TEST(testCase, var_dataBlock_split_test) {
   infoData1.info.type = TSDB_DATA_TYPE_BINARY;
   infoData1.info.colId = 2;
 
-  infoData1.varmeta.offset = (int32_t*) taosMemoryCalloc(numOfRows, sizeof(uint32_t));
+  infoData1.varmeta.offset = (int32_t*)taosMemoryCalloc(numOfRows, sizeof(uint32_t));
   taosArrayPush(b->pDataBlock, &infoData1);
 
   char buf[41] = {0};
   char buf1[100] = {0};
 
-  for(int32_t i = 0; i < numOfRows; ++i) {
+  for (int32_t i = 0; i < numOfRows; ++i) {
     SColumnInfoData* p0 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 0);
     SColumnInfoData* p1 = (SColumnInfoData*)taosArrayGet(b->pDataBlock, 1);
 
@@ -278,10 +267,10 @@ TEST(testCase, var_dataBlock_split_test) {
 
   int32_t pageSize = 64 * 1024;
 
-  int32_t startIndex= 0;
+  int32_t startIndex = 0;
   int32_t stopIndex = 0;
   int32_t count = 1;
-  while(1) {
+  while (1) {
     blockDataSplitRows(b, true, startIndex, &stopIndex, pageSize);
     printf("the %d split, from: %d to %d\n", count++, startIndex, stopIndex);
 

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

@@ -20,6 +20,7 @@
 #include "mndShow.h"
 #include "mndStb.h"
 #include "mndUser.h"
+#include "mndDnode.h"
 #include "tglobal.h"
 #include "version.h"
 
@@ -227,6 +228,7 @@ static int32_t mndProcessConnectReq(SNodeMsg *pReq) {
   connectRsp.clusterId = pMnode->clusterId;
   connectRsp.connId = pConn->id;
   connectRsp.connType = connReq.connType;
+  connectRsp.dnodeNum = mndGetDnodeSize(pMnode);
 
   snprintf(connectRsp.sVersion, sizeof(connectRsp.sVersion), "ver:%s\nbuild:%s\ngitinfo:%s", version, buildinfo,
            gitinfo);

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

@@ -102,7 +102,7 @@ int32_t      tsdbUpdateSmaWindow(STsdb* pTsdb, SSubmitReq* pMsg, int64_t version
 int32_t      tsdbCreateTSma(STsdb* pTsdb, char* pMsg);
 int32_t      tsdbInsertTSmaData(STsdb* pTsdb, int64_t indexUid, const char* msg);
 int          tsdbInsertData(STsdb* pTsdb, int64_t version, SSubmitReq* pMsg, SSubmitRsp* pRsp);
-int          tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, int32_t* pAffectedRows);
+int          tsdbInsertTableData(STsdb* pTsdb, SSubmitMsgIter* pMsgIter, SSubmitBlk* pBlock, SSubmitBlkRsp* pRsp);
 tsdbReaderT* tsdbQueryTables(SVnode* pVnode, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,
                              uint64_t taskId);
 tsdbReaderT  tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableGroupInfo* groupList, uint64_t qId,

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

@@ -264,6 +264,8 @@ STSchema *metaGetTbTSchema(SMeta *pMeta, tb_uid_t uid, int32_t sver) {
   metaReaderClear(&mr);
 
   pSW = metaGetTableSchema(pMeta, quid, sver, 0);
+  if (!pSW) return NULL;
+  
   tdInitTSchemaBuilder(&sb, 0);
   for (int i = 0; i < pSW->nCols; i++) {
     pSchema = pSW->pSchema + i;

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

@@ -382,6 +382,7 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
     } else {
       break;
     }
+
     if (pIter && pIter->pTable && (!pIdx || (pIter->pTable->uid <= pIdx->uid))) {
       if (tsdbCommitToTable(pCommith, mIter) < 0) {
         tsdbCloseCommitFile(pCommith, true);
@@ -394,6 +395,9 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
         ++fIter;
       }
       ++mIter;
+    } else if (pIter && !pIter->pTable) {
+      // When table already dropped during commit, pIter is not NULL but pIter->pTable is NULL.
+      ++mIter; // skip the table and do nothing
     } else if (pIdx) {
       if (tsdbMoveBlkIdx(pCommith, pIdx) < 0) {
         tsdbCloseCommitFile(pCommith, true);
@@ -439,6 +443,7 @@ static int tsdbCreateCommitIters(SCommitH *pCommith) {
   SCommitIter       *pCommitIter;
   SSkipListNode     *pNode;
   STbData           *pTbData;
+  STSchema          *pTSchema = NULL;
 
   pCommith->niters = SL_SIZE(pMem->pSlIdx);
   pCommith->iters = (SCommitIter *)taosMemoryCalloc(pCommith->niters, sizeof(SCommitIter));
@@ -459,13 +464,17 @@ static int tsdbCreateCommitIters(SCommitH *pCommith) {
     pTbData = (STbData *)pNode->pData;
 
     pCommitIter = pCommith->iters + i;
+    pTSchema = metaGetTbTSchema(REPO_META(pRepo), pTbData->uid, 0);  // TODO: schema version
+
+    if (pTSchema) {
       pCommitIter->pIter = tSkipListCreateIter(pTbData->pData);
       tSkipListIterNext(pCommitIter->pIter);
 
       pCommitIter->pTable = (STable *)taosMemoryMalloc(sizeof(STable));
       pCommitIter->pTable->uid = pTbData->uid;
       pCommitIter->pTable->tid = pTbData->uid;
-    pCommitIter->pTable->pSchema = metaGetTbTSchema(REPO_META(pRepo), pTbData->uid, 0);
+      pCommitIter->pTable->pSchema = pTSchema;  // metaGetTbTSchema(REPO_META(pRepo), pTbData->uid, 0);
+    }
   }
 
   return 0;

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

@@ -290,7 +290,7 @@ int tsdbLoadDataFromCache(STable *pTable, SSkipListIterator *pIter, TSKEY maxKey
   return 0;
 }
 
-int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, int32_t *pAffectedRows) {
+int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlock, SSubmitBlkRsp *pRsp) {
   SSubmitBlkIter blkIter = {0};
   STsdbMemTable *pMemTable = pTsdb->mem;
   void          *tptr;
@@ -344,7 +344,8 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
   if (pMemTable->keyMin > keyMin) pMemTable->keyMin = keyMin;
   if (pMemTable->keyMax < keyMax) pMemTable->keyMax = keyMax;
 
-  (*pAffectedRows) = pMsgIter->numOfRows;
+  pRsp->numOfRows = pMsgIter->numOfRows;
+  pRsp->affectedRows = pMsgIter->numOfRows;
 
   return 0;
 }

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

@@ -38,7 +38,7 @@ struct SMemTable {
 
 struct SMemSkipListNode {
   int8_t            level;
-  SMemSkipListNode *forwards[];
+  SMemSkipListNode *forwards[1]; // Windows does not allow 0 
 };
 
 struct SMemSkipList {
@@ -46,7 +46,7 @@ struct SMemSkipList {
   int8_t           maxLevel;
   int8_t           level;
   int32_t          size;
-  SMemSkipListNode pHead[];
+  SMemSkipListNode pHead[1]; // Windows does not allow 0 
 };
 
 struct SMemData {

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

@@ -36,9 +36,10 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *
   // loop to insert
   tInitSubmitMsgIter(pMsg, &msgIter);
   while (true) {
+    SSubmitBlkRsp r = {0};
     tGetSubmitMsgNext(&msgIter, &pBlock);
     if (pBlock == NULL) break;
-    if (tsdbInsertTableData(pTsdb, &msgIter, pBlock, &affectedrows) < 0) {
+    if (tsdbInsertTableData(pTsdb, &msgIter, pBlock, &r) < 0) {
       return -1;
     }
 
@@ -46,8 +47,8 @@ int tsdbInsertData(STsdb *pTsdb, int64_t version, SSubmitReq *pMsg, SSubmitRsp *
   }
 
   if (pRsp != NULL) {
-    pRsp->affectedRows = affectedrows;
-    pRsp->numOfRows = numOfRows;
+    // pRsp->affectedRows = affectedrows;
+    // pRsp->numOfRows = numOfRows;
   }
 
   return 0;

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

@@ -119,7 +119,7 @@ _exit:
   taosMemoryFree(metaRsp.pSchemas);
   metaReaderClear(&mer2);
   metaReaderClear(&mer1);
-  return code;
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {

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

@@ -392,7 +392,7 @@ static int vnodeProcessCreateTbReq(SVnode *pVnode, int64_t version, void *pReq,
   tEncoderClear(&encoder);
 
 _exit:
-  taosArrayClear(rsp.pArray);
+  taosArrayDestroy(rsp.pArray);
   tDecoderClear(&decoder);
   tEncoderClear(&encoder);
   return rcode;
@@ -454,6 +454,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
   SVDropTbBatchReq req = {0};
   SVDropTbBatchRsp rsp = {0};
   SDecoder         decoder = {0};
+  SEncoder         encoder = {0};
   int              ret;
 
   pRsp->msgType = TDMT_VND_DROP_TABLE_RSP;
@@ -471,7 +472,7 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
   }
 
   // process req
-  rsp.pArray = taosArrayInit(sizeof(SVDropTbRsp), req.nReqs);
+  rsp.pArray = taosArrayInit(req.nReqs, sizeof(SVDropTbRsp));
   for (int iReq = 0; iReq < req.nReqs; iReq++) {
     SVDropTbReq *pDropTbReq = req.pReqs + iReq;
     SVDropTbRsp  dropTbRsp = {0};
@@ -493,59 +494,74 @@ static int vnodeProcessDropTbReq(SVnode *pVnode, int64_t version, void *pReq, in
 
 _exit:
   tDecoderClear(&decoder);
-  // encode rsp (TODO)
+  tEncodeSize(tEncodeSVDropTbBatchRsp, &rsp, pRsp->contLen, ret);
+  pRsp->pCont = rpcMallocCont(pRsp->contLen);
+  tEncoderInit(&encoder, pRsp->pCont, pRsp->contLen);
+  tEncodeSVDropTbBatchRsp(&encoder, &rsp);
+  tEncoderClear(&encoder);
   return 0;
 }
 
-static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char* tags) {
-  ASSERT(pMsg != NULL);
-  SSubmitMsgIter msgIter = {0};
-  SMeta         *pMeta = pVnode->pMeta;
-  SSubmitBlk    *pBlock = NULL;
+static int vnodeDebugPrintSingleSubmitMsg(SMeta         *pMeta, SSubmitBlk  *pBlock, SSubmitMsgIter *msgIter, const char *tags) {
   SSubmitBlkIter blkIter = {0};
-  STSRow        *row = NULL;
   STSchema      *pSchema = NULL;
   tb_uid_t       suid = 0;
+  STSRow        *row = NULL;
 
-  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
-  while (true) {
-    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
-    if (pBlock == NULL) break;
-    tInitSubmitBlkIter(&msgIter, pBlock, &blkIter);
-    if (blkIter.row == NULL) continue;
-    if (!pSchema || (suid != msgIter.suid)) {
+  tInitSubmitBlkIter(msgIter, pBlock, &blkIter);
+  if (blkIter.row == NULL) return 0;
+  if (!pSchema || (suid != msgIter->suid)) {
     if (pSchema) {
       taosMemoryFreeClear(pSchema);
     }
-      pSchema = metaGetTbTSchema(pMeta, msgIter.suid, 0);  // TODO: use the real schema
-      if(pSchema) {
-        suid = msgIter.suid;
+    pSchema = metaGetTbTSchema(pMeta, msgIter->suid, 0);  // TODO: use the real schema
+    if (pSchema) {
+      suid = msgIter->suid;
     }
   }
-    if(!pSchema) {
+  if (!pSchema) {
     printf("%s:%d no valid schema\n", tags, __LINE__);
-      continue;
+    return -1;
   }
   char __tags[128] = {0};
-    snprintf(__tags, 128, "%s: uid %" PRIi64 " ", tags, msgIter.uid);
+  snprintf(__tags, 128, "%s: uid %" PRIi64 " ", tags, msgIter->uid);
   while ((row = tGetSubmitBlkNext(&blkIter))) {
     tdSRowPrint(row, pSchema, __tags);
   }
-  }
 
   taosMemoryFreeClear(pSchema);
 
+  return TSDB_CODE_SUCCESS;
+}
+
+static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char *tags) {
+  ASSERT(pMsg != NULL);
+  SSubmitMsgIter msgIter = {0};
+  SMeta         *pMeta = pVnode->pMeta;
+  SSubmitBlk    *pBlock = NULL;
+
+  if (tInitSubmitMsgIter(pMsg, &msgIter) < 0) return -1;
+  while (true) {
+    if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
+    if (pBlock == NULL) break;
+    
+    vnodeDebugPrintSingleSubmitMsg(pMeta, pBlock, &msgIter, tags);
+  }
+
   return 0;
 }
 
 static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, int32_t len, SRpcMsg *pRsp) {
   SSubmitReq    *pSubmitReq = (SSubmitReq *)pReq;
+  SSubmitRsp     submitRsp = {0};
   SSubmitMsgIter msgIter = {0};
   SSubmitBlk    *pBlock;
   SSubmitRsp     rsp = {0};
   SVCreateTbReq  createTbReq = {0};
   SDecoder       decoder = {0};
   int32_t        nRows;
+  int32_t        tsize, ret;
+  SEncoder       encoder = {0};
 
   pRsp->code = 0;
 
@@ -559,12 +575,17 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
     goto _exit;
   }
 
-  for (;;) {
+  submitRsp.pArray = taosArrayInit(pSubmitReq->numOfBlocks, sizeof(SSubmitBlkRsp));
+  for (int i = 0;;) {
     tGetSubmitMsgNext(&msgIter, &pBlock);
     if (pBlock == NULL) break;
 
+    SSubmitBlkRsp submitBlkRsp = {0};
+
     // create table for auto create table mode
     if (msgIter.schemaLen > 0) {
+      submitBlkRsp.hashMeta = 1;
+
       tDecoderInit(&decoder, pBlock->data, msgIter.schemaLen);
       if (tDecodeSVCreateTbReq(&decoder, &createTbReq) < 0) {
         pRsp->code = TSDB_CODE_INVALID_MSG;
@@ -580,6 +601,10 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
         }
       }
 
+      submitBlkRsp.uid = createTbReq.uid;
+      submitBlkRsp.tblFName = taosMemoryMalloc(strlen(pVnode->config.dbname) + strlen(createTbReq.name) + 2);
+      sprintf(submitBlkRsp.tblFName, "%s.%s", pVnode->config.dbname, createTbReq.name);
+
       msgIter.uid = createTbReq.uid;
       if (createTbReq.type == TSDB_CHILD_TABLE) {
         msgIter.suid = createTbReq.ctb.suid;
@@ -587,23 +612,33 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
         msgIter.suid = 0;
       }
 
+      vnodeDebugPrintSingleSubmitMsg(pVnode->pMeta, pBlock, &msgIter, "real uid");
       tDecoderClear(&decoder);
     }
 
-    if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &nRows) < 0) {
+    if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &submitBlkRsp) < 0) {
       pRsp->code = terrno;
       goto _exit;
     }
 
-    rsp.affectedRows += nRows;
-    
+    submitRsp.numOfRows += submitBlkRsp.numOfRows;
+    submitRsp.affectedRows += submitBlkRsp.affectedRows;
+    taosArrayPush(submitRsp.pArray, &submitBlkRsp);
   }
 
 _exit:
-  // encode the response (TODO)
-  pRsp->pCont = rpcMallocCont(sizeof(SSubmitRsp));
-  memcpy(pRsp->pCont, &rsp, sizeof(rsp));
-  pRsp->contLen = sizeof(SSubmitRsp);
+  tEncodeSize(tEncodeSSubmitRsp, &submitRsp, tsize, ret);
+  pRsp->pCont = rpcMallocCont(tsize);
+  pRsp->contLen = tsize;
+  tEncoderInit(&encoder, pRsp->pCont, tsize);
+  tEncodeSSubmitRsp(&encoder, &submitRsp);
+  tEncoderClear(&encoder);
+
+  for (int32_t i = 0; i < taosArrayGetSize(submitRsp.pArray); i++) {
+    taosMemoryFree(((SSubmitBlkRsp *)taosArrayGet(submitRsp.pArray, i))[0].tblFName);
+  }
+
+  taosArrayDestroy(submitRsp.pArray);
 
   tsdbTriggerRSma(pVnode->pTsdb, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK);
 

source/libs/executor/src/indexoperator.c

@@ -22,6 +22,8 @@
 typedef struct SIFCtx {
   int32_t   code;
   SHashObj *pRes;    /* element is SScalarParam */
+  bool      noExec;  // true: just iterate condition tree, and add hint to executor plan
+  // SIdxFltStatus st;
 } SIFCtx;
 
 #define SIF_ERR_RET(c)                \
@@ -55,6 +57,7 @@ typedef struct SIFParam {
   SArray *result;
   char *  condValue;
 
+  SIdxFltStatus status;
   uint8_t       colValType;
   col_id_t      colId;
   int64_t       suid;  // add later
@@ -82,6 +85,9 @@ static int32_t sifGetFuncFromSql(EOperatorType src, EIndexQueryType *dst) {
 }
 
 typedef int32_t (*sif_func_t)(SIFParam *left, SIFParam *rigth, SIFParam *output);
+
+static sif_func_t sifNullFunc = NULL;
+// typedef struct SIFWalkParm
 // construct tag filter operator later
 static void destroyTagFilterOperatorInfo(void *param) {
   STagFilterOperatorInfo *pInfo = (STagFilterOperatorInfo *)param;
@@ -114,6 +120,24 @@ static int32_t sifValidateColumn(SColumnNode *cn) {
   return TSDB_CODE_SUCCESS;
 }
 
+static SIdxFltStatus sifMergeCond(ELogicConditionType type, SIdxFltStatus ls, SIdxFltStatus rs) {
+  // enh rule later
+  if (type == LOGIC_COND_TYPE_AND) {
+    if (ls == SFLT_NOT_INDEX || rs == SFLT_NOT_INDEX) {
+      if (ls == SFLT_NOT_INDEX)
+        return rs;
+      else
+        return ls;
+    }
+    return SFLT_COARSE_INDEX;
+  } else if (type == LOGIC_COND_TYPE_OR) {
+    return SFLT_COARSE_INDEX;
+  } else if (type == LOGIC_COND_TYPE_NOT) {
+    return SFLT_NOT_INDEX;
+  }
+  return SFLT_NOT_INDEX;
+}
+
 static int32_t sifGetValueFromNode(SNode *node, char **value) {
   // covert data From snode;
   SValueNode *vn = (SValueNode *)node;
@@ -257,11 +281,11 @@ static int32_t sifDoIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFP
   if (tm == NULL) {
     return TSDB_CODE_QRY_OUT_OF_MEMORY;
   }
-  SIndexMultiTermQuery *mtm = indexMultiTermQueryCreate(MUST);
 
   EIndexQueryType qtype = 0;
   SIF_ERR_RET(sifGetFuncFromSql(operType, &qtype));
 
+  SIndexMultiTermQuery *mtm = indexMultiTermQueryCreate(MUST);
   indexMultiTermQueryAdd(mtm, tm, qtype);
   int ret = indexSearch(NULL, mtm, output->result);
   indexMultiTermQueryDestroy(mtm);
@@ -319,6 +343,7 @@ static int32_t sifNotMatchFunc(SIFParam *left, SIFParam *right, SIFParam *output
   int id = OP_TYPE_NMATCH;
   return sifDoIndex(left, right, id, output);
 }
+
 static int32_t sifDefaultFunc(SIFParam *left, SIFParam *right, SIFParam *output) {
   // add more except
   return TSDB_CODE_QRY_INVALID_INPUT;
@@ -352,9 +377,9 @@ static sif_func_t sifGetOperFn(int32_t funcId) {
     case OP_TYPE_NMATCH:
       return sifNotMatchFunc;
     default:
-      return sifDefaultFunc;
+      return sifNullFunc;
   }
-  return sifDefaultFunc;
+  return sifNullFunc;
 }
 static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
   int32_t code = 0;
@@ -367,6 +392,11 @@ static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) {
   SIF_ERR_RET(sifInitOperParams(&params, node, ctx));
 
   sif_func_t operFn = sifGetOperFn(node->opType);
+  if (ctx->noExec && operFn == NULL) {
+    output->status = SFLT_NOT_INDEX;
+  } else {
+    output->status = SFLT_ACCURATE_INDEX;
+  }
 
   return operFn(&params[0], nParam > 1 ? &params[1] : NULL, output);
 _return:
@@ -385,6 +415,7 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou
   SIFParam *params = NULL;
   SIF_ERR_RET(sifInitParamList(&params, node->pParameterList, ctx));
 
+  if (ctx->noExec == false) {
     for (int32_t m = 0; m < node->pParameterList->length; m++) {
       // add impl later
       if (node->condType == LOGIC_COND_TYPE_AND) {
@@ -395,6 +426,11 @@ static int32_t sifExecLogic(SLogicConditionNode *node, SIFCtx *ctx, SIFParam *ou
         taosArrayAddAll(output->result, params[m].result);
       }
     }
+  } else {
+    for (int32_t m = 0; m < node->pParameterList->length; m++) {
+      output->status = sifMergeCond(node->condType, output->status, params[m].status);
+    }
+  }
 _return:
   taosMemoryFree(params);
   SIF_RET(code);
@@ -486,7 +522,7 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) {
     return TSDB_CODE_QRY_INVALID_INPUT;
   }
   int32_t code = 0;
-  SIFCtx  ctx = {.code = 0};
+  SIFCtx  ctx = {.code = 0, .noExec = false};
   ctx.pRes = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
   if (NULL == ctx.pRes) {
     qError("index-filter failed to taosHashInit");
@@ -510,6 +546,36 @@ static int32_t sifCalculate(SNode *pNode, SIFParam *pDst) {
   SIF_RET(code);
 }
 
+static int32_t sifGetFltHint(SNode *pNode, SIdxFltStatus *status) {
+  int32_t code = TSDB_CODE_SUCCESS;
+  if (pNode == NULL) {
+    return TSDB_CODE_QRY_INVALID_INPUT;
+  }
+
+  SIFCtx ctx = {.code = 0, .noExec = true};
+  ctx.pRes = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
+  if (NULL == ctx.pRes) {
+    qError("index-filter failed to taosHashInit");
+    return TSDB_CODE_QRY_OUT_OF_MEMORY;
+  }
+
+  nodesWalkExprPostOrder(pNode, sifCalcWalker, &ctx);
+
+  SIF_ERR_RET(ctx.code);
+
+  SIFParam *res = (SIFParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES);
+  if (res == NULL) {
+    qError("no valid res in hash, node:(%p), type(%d)", (void *)&pNode, nodeType(pNode));
+    SIF_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
+  }
+  *status = res->status;
+
+  sifFreeParam(res);
+  taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES);
+
+  SIF_RET(code);
+}
+
 int32_t doFilterTag(const SNode *pFilterNode, SArray *result) {
   if (pFilterNode == NULL) {
     return TSDB_CODE_SUCCESS;
@@ -528,10 +594,14 @@ int32_t doFilterTag(const SNode *pFilterNode, SArray *result) {
 }
 
 SIdxFltStatus idxGetFltStatus(SNode *pFilterNode) {
+  SIdxFltStatus st = SFLT_NOT_INDEX;
   if (pFilterNode == NULL) {
     return SFLT_NOT_INDEX;
   }
+  SFilterInfo *filter = NULL;
+  // todo move to the initialization function
+  SIF_ERR_RET(filterInitFromNode((SNode *)pFilterNode, &filter, 0));
 
-  // impl later
-  return SFLT_ACCURATE_INDEX;
+  SIF_ERR_RET(sifGetFltHint((SNode *)pFilterNode, &st));
+  return st;
 }

source/libs/function/src/builtins.c

@@ -239,7 +239,7 @@ static int32_t translateLeastSQR(SFunctionNode* pFunc, char* pErrBuf, int32_t le
     }
   }
 
-  pFunc->node.resType = (SDataType) { .bytes = 64, .type = TSDB_DATA_TYPE_BINARY };
+  pFunc->node.resType = (SDataType){.bytes = 64, .type = TSDB_DATA_TYPE_BINARY};
   return TSDB_CODE_SUCCESS;
 }
 
@@ -1042,26 +1042,6 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
     .sprocessFunc = timezoneFunction,
     .finalizeFunc = NULL
   },
-  {
-    .name = "_rowts",
-    .type = FUNCTION_TYPE_ROWTS,
-    .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
-    .translateFunc = translateTimePseudoColumn,
-    .getEnvFunc   = getTimePseudoFuncEnv,
-    .initFunc     = NULL,
-    .sprocessFunc = NULL,
-    .finalizeFunc = NULL
-  },
-  {
-    .name = "_c0",
-    .type = FUNCTION_TYPE_ROWTS,
-    .classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
-    .translateFunc = translateTimePseudoColumn,
-    .getEnvFunc   = getTimePseudoFuncEnv,
-    .initFunc     = NULL,
-    .sprocessFunc = NULL,
-    .finalizeFunc = NULL
-  },
   {
     .name = "tbname",
     .type = FUNCTION_TYPE_TBNAME,

source/libs/parser/inc/parInsertData.h

@@ -137,7 +137,7 @@ void destroyBlockArrayList(SArray* pDataBlockList);
 void destroyBlockHashmap(SHashObj* pDataBlockHash);
 int  initRowBuilder(SRowBuilder *pBuilder, int16_t schemaVer, SParsedDataColInfo *pColInfo);
 int32_t allocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t * numOfRows);
-int32_t getDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, int32_t startOffset, int32_t rowSize,
+int32_t getDataBlockFromList(SHashObj* pHashList, void* id, int32_t idLen, int32_t size, int32_t startOffset, int32_t rowSize,
      STableMeta* pTableMeta, STableDataBlocks** dataBlocks, SArray* pBlockList, SVCreateTbReq* pCreateTbReq);
 int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** pVgDataBlocks);
 int32_t buildCreateTbMsg(STableDataBlocks* pBlocks, SVCreateTbReq* pCreateTbReq);

source/libs/parser/src/parAstCreater.c

@@ -350,7 +350,18 @@ SNode* createNotBetweenAnd(SAstCreateContext* pCxt, SNode* pExpr, SNode* pLeft,
                                   createOperatorNode(pCxt, OP_TYPE_GREATER_THAN, nodesCloneNode(pExpr), pRight));
 }
 
+static SNode* createPrimaryKeyCol(SAstCreateContext* pCxt) {
+  SColumnNode* pCol = nodesMakeNode(QUERY_NODE_COLUMN);
+  CHECK_OUT_OF_MEM(pCol);
+  pCol->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
+  strcpy(pCol->colName, PK_TS_COL_INTERNAL_NAME);
+  return (SNode*)pCol;
+}
+
 SNode* createFunctionNode(SAstCreateContext* pCxt, const SToken* pFuncName, SNodeList* pParameterList) {
+  if (0 == strncasecmp("_rowts", pFuncName->z, pFuncName->n) || 0 == strncasecmp("_c0", pFuncName->z, pFuncName->n)) {
+    return createPrimaryKeyCol(pCxt);
+  }
   SFunctionNode* func = (SFunctionNode*)nodesMakeNode(QUERY_NODE_FUNCTION);
   CHECK_OUT_OF_MEM(func);
   strncpy(func->functionName, pFuncName->z, pFuncName->n);
@@ -467,13 +478,11 @@ SNode* createSessionWindowNode(SAstCreateContext* pCxt, SNode* pCol, SNode* pGap
 SNode* createStateWindowNode(SAstCreateContext* pCxt, SNode* pExpr) {
   SStateWindowNode* state = (SStateWindowNode*)nodesMakeNode(QUERY_NODE_STATE_WINDOW);
   CHECK_OUT_OF_MEM(state);
-  state->pCol = nodesMakeNode(QUERY_NODE_COLUMN);
+  state->pCol = createPrimaryKeyCol(pCxt);
   if (NULL == state->pCol) {
     nodesDestroyNode(state);
     CHECK_OUT_OF_MEM(state->pCol);
   }
-  ((SColumnNode*)state->pCol)->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
-  strcpy(((SColumnNode*)state->pCol)->colName, PK_TS_COL_INTERNAL_NAME);
   state->pExpr = pExpr;
   return (SNode*)state;
 }
@@ -482,13 +491,11 @@ SNode* createIntervalWindowNode(SAstCreateContext* pCxt, SNode* pInterval, SNode
                                 SNode* pFill) {
   SIntervalWindowNode* interval = (SIntervalWindowNode*)nodesMakeNode(QUERY_NODE_INTERVAL_WINDOW);
   CHECK_OUT_OF_MEM(interval);
-  interval->pCol = nodesMakeNode(QUERY_NODE_COLUMN);
+  interval->pCol = createPrimaryKeyCol(pCxt);
   if (NULL == interval->pCol) {
     nodesDestroyNode(interval);
     CHECK_OUT_OF_MEM(interval->pCol);
   }
-  ((SColumnNode*)interval->pCol)->colId = PRIMARYKEY_TIMESTAMP_COL_ID;
-  strcpy(((SColumnNode*)interval->pCol)->colName, PK_TS_COL_INTERNAL_NAME);
   interval->pInterval = pInterval;
   interval->pOffset = pOffset;
   interval->pSliding = pSliding;
@@ -667,7 +674,7 @@ SNode* setDatabaseOption(SAstCreateContext* pCxt, SNode* pOptions, EDatabaseOpti
     case DB_OPTION_DAYS: {
       SToken* pToken = pVal;
       if (TK_NK_INTEGER == pToken->type) {
-        ((SDatabaseOptions*)pOptions)->daysPerFile = strtol(pToken->z, NULL, 10);
+        ((SDatabaseOptions*)pOptions)->daysPerFile = strtol(pToken->z, NULL, 10) * 1440;
       } else {
         ((SDatabaseOptions*)pOptions)->pDaysPerFile = (SValueNode*)createDurationValueNode(pCxt, pToken);
       }

source/libs/parser/src/parInsertData.c

@@ -185,11 +185,11 @@ int32_t buildCreateTbMsg(STableDataBlocks* pBlocks, SVCreateTbReq* pCreateTbReq)
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t getDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, int32_t startOffset, int32_t rowSize,
+int32_t getDataBlockFromList(SHashObj* pHashList, void* id, int32_t idLen, int32_t size, int32_t startOffset, int32_t rowSize,
                              STableMeta* pTableMeta, STableDataBlocks** dataBlocks, SArray* pBlockList,
                              SVCreateTbReq* pCreateTbReq) {
   *dataBlocks = NULL;
-  STableDataBlocks** t1 = (STableDataBlocks**)taosHashGet(pHashList, (const char*)&id, sizeof(id));
+  STableDataBlocks** t1 = (STableDataBlocks**)taosHashGet(pHashList, (const char*)id, idLen);
   if (t1 != NULL) {
     *dataBlocks = *t1;
   }
@@ -207,7 +207,7 @@ int32_t getDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, int3
       }
     }
 
-    taosHashPut(pHashList, (const char*)&id, sizeof(int64_t), (char*)dataBlocks, POINTER_BYTES);
+    taosHashPut(pHashList, (const char*)id, idLen, (char*)dataBlocks, POINTER_BYTES);
     if (pBlockList) {
       taosArrayPush(pBlockList, dataBlocks);
     }
@@ -445,7 +445,7 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** p
   const int INSERT_HEAD_SIZE = sizeof(SSubmitReq);
   int       code = 0;
   bool      isRawPayload = IS_RAW_PAYLOAD(payloadType);
-  SHashObj* pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
+  SHashObj* pVnodeDataBlockHashList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, false);
   SArray*   pVnodeDataBlockList = taosArrayInit(8, POINTER_BYTES);
 
   STableDataBlocks** p = taosHashIterate(pHashObj, NULL);
@@ -457,7 +457,7 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** p
       STableDataBlocks* dataBuf = NULL;
       pOneTableBlock->pTableMeta->vgId = pOneTableBlock->vgId;    // for schemaless, restore origin vgId
       int32_t           ret =
-          getDataBlockFromList(pVnodeDataBlockHashList, pOneTableBlock->vgId, TSDB_PAYLOAD_SIZE, INSERT_HEAD_SIZE, 0,
+          getDataBlockFromList(pVnodeDataBlockHashList, &pOneTableBlock->vgId, sizeof(pOneTableBlock->vgId), TSDB_PAYLOAD_SIZE, INSERT_HEAD_SIZE, 0,
                                pOneTableBlock->pTableMeta, &dataBuf, pVnodeDataBlockList, NULL);
       if (ret != TSDB_CODE_SUCCESS) {
         taosHashCleanup(pVnodeDataBlockHashList);
@@ -620,7 +620,7 @@ int32_t qCloneStmtDataBlock(void** pDst, void* pSrc) {
   return qResetStmtDataBlock(*pDst, false);
 }
 
-int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc) {
+int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t vgId) {
   int32_t code = qCloneStmtDataBlock(pDst, pSrc);
   if (code) {
     return code;
@@ -633,11 +633,22 @@ int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
 
+  pBlock->vgId = vgId;
+  
+  if (pBlock->pTableMeta) {
+    pBlock->pTableMeta->uid = uid;
+    pBlock->pTableMeta->vgId = vgId;
+  }
+  
   memset(pBlock->pData, 0, sizeof(SSubmitBlk));
 
   return TSDB_CODE_SUCCESS;
 }
 
+STableMeta *qGetTableMetaInDataBlock(void* pDataBlock) {
+  return ((STableDataBlocks*)pDataBlock)->pTableMeta;
+}
+
 void qFreeStmtDataBlock(void* pDataBlock) {
   if (pDataBlock == NULL) {
     return;

source/libs/parser/src/parTranslater.c

@@ -1886,6 +1886,9 @@ static int32_t checkDbKeepOption(STranslateContext* pCxt, SDatabaseOptions* pOpt
         TIME_UNIT_DAY != pVal->unit) {
       return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_KEEP_UNIT, pVal->unit);
     }
+    if (!pVal->isDuration) {
+      pVal->datum.i = pVal->datum.i * 1440;
+    }
   }
 
   pOptions->keep[0] = getBigintFromValueNode((SValueNode*)nodesListGetNode(pOptions->pKeep, 0));

source/libs/parser/test/parInitialCTest.cpp

@@ -3,7 +3,7 @@
  *
  * 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.
+ * 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
@@ -14,6 +14,7 @@
  */
 
 #include "parTestUtil.h"
+#include "ttime.h"
 
 using namespace std;
 
@@ -26,72 +27,226 @@ class ParserInitialCTest : public ParserDdlTest {};
 TEST_F(ParserInitialCTest, createAccount) {
   useDb("root", "test");
 
-  run("create account ac_wxy pass '123456'", TSDB_CODE_PAR_EXPRIE_STATEMENT);
+  run("CREATE ACCOUNT ac_wxy PASS '123456'", TSDB_CODE_PAR_EXPRIE_STATEMENT);
 }
 
 TEST_F(ParserInitialCTest, createBnode) {
   useDb("root", "test");
 
-  run("create bnode on dnode 1");
+  run("CREATE BNODE ON DNODE 1");
 }
 
+/*
+ * CREATE DATABASE [IF NOT EXISTS] db_name [database_options]
+ *
+ * database_options:
+ *     database_option ...
+ *
+ * database_option: {
+ *     BUFFER value
+ *   | CACHELAST value
+ *   | COMP {0 | 1 | 2}
+ *   | DAYS value
+ *   | FSYNC value
+ *   | MAXROWS value
+ *   | MINROWS value
+ *   | KEEP value
+ *   | PAGES value
+ *   | PAGESIZE  value
+ *   | PRECISION {'ms' | 'us' | 'ns'}
+ *   | REPLICA value
+ *   | RETENTIONS ingestion_duration:keep_duration ...
+ *   | STRICT value
+ *   | WAL value
+ *   | VGROUPS value
+ *   | SINGLE_STABLE {0 | 1}
+ * }
+ */
 TEST_F(ParserInitialCTest, createDatabase) {
   useDb("root", "test");
 
-  run("create database wxy_db");
-
-  run("create database if not exists wxy_db "
-      "cachelast 2 "
-      "comp 1 "
-      "days 100 "
-      "fsync 100 "
-      "maxrows 1000 "
-      "minrows 100 "
-      "keep 1440 "
-      "precision 'ms' "
-      "replica 3 "
-      "wal 2 "
-      "vgroups 100 "
-      "single_stable 0 "
-      "retentions 15s:7d,1m:21d,15m:5y");
-
-  run("create database if not exists wxy_db "
-      "days 100m "
-      "keep 1440m,300h,400d ");
+  SCreateDbReq expect = {0};
+
+  auto setCreateDbReqFunc = [&](const char* pDbname, int8_t igExists = 0) {
+    memset(&expect, 0, sizeof(SCreateDbReq));
+    int32_t len = snprintf(expect.db, sizeof(expect.db), "0.%s", pDbname);
+    expect.db[len] = '\0';
+    expect.ignoreExist = igExists;
+    expect.buffer = TSDB_DEFAULT_BUFFER_PER_VNODE;
+    expect.cacheLastRow = TSDB_DEFAULT_CACHE_LAST_ROW;
+    expect.compression = TSDB_DEFAULT_COMP_LEVEL;
+    expect.daysPerFile = TSDB_DEFAULT_DAYS_PER_FILE;
+    expect.fsyncPeriod = TSDB_DEFAULT_FSYNC_PERIOD;
+    expect.maxRows = TSDB_DEFAULT_MAXROWS_FBLOCK;
+    expect.minRows = TSDB_DEFAULT_MINROWS_FBLOCK;
+    expect.daysToKeep0 = TSDB_DEFAULT_KEEP;
+    expect.daysToKeep1 = TSDB_DEFAULT_KEEP;
+    expect.daysToKeep2 = TSDB_DEFAULT_KEEP;
+    expect.pages = TSDB_DEFAULT_PAGES_PER_VNODE;
+    expect.pageSize = TSDB_DEFAULT_PAGESIZE_PER_VNODE;
+    expect.precision = TSDB_DEFAULT_PRECISION;
+    expect.replications = TSDB_DEFAULT_DB_REPLICA;
+    expect.strict = TSDB_DEFAULT_DB_STRICT;
+    expect.walLevel = TSDB_DEFAULT_WAL_LEVEL;
+    expect.numOfVgroups = TSDB_DEFAULT_VN_PER_DB;
+    expect.numOfStables = TSDB_DEFAULT_DB_SINGLE_STABLE;
+  };
+
+  auto setDbBufferFunc = [&](int32_t buffer) { expect.buffer = buffer; };
+  auto setDbCachelastFunc = [&](int8_t CACHELAST) { expect.cacheLastRow = CACHELAST; };
+  auto setDbCompressionFunc = [&](int8_t compressionLevel) { expect.compression = compressionLevel; };
+  auto setDbDaysFunc = [&](int32_t daysPerFile) { expect.daysPerFile = daysPerFile; };
+  auto setDbFsyncFunc = [&](int32_t fsyncPeriod) { expect.fsyncPeriod = fsyncPeriod; };
+  auto setDbMaxRowsFunc = [&](int32_t maxRowsPerBlock) { expect.maxRows = maxRowsPerBlock; };
+  auto setDbMinRowsFunc = [&](int32_t minRowsPerBlock) { expect.minRows = minRowsPerBlock; };
+  auto setDbKeepFunc = [&](int32_t keep0, int32_t keep1 = 0, int32_t keep2 = 0) {
+    expect.daysToKeep0 = keep0;
+    expect.daysToKeep1 = 0 == keep1 ? expect.daysToKeep0 : keep1;
+    expect.daysToKeep2 = 0 == keep2 ? expect.daysToKeep1 : keep2;
+  };
+  auto setDbPagesFunc = [&](int32_t pages) { expect.pages = pages; };
+  auto setDbPageSizeFunc = [&](int32_t pagesize) { expect.pageSize = pagesize; };
+  auto setDbPrecisionFunc = [&](int8_t precision) { expect.precision = precision; };
+  auto setDbReplicaFunc = [&](int8_t replica) { expect.replications = replica; };
+  auto setDbStrictaFunc = [&](int8_t strict) { expect.strict = strict; };
+  auto setDbWalLevelFunc = [&](int8_t walLevel) { expect.walLevel = walLevel; };
+  auto setDbVgroupsFunc = [&](int32_t numOfVgroups) { expect.numOfVgroups = numOfVgroups; };
+  auto setDbSingleStableFunc = [&](int8_t singleStable) { expect.numOfStables = singleStable; };
+  auto addDbRetentionFunc = [&](int64_t freq, int64_t keep, int8_t freqUnit, int8_t keepUnit) {
+    SRetention retention = {0};
+    retention.freq = freq;
+    retention.keep = keep;
+    retention.freqUnit = freqUnit;
+    retention.keepUnit = keepUnit;
+    if (NULL == expect.pRetensions) {
+      expect.pRetensions = taosArrayInit(TARRAY_MIN_SIZE, sizeof(SRetention));
+    }
+    taosArrayPush(expect.pRetensions, &retention);
+    ++expect.numOfRetensions;
+  };
+
+  setCheckDdlFunc([&](const SQuery* pQuery, ParserStage stage) {
+    ASSERT_EQ(nodeType(pQuery->pRoot), QUERY_NODE_CREATE_DATABASE_STMT);
+    SCreateDbReq req = {0};
+    ASSERT_TRUE(TSDB_CODE_SUCCESS == tDeserializeSCreateDbReq(pQuery->pCmdMsg->pMsg, pQuery->pCmdMsg->msgLen, &req));
+
+    ASSERT_EQ(std::string(req.db), std::string(expect.db));
+    ASSERT_EQ(req.numOfVgroups, expect.numOfVgroups);
+    ASSERT_EQ(req.numOfStables, expect.numOfStables);
+    ASSERT_EQ(req.buffer, expect.buffer);
+    ASSERT_EQ(req.pageSize, expect.pageSize);
+    ASSERT_EQ(req.pages, expect.pages);
+    ASSERT_EQ(req.daysPerFile, expect.daysPerFile);
+    ASSERT_EQ(req.daysToKeep0, expect.daysToKeep0);
+    ASSERT_EQ(req.daysToKeep1, expect.daysToKeep1);
+    ASSERT_EQ(req.daysToKeep2, expect.daysToKeep2);
+    ASSERT_EQ(req.minRows, expect.minRows);
+    ASSERT_EQ(req.maxRows, expect.maxRows);
+    ASSERT_EQ(req.fsyncPeriod, expect.fsyncPeriod);
+    ASSERT_EQ(req.walLevel, expect.walLevel);
+    ASSERT_EQ(req.precision, expect.precision);
+    ASSERT_EQ(req.compression, expect.compression);
+    ASSERT_EQ(req.replications, expect.replications);
+    ASSERT_EQ(req.strict, expect.strict);
+    ASSERT_EQ(req.cacheLastRow, expect.cacheLastRow);
+    ASSERT_EQ(req.ignoreExist, expect.ignoreExist);
+    ASSERT_EQ(req.numOfRetensions, expect.numOfRetensions);
+    if (expect.numOfRetensions > 0) {
+      ASSERT_EQ(taosArrayGetSize(req.pRetensions), expect.numOfRetensions);
+      ASSERT_EQ(taosArrayGetSize(req.pRetensions), taosArrayGetSize(expect.pRetensions));
+      for (int32_t i = 0; i < expect.numOfRetensions; ++i) {
+        SRetention* pReten = (SRetention*)taosArrayGet(req.pRetensions, i);
+        SRetention* pExpectReten = (SRetention*)taosArrayGet(expect.pRetensions, i);
+        ASSERT_EQ(pReten->freq, pExpectReten->freq);
+        ASSERT_EQ(pReten->keep, pExpectReten->keep);
+        ASSERT_EQ(pReten->freqUnit, pExpectReten->freqUnit);
+        ASSERT_EQ(pReten->keepUnit, pExpectReten->keepUnit);
+      }
+    }
+  });
+
+  setCreateDbReqFunc("wxy_db");
+  run("CREATE DATABASE wxy_db");
+
+  setCreateDbReqFunc("wxy_db", 1);
+  setDbBufferFunc(64);
+  setDbCachelastFunc(2);
+  setDbCompressionFunc(1);
+  setDbDaysFunc(100 * 1440);
+  setDbFsyncFunc(100);
+  setDbMaxRowsFunc(1000);
+  setDbMinRowsFunc(100);
+  setDbKeepFunc(1440 * 1440);
+  setDbPagesFunc(96);
+  setDbPageSizeFunc(8);
+  setDbPrecisionFunc(TSDB_TIME_PRECISION_NANO);
+  setDbReplicaFunc(3);
+  addDbRetentionFunc(15 * MILLISECOND_PER_SECOND, 7 * MILLISECOND_PER_DAY, TIME_UNIT_SECOND, TIME_UNIT_DAY);
+  addDbRetentionFunc(1 * MILLISECOND_PER_MINUTE, 21 * MILLISECOND_PER_DAY, TIME_UNIT_MINUTE, TIME_UNIT_DAY);
+  addDbRetentionFunc(15 * MILLISECOND_PER_MINUTE, 5, TIME_UNIT_MINUTE, TIME_UNIT_YEAR);
+  setDbStrictaFunc(1);
+  setDbWalLevelFunc(2);
+  setDbVgroupsFunc(100);
+  setDbSingleStableFunc(1);
+  run("CREATE DATABASE IF NOT EXISTS wxy_db "
+      "BUFFER 64 "
+      "CACHELAST 2 "
+      "COMP 1 "
+      "DAYS 100 "
+      "FSYNC 100 "
+      "MAXROWS 1000 "
+      "MINROWS 100 "
+      "KEEP 1440 "
+      "PAGES 96 "
+      "PAGESIZE 8 "
+      "PRECISION 'ns' "
+      "REPLICA 3 "
+      "RETENTIONS 15s:7d,1m:21d,15m:5y "
+      "STRICT 1 "
+      "WAL 2 "
+      "VGROUPS 100 "
+      "SINGLE_STABLE 1 ");
+
+  setCreateDbReqFunc("wxy_db", 1);
+  setDbDaysFunc(100);
+  setDbKeepFunc(1440, 300 * 60, 400 * 1440);
+  run("CREATE DATABASE IF NOT EXISTS wxy_db "
+      "DAYS 100m "
+      "KEEP 1440m,300h,400d ");
 }
 
 TEST_F(ParserInitialCTest, createDnode) {
   useDb("root", "test");
 
-  run("create dnode abc1 port 7000");
+  run("CREATE DNODE abc1 PORT 7000");
 
-  run("create dnode 1.1.1.1 port 9000");
+  run("CREATE DNODE 1.1.1.1 PORT 9000");
 }
 
-// todo create function
+// todo CREATE FUNCTION
 
 TEST_F(ParserInitialCTest, createIndexSma) {
   useDb("root", "test");
 
-  run("create sma index index1 on t1 function(max(c1), min(c3 + 10), sum(c4)) INTERVAL(10s)");
+  run("CREATE SMA INDEX index1 ON t1 FUNCTION(MAX(c1), MIN(c3 + 10), SUM(c4)) INTERVAL(10s)");
 }
 
 TEST_F(ParserInitialCTest, createMnode) {
   useDb("root", "test");
 
-  run("create mnode on dnode 1");
+  run("CREATE MNODE ON DNODE 1");
 }
 
 TEST_F(ParserInitialCTest, createQnode) {
   useDb("root", "test");
 
-  run("create qnode on dnode 1");
+  run("CREATE QNODE ON DNODE 1");
 }
 
 TEST_F(ParserInitialCTest, createSnode) {
   useDb("root", "test");
 
-  run("create snode on dnode 1");
+  run("CREATE SNODE ON DNODE 1");
 }
 
 TEST_F(ParserInitialCTest, createStable) {
@@ -194,7 +349,7 @@ TEST_F(ParserInitialCTest, createStable) {
   addFieldToCreateStbReqFunc(true, "ts", TSDB_DATA_TYPE_TIMESTAMP);
   addFieldToCreateStbReqFunc(true, "c1", TSDB_DATA_TYPE_INT);
   addFieldToCreateStbReqFunc(false, "id", TSDB_DATA_TYPE_INT);
-  run("create stable t1(ts timestamp, c1 int) TAGS(id int)");
+  run("CREATE STABLE t1(ts TIMESTAMP, c1 INT) TAGS(id INT)");
 
   setCreateStbReqFunc("t1", 1, 0.1, 2, 100, "test create table");
   addFieldToCreateStbReqFunc(true, "ts", TSDB_DATA_TYPE_TIMESTAMP, 0, 0);
@@ -227,80 +382,72 @@ TEST_F(ParserInitialCTest, createStable) {
   addFieldToCreateStbReqFunc(false, "a13", TSDB_DATA_TYPE_BOOL);
   addFieldToCreateStbReqFunc(false, "a14", TSDB_DATA_TYPE_NCHAR, 30 * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
   addFieldToCreateStbReqFunc(false, "a15", TSDB_DATA_TYPE_VARCHAR, 50 + VARSTR_HEADER_SIZE);
-  run("create stable if not exists test.t1("
+  run("CREATE STABLE IF NOT EXISTS test.t1("
       "ts TIMESTAMP, c1 INT, c2 INT UNSIGNED, c3 BIGINT, c4 BIGINT UNSIGNED, c5 FLOAT, c6 DOUBLE, c7 BINARY(20), "
       "c8 SMALLINT, c9 SMALLINT UNSIGNED COMMENT 'test column comment', c10 TINYINT, c11 TINYINT UNSIGNED, c12 BOOL, "
       "c13 NCHAR(30), c14 VARCHAR(50)) "
       "TAGS (a1 TIMESTAMP, a2 INT, a3 INT UNSIGNED, a4 BIGINT, a5 BIGINT UNSIGNED, a6 FLOAT, a7 DOUBLE, "
       "a8 BINARY(20), a9 SMALLINT, a10 SMALLINT UNSIGNED COMMENT 'test column comment', a11 TINYINT, "
       "a12 TINYINT UNSIGNED, a13 BOOL, a14 NCHAR(30), a15 VARCHAR(50)) "
-      "TTL 100 COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (min) FILE_FACTOR 0.1 DELAY 2");
+      "TTL 100 COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (MIN) FILE_FACTOR 0.1 DELAY 2");
 }
 
 TEST_F(ParserInitialCTest, createStream) {
   useDb("root", "test");
 
-  run("create stream s1 as select * from t1");
+  run("CREATE STREAM s1 AS SELECT * FROM t1");
 
-  run("create stream if not exists s1 as select * from t1");
+  run("CREATE STREAM IF NOT EXISTS s1 AS SELECT * FROM t1");
 
-  run("create stream s1 into st1 as select * from t1");
+  run("CREATE STREAM s1 INTO st1 AS SELECT * FROM t1");
 
-  run("create stream if not exists s1 trigger window_close watermark 10s into st1 as select * from t1");
+  run("CREATE STREAM IF NOT EXISTS s1 TRIGGER WINDOW_CLOSE WATERMARK 10s INTO st1 AS SELECT * FROM t1");
 }
 
 TEST_F(ParserInitialCTest, createTable) {
   useDb("root", "test");
 
-  run("create table t1(ts timestamp, c1 int)");
+  run("CREATE TABLE t1(ts TIMESTAMP, c1 INT)");
 
-  run("create table if not exists test.t1("
-      "ts TIMESTAMP, c1 INT, c2 INT UNSIGNED, c3 BIGINT, c4 BIGINT UNSIGNED, c5 FLOAT, c6 DOUBLE, c7 BINARY(20), c8 "
-      "SMALLINT, "
-      "c9 SMALLINT UNSIGNED COMMENT 'test column comment', c10 TINYINT, c11 TINYINT UNSIGNED, c12 BOOL, c13 "
-      "NCHAR(30), "
-      "c15 VARCHAR(50)) "
+  run("CREATE TABLE IF NOT EXISTS test.t1("
+      "ts TIMESTAMP, c1 INT, c2 INT UNSIGNED, c3 BIGINT, c4 BIGINT UNSIGNED, c5 FLOAT, c6 DOUBLE, c7 BINARY(20), "
+      "c8 SMALLINT, c9 SMALLINT UNSIGNED COMMENT 'test column comment', c10 TINYINT, c11 TINYINT UNSIGNED, c12 BOOL, "
+      "c13 NCHAR(30), c15 VARCHAR(50)) "
       "TTL 100 COMMENT 'test create table' SMA(c1, c2, c3)");
 
-  run("create table if not exists test.t1("
-      "ts TIMESTAMP, c1 INT, c2 INT UNSIGNED, c3 BIGINT, c4 BIGINT UNSIGNED, c5 FLOAT, c6 DOUBLE, c7 BINARY(20), c8 "
-      "SMALLINT, "
-      "c9 SMALLINT UNSIGNED COMMENT 'test column comment', c10 TINYINT, c11 TINYINT UNSIGNED, c12 BOOL, c13 "
-      "NCHAR(30), "
-      "c15 VARCHAR(50)) "
-      "TAGS (tsa TIMESTAMP, a1 INT, a2 INT UNSIGNED, a3 BIGINT, a4 BIGINT UNSIGNED, "
-      "a5 FLOAT, a6 DOUBLE, a7 "
-      "BINARY(20), a8 SMALLINT, "
-      "a9 SMALLINT UNSIGNED COMMENT 'test column comment', a10 "
-      "TINYINT, a11 TINYINT UNSIGNED, a12 BOOL, a13 NCHAR(30), "
-      "a15 VARCHAR(50)) "
-      "TTL 100 COMMENT 'test create "
-      "table' SMA(c1, c2, c3) ROLLUP (min) FILE_FACTOR 0.1 DELAY 2");
-
-  run("create table if not exists t1 using st1 tags(1, 'wxy')");
-
-  run("create table "
-      "if not exists test.t1 using test.st1 (tag1, tag2) tags(1, 'abc') "
-      "if not exists test.t2 using test.st1 (tag1, tag2) tags(2, 'abc') "
-      "if not exists test.t3 using test.st1 (tag1, tag2) tags(3, 'abc') ");
+  run("CREATE TABLE IF NOT EXISTS test.t1("
+      "ts TIMESTAMP, c1 INT, c2 INT UNSIGNED, c3 BIGINT, c4 BIGINT UNSIGNED, c5 FLOAT, c6 DOUBLE, c7 BINARY(20), "
+      "c8 SMALLINT, c9 SMALLINT UNSIGNED COMMENT 'test column comment', c10 TINYINT, c11 TINYINT UNSIGNED, c12 BOOL, "
+      "c13 NCHAR(30), c14 VARCHAR(50)) "
+      "TAGS (a1 TIMESTAMP, a2 INT, a3 INT UNSIGNED, a4 BIGINT, a5 BIGINT UNSIGNED, a6 FLOAT, a7 DOUBLE, a8 BINARY(20), "
+      "a9 SMALLINT, a10 SMALLINT UNSIGNED COMMENT 'test column comment', a11 TINYINT, a12 TINYINT UNSIGNED, a13 BOOL, "
+      "a14 NCHAR(30), a15 VARCHAR(50)) "
+      "TTL 100 COMMENT 'test create table' SMA(c1, c2, c3) ROLLUP (MIN) FILE_FACTOR 0.1 DELAY 2");
+
+  run("CREATE TABLE IF NOT EXISTS t1 USING st1 TAGS(1, 'wxy')");
+
+  run("CREATE TABLE "
+      "IF NOT EXISTS test.t1 USING test.st1 (tag1, tag2) TAGS(1, 'abc') "
+      "IF NOT EXISTS test.t2 USING test.st1 (tag1, tag2) TAGS(2, 'abc') "
+      "IF NOT EXISTS test.t3 USING test.st1 (tag1, tag2) TAGS(3, 'abc') ");
 }
 
 TEST_F(ParserInitialCTest, createTopic) {
   useDb("root", "test");
 
-  run("create topic tp1 as select * from t1");
+  run("CREATE TOPIC tp1 AS SELECT * FROM t1");
 
-  run("create topic if not exists tp1 as select * from t1");
+  run("CREATE TOPIC IF NOT EXISTS tp1 AS SELECT * FROM t1");
 
-  run("create topic tp1 as test");
+  run("CREATE TOPIC tp1 AS test");
 
-  run("create topic if not exists tp1 as test");
+  run("CREATE TOPIC IF NOT EXISTS tp1 AS test");
 }
 
 TEST_F(ParserInitialCTest, createUser) {
   useDb("root", "test");
 
-  run("create user wxy pass '123456'");
+  run("CREATE USER wxy PASS '123456'");
 }
 
 }  // namespace ParserTest

source/libs/planner/src/planner.c

@@ -16,6 +16,7 @@
 #include "planner.h"
 
 #include "planInt.h"
+#include "scalar.h"
 
 typedef struct SCollectPlaceholderValuesCxt {
   int32_t errCode;
@@ -146,7 +147,8 @@ static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
       }
 
       int32_t output = 0;
-      if (!taosMbsToUcs4(pParam->buffer, inputSize, (TdUcs4*)varDataVal(pVal->datum.p), pVal->node.resType.bytes, &output)) {
+      if (!taosMbsToUcs4(pParam->buffer, inputSize, (TdUcs4*)varDataVal(pVal->datum.p), pVal->node.resType.bytes,
+                         &output)) {
         return errno;
       }
       varDataSetLen(pVal->datum.p, output);
@@ -181,7 +183,7 @@ static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
 }
 
 static EDealRes updatePlanQueryId(SNode* pNode, void* pContext) {
-  int64_t queryId = *(uint64_t *)pContext;
+  int64_t queryId = *(uint64_t*)pContext;
 
   if (QUERY_NODE_PHYSICAL_PLAN == nodeType(pNode)) {
     SQueryPlan* planNode = (SQueryPlan*)pNode;
@@ -194,7 +196,127 @@ static EDealRes updatePlanQueryId(SNode* pNode, void* pContext) {
   return DEAL_RES_CONTINUE;
 }
 
-int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId) {
+static int32_t calcConstNode(SNode** pNode) {
+  if (NULL == *pNode) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  SNode*  pNew = NULL;
+  int32_t code = scalarCalculateConstants(*pNode, &pNew);
+  if (TSDB_CODE_SUCCESS == code) {
+    *pNode = pNew;
+  }
+  return code;
+}
+
+static int32_t calcConstList(SNodeList* pList) {
+  SNode* pNode = NULL;
+  FOREACH(pNode, pList) {
+    SNode*  pNew = NULL;
+    int32_t code = scalarCalculateConstants(pNode, &pNew);
+    if (TSDB_CODE_SUCCESS == code) {
+      REPLACE_NODE(pNew);
+    } else {
+      return code;
+    }
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static bool isEmptyResultCond(SNode** pCond) {
+  if (QUERY_NODE_VALUE != nodeType(*pCond)) {
+    return false;
+  }
+  if (((SValueNode*)*pCond)->datum.b) {
+    nodesDestroyNode(*pCond);
+    *pCond = NULL;
+    return false;
+  }
+  return true;
+}
+
+static int32_t calcConstSpecificPhysiNode(SPhysiNode* pPhyNode) {
+  switch (nodeType(pPhyNode)) {
+    case QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN:
+    case QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN:
+    case QUERY_NODE_PHYSICAL_PLAN_TABLE_SEQ_SCAN:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN:
+    case QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN:
+    case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE:
+    case QUERY_NODE_PHYSICAL_PLAN_FILL:
+      return TSDB_CODE_SUCCESS;
+    case QUERY_NODE_PHYSICAL_PLAN_PROJECT:
+      return calcConstList(((SProjectPhysiNode*)pPhyNode)->pProjections);
+    case QUERY_NODE_PHYSICAL_PLAN_JOIN:
+      return calcConstNode(&(((SJoinPhysiNode*)pPhyNode)->pOnConditions));
+    case QUERY_NODE_PHYSICAL_PLAN_AGG:
+      return calcConstList(((SAggPhysiNode*)pPhyNode)->pExprs);
+    case QUERY_NODE_PHYSICAL_PLAN_SORT:
+      return calcConstList(((SSortPhysiNode*)pPhyNode)->pExprs);
+    case QUERY_NODE_PHYSICAL_PLAN_INTERVAL:
+    case QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL:
+    case QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW:
+    case QUERY_NODE_PHYSICAL_PLAN_STATE_WINDOW:
+      return calcConstList(((SWinodwPhysiNode*)pPhyNode)->pExprs);
+    case QUERY_NODE_PHYSICAL_PLAN_PARTITION:
+      return calcConstList(((SPartitionPhysiNode*)pPhyNode)->pExprs);
+    default:
+      break;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t calcConstSubplan(SPhysiNode* pPhyNode, bool* pEmptyResult) {
+  int32_t code = calcConstNode(&pPhyNode->pConditions);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = calcConstSpecificPhysiNode(pPhyNode);
+  }
+  if (TSDB_CODE_SUCCESS != code) {
+    return code;
+  }
+
+  *pEmptyResult = isEmptyResultCond(&pPhyNode->pConditions);
+  if (*pEmptyResult) {
+    return TSDB_CODE_SUCCESS;
+  }
+
+  *pEmptyResult = true;
+
+  bool   subEmptyResult = false;
+  SNode* pChild = NULL;
+  FOREACH(pChild, pPhyNode->pChildren) {
+    code = calcConstSubplan((SPhysiNode*)pChild, &subEmptyResult);
+    if (TSDB_CODE_SUCCESS != code) {
+      return code;
+    }
+    if (!subEmptyResult) {
+      *pEmptyResult = false;
+    }
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t calcConstPhysiPlan(SQueryPlan* pPlan, bool* pEmptyResult) {
+  *pEmptyResult = true;
+
+  bool           subEmptyResult = false;
+  SNodeListNode* pNode = nodesListGetNode(pPlan->pSubplans, 0);
+  SNode*         pSubplan = NULL;
+  FOREACH(pSubplan, pNode->pNodeList) {
+    int32_t code = calcConstSubplan(((SSubplan*)pSubplan)->pNode, pEmptyResult);
+    if (TSDB_CODE_SUCCESS != code) {
+      return code;
+    }
+    if (!subEmptyResult) {
+      *pEmptyResult = false;
+    }
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colIdx, uint64_t queryId,
+                       bool* pEmptyResult) {
   int32_t size = taosArrayGetSize(pPlan->pPlaceholderValues);
   int32_t code = 0;
 
@@ -214,9 +336,10 @@ int32_t qStmtBindParam(SQueryPlan* pPlan, TAOS_MULTI_BIND* pParams, int32_t colI
 
   if (colIdx < 0 || ((colIdx + 1) == size)) {
     nodesWalkPhysiPlan((SNode*)pPlan, updatePlanQueryId, &queryId);
+    code = calcConstPhysiPlan(pPlan, pEmptyResult);
   }
 
-  return TSDB_CODE_SUCCESS;
+  return code;
 }
 
 int32_t qSubPlanToString(const SSubplan* pSubplan, char** pStr, int32_t* pLen) {

source/libs/scalar/src/filter.c

@@ -260,7 +260,7 @@ int8_t filterGetCompFuncIdx(int32_t type, int32_t optr) {
         comparFn = 20;
       } else if (optr == OP_TYPE_LIKE) {
         comparFn = 9;
-      } else if (optr == OP_TYPE_LIKE) {
+      } else if (optr == OP_TYPE_NOT_LIKE) {
         comparFn = 27;
       } else if (optr == OP_TYPE_IN) {
         comparFn = 8;
@@ -3513,6 +3513,7 @@ void fltConvertToTsValueNode(SFltTreeStat *stat, SValueNode* valueNode) {
   }
   taosMemoryFree(timeStr);
 
+  valueNode->typeData = valueNode->datum.i;
   valueNode->node.resType.type = TSDB_DATA_TYPE_TIMESTAMP;
   valueNode->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes;
 }

source/libs/scalar/src/scalar.c

@@ -92,8 +92,9 @@ int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
       }
 
       if (IS_VAR_DATA_TYPE(type)) {
-        len = varDataLen(out.columnData->pData);
-        buf = varDataVal(out.columnData->pData);
+        char* data = colDataGetVarData(out.columnData, 0);
+        len = varDataLen(data);
+        buf = varDataVal(data);
       } else {
         len = tDataTypes[type].bytes;
         buf = out.columnData->pData;
@@ -109,7 +110,7 @@ int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
     }
     
     if (taosHashPut(pObj, buf, (size_t)len, NULL, 0)) {
-      sclError("taosHashPut failed");
+      sclError("taosHashPut to set failed");
       SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
     }
 

source/libs/scalar/src/sclvector.c

@@ -328,9 +328,10 @@ static FORCE_INLINE void varToBool(char *buf, SScalarParam* pOut, int32_t rowInd
 static FORCE_INLINE void varToNchar(char* buf, SScalarParam* pOut, int32_t rowIndex) {
   int32_t len = 0;
   int32_t inputLen = varDataLen(buf);
+  int32_t outputMaxLen = (inputLen + 1) * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
 
-  char* t = taosMemoryCalloc(1,(inputLen + 1) * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
-  /*int32_t resLen = */taosMbsToUcs4(varDataVal(buf), inputLen, (TdUcs4*) varDataVal(t), pOut->columnData->info.bytes, &len);
+  char* t = taosMemoryCalloc(1, outputMaxLen);
+  /*int32_t resLen = */taosMbsToUcs4(varDataVal(buf), inputLen, (TdUcs4*) varDataVal(t), outputMaxLen, &len);
   varDataSetLen(t, len);
 
   colDataAppend(pOut->columnData, rowIndex, t, false);
@@ -512,7 +513,7 @@ int32_t vectorConvertToVarData(const SScalarParam* pIn, SScalarParam* pOut, int1
       if (outType == TSDB_DATA_TYPE_NCHAR) {
         varToNchar(tmp, pOut, i);
       } else {
-        colDataAppend(pOutputCol, i, (char *)&value, false);
+        colDataAppend(pOutputCol, i, (char *)tmp, false);
       }
     }
   } else if (IS_UNSIGNED_NUMERIC_TYPE(inType)) {
@@ -529,7 +530,7 @@ int32_t vectorConvertToVarData(const SScalarParam* pIn, SScalarParam* pOut, int1
       if (outType == TSDB_DATA_TYPE_NCHAR) {
         varToNchar(tmp, pOut, i);
       } else {
-        colDataAppend(pOutputCol, i, (char *)&value, false);
+        colDataAppend(pOutputCol, i, (char *)tmp, false);
       }
     }
   } else if (IS_FLOAT_TYPE(inType)) {
@@ -546,7 +547,7 @@ int32_t vectorConvertToVarData(const SScalarParam* pIn, SScalarParam* pOut, int1
       if (outType == TSDB_DATA_TYPE_NCHAR) {
         varToNchar(tmp, pOut, i);
       } else {
-        colDataAppend(pOutputCol, i, (char *)&value, false);
+        colDataAppend(pOutputCol, i, (char *)tmp, false);
       }
     }
   } else {

source/libs/scheduler/inc/schedulerInt.h

@@ -159,6 +159,7 @@ typedef struct SSchTask {
 
 typedef struct SSchJobAttr {
   EExplainMode explainMode;
+  bool         needRes;
   bool         syncSchedule;
   bool         queryJob;
   bool         needFlowCtrl;
@@ -190,6 +191,7 @@ typedef struct SSchJob {
   SSchTask        *fetchTask;
   int32_t          errCode;
   SArray          *errList;    // SArray<SQueryErrorInfo>
+  SRWLatch         resLock;
   void            *resData;         //TODO free it or not
   int32_t          resNumOfRows;
   const char      *sql;

source/libs/scheduler/src/scheduler.c

@@ -70,7 +70,7 @@ int32_t schInitTask(SSchJob *pJob, SSchTask *pTask, SSubplan *pPlan, SSchLevel *
 }
 
 int32_t schInitJob(SSchJob **pSchJob, SQueryPlan *pDag, void *transport, SArray *pNodeList, const char *sql,
-                   int64_t startTs, bool syncSchedule) {
+                   int64_t startTs, bool needRes, bool syncSchedule) {
   int32_t  code = 0;
   int64_t  refId = -1;
   SSchJob *pJob = taosMemoryCalloc(1, sizeof(SSchJob));
@@ -81,6 +81,7 @@ int32_t schInitJob(SSchJob **pSchJob, SQueryPlan *pDag, void *transport, SArray
 
   pJob->attr.explainMode = pDag->explainInfo.mode;
   pJob->attr.syncSchedule = syncSchedule;
+  pJob->attr.needRes = needRes;
   pJob->transport = transport;
   pJob->sql = sql;
 
@@ -1133,16 +1134,39 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
       break;
     }
     case TDMT_VND_SUBMIT_RSP: {
+      SCH_ERR_JRET(rspCode);
+
       if (msg) {
-        SSubmitRsp *rsp = (SSubmitRsp *)msg;
-        SCH_ERR_JRET(rsp->code);
+        SDecoder coder = {0};
+        SSubmitRsp *rsp = taosMemoryMalloc(sizeof(*rsp));
+        tDecoderInit(&coder, msg, msgSize);
+        code = tDecodeSSubmitRsp(&coder, rsp);
+        if (code) {
+          SCH_TASK_ELOG("decode submitRsp failed, code:%d", code);
+          tFreeSSubmitRsp(rsp);
+          SCH_ERR_JRET(code);
         }
         
-      SCH_ERR_JRET(rspCode);
+        atomic_add_fetch_32(&pJob->resNumOfRows, rsp->affectedRows);
+        SCH_TASK_DLOG("submit succeed, affectedRows:%d", rsp->affectedRows);
         
-      SSubmitRsp *rsp = (SSubmitRsp *)msg;
-      if (rsp) {
-        pJob->resNumOfRows += rsp->affectedRows;
+        if (pJob->attr.needRes) {
+          SCH_LOCK(SCH_WRITE, &pJob->resLock);
+          if (pJob->resData) {
+            SSubmitRsp *sum = pJob->resData;
+            sum->affectedRows += rsp->affectedRows;
+            sum->nBlocks += rsp->nBlocks;
+            sum->pBlocks = taosMemoryRealloc(sum->pBlocks, sum->nBlocks * sizeof(*sum->pBlocks));
+            memcpy(sum->pBlocks + sum->nBlocks - rsp->nBlocks, rsp->pBlocks, rsp->nBlocks * sizeof(*sum->pBlocks));
+            taosMemoryFree(rsp->pBlocks);
+            taosMemoryFree(rsp);
+          } else {        
+            pJob->resData = rsp;
+          }
+          SCH_UNLOCK(SCH_WRITE, &pJob->resLock);
+        } else {
+          tFreeSSubmitRsp(rsp);
+        }
       }
 
       SCH_ERR_RET(schProcessOnTaskSuccess(pJob, pTask));
@@ -2350,7 +2374,7 @@ void schFreeJobImpl(void *job) {
 }
 
 static int32_t schExecJobImpl(void *transport, SArray *pNodeList, SQueryPlan *pDag, int64_t *job, const char *sql,
-                              int64_t startTs, bool syncSchedule) {
+                              int64_t startTs, bool needRes, bool syncSchedule) {
   qDebug("QID:0x%" PRIx64 " job started", pDag->queryId);
 
   if (pNodeList == NULL || taosArrayGetSize(pNodeList) <= 0) {
@@ -2359,7 +2383,7 @@ static int32_t schExecJobImpl(void *transport, SArray *pNodeList, SQueryPlan *pD
 
   int32_t  code = 0;
   SSchJob *pJob = NULL;
-  SCH_ERR_JRET(schInitJob(&pJob, pDag, transport, pNodeList, sql, startTs, syncSchedule));
+  SCH_ERR_JRET(schInitJob(&pJob, pDag, transport, pNodeList, sql, startTs, needRes, syncSchedule));
 
   SCH_ERR_JRET(schLaunchJob(pJob));
 
@@ -2473,7 +2497,7 @@ int32_t schedulerInit(SSchedulerCfg *cfg) {
 }
 
 int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, int64_t *pJob, const char *sql,
-                         int64_t startTs, SQueryResult *pRes) {
+                         int64_t startTs, bool needRes, SQueryResult *pRes) {
   if (NULL == transport || NULL == pDag || NULL == pDag->pSubplans || NULL == pJob || NULL == pRes) {
     SCH_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
   }
@@ -2481,13 +2505,17 @@ int32_t schedulerExecJob(void *transport, SArray *nodeList, SQueryPlan *pDag, in
   if (EXPLAIN_MODE_STATIC == pDag->explainInfo.mode) {
     SCH_ERR_RET(schExecStaticExplain(transport, nodeList, pDag, pJob, sql, true));
   } else {
-    SCH_ERR_RET(schExecJobImpl(transport, nodeList, pDag, pJob, sql, startTs, true));
+    SCH_ERR_RET(schExecJobImpl(transport, nodeList, pDag, pJob, sql, startTs, needRes, true));
   }
 
   SSchJob *job = schAcquireJob(*pJob);
 
   pRes->code = atomic_load_32(&job->errCode);
   pRes->numOfRows = job->resNumOfRows;
+  if (needRes) {
+    pRes->res = job->resData;
+    job->resData = NULL;
+  }
 
   schReleaseJob(*pJob);
 
@@ -2502,7 +2530,7 @@ int32_t schedulerAsyncExecJob(void *transport, SArray *pNodeList, SQueryPlan *pD
   if (EXPLAIN_MODE_STATIC == pDag->explainInfo.mode) {
     SCH_ERR_RET(schExecStaticExplain(transport, pNodeList, pDag, pJob, sql, false));
   } else {
-    SCH_ERR_RET(schExecJobImpl(transport, pNodeList, pDag, pJob, sql, 0, false));
+    SCH_ERR_RET(schExecJobImpl(transport, pNodeList, pDag, pJob, sql, 0, false, false));
   }
 
   return TSDB_CODE_SUCCESS;

source/libs/scheduler/test/schedulerTests.cpp

@@ -985,7 +985,7 @@ TEST(insertTest, normalCase) {
   taosThreadCreate(&(thread1), &thattr, schtSendRsp, &insertJobRefId);
 
   SQueryResult res = {0};
-  code = schedulerExecJob(mockPointer, qnodeList, &dag, &insertJobRefId, "insert into tb values(now,1)", 0, &res);
+  code = schedulerExecJob(mockPointer, qnodeList, &dag, &insertJobRefId, "insert into tb values(now,1)", 0, false, &res);
   ASSERT_EQ(code, 0);
   ASSERT_EQ(res.numOfRows, 20);
 

source/os/src/osMemory.c

@@ -103,7 +103,8 @@ Dwarf_Debug tDbg;
 static TdThreadOnce traceThreadInit = PTHREAD_ONCE_INIT;
 
 void endTrace() {
-  if (traceThreadInit != PTHREAD_ONCE_INIT) {
+  TdThreadOnce tmp = PTHREAD_ONCE_INIT;
+  if (memcmp(&traceThreadInit, &tmp, sizeof(TdThreadOnce)) != 0) {
     delete_lookup_table(&lookup_table);
     dwarf_finish(tDbg);
   }

source/util/src/tcache.c

@@ -911,7 +911,8 @@ void taosCacheRefresh(SCacheObj *pCacheObj, __cache_trav_fn_t fp, void *param1)
 
 void taosStopCacheRefreshWorker(void) {
   stopRefreshWorker = true;
-  if(cacheThreadInit != PTHREAD_ONCE_INIT) taosThreadJoin(cacheRefreshWorker, NULL);
+  TdThreadOnce tmp = PTHREAD_ONCE_INIT;
+  if (memcmp(&cacheRefreshWorker, &tmp, sizeof(TdThreadOnce)) != 0) taosThreadJoin(cacheRefreshWorker, NULL);
   taosArrayDestroy(pCacheArrayList);
 }
 

source/util/src/tuuid.c

@@ -48,10 +48,18 @@ int64_t tGenIdPI64(void) {
     }
   }
 
+  int64_t id;
+  
+  while (true) {
     int64_t  ts = taosGetTimestampMs();
     uint64_t pid = taosGetPId();
     int32_t  val = atomic_add_fetch_32(&tUUIDSerialNo, 1);
 
-  int64_t id = ((tUUIDHashId & 0x07FF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
+    id = ((tUUIDHashId & 0x07FF) << 52) | ((pid & 0x0FFF) << 40) | ((ts & 0xFFFFFF) << 16) | (val & 0xFFFF);
+    if (id) {
+      break;
+    }
+  }
+
   return id;
 }

tests/script/tsim/db/alter_option.sim

@@ -66,7 +66,7 @@ print ============= create database
 #  | REPLICA value      [1 | 3]
 #  | WAL value          [1 | 2]
 
-sql create database db CACHELAST 3 COMP 0 DAYS 345600 FSYNC 1000 MAXROWS 8000 MINROWS 10 KEEP 1440000 PRECISION 'ns' REPLICA 3 WAL 2 VGROUPS 6 SINGLE_STABLE 1
+sql create database db CACHELAST 3 COMP 0 DAYS 240 FSYNC 1000 MAXROWS 8000 MINROWS 10 KEEP 1000 PRECISION 'ns' REPLICA 3 WAL 2 VGROUPS 6 SINGLE_STABLE 1
 sql show databases
 print rows: $rows
 print $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
@@ -229,7 +229,7 @@ sql_error alter database db days 0
 sql_error alter database db days 14400  # set over than keep
 
 print ============== modify keep
-sql alter database db keep 3456000
+sql alter database db keep 2400
 sql show databases
 print keep $data7_db
 if $data7_db != 3456000,3456000,3456000 then

tests/script/tsim/db/basic6.sim

@@ -15,8 +15,7 @@ $tb = $tbPrefix . $i
 
 print =============== step1
 # quorum  presicion 
-#sql create database $db vgroups 8 replica 1 days 2880 keep 3650 cache 32 blocks 12 minrows 80 maxrows 10000 wal 2 fsync 1000 comp 0 cachelast 2 precision 'us'
-sql create database $db vgroups 8 replica 1 days 2880 keep 3650 minrows 80 maxrows 10000 wal 2 fsync 1000 comp 0 cachelast 2 precision 'us'
+sql create database $db vgroups 8 replica 1 days 2 keep 10 minrows 80 maxrows 10000 wal 2 fsync 1000 comp 0 cachelast 2 precision 'us'
 sql show databases
 print $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07 $data08 $data09
 
@@ -38,7 +37,7 @@ endi
 if $data26 != 2880 then
   return -1
 endi
-if $data27 != 3650,3650,3650 then
+if $data27 != 14400,14400,14400 then
   return -1
 endi
 #if $data28 != 32 then
@@ -67,7 +66,7 @@ print =============== step4
 sql_error drop database $db
 
 print =============== step5
-sql create database $db replica 1 days 21600 keep 2160000
+sql create database $db replica 1 days 15 keep 1500
 sql show databases
 print $data00 $data01 $data02 $data03 $data04 $data05 $data06 $data07
 if $data20 != $db then

tests/system-test/7-tmq/basic5.py

+
+import taos
+import sys
+import time
+import socket
+import os
+import threading
+
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+
+class TDTestCase:
+    hostname = socket.gethostname()
+    rpcDebugFlagVal = '143'
+    clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    clientCfgDict["rpcDebugFlag"]  = rpcDebugFlagVal
+
+    updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal
+
+    print ("===================: ", updatecfgDict)
+
+    def init(self, conn, logSql):
+        tdLog.debug(f"start to excute {__file__}")
+        #tdSql.init(conn.cursor())
+        tdSql.init(conn.cursor(), logSql)  # output sql.txt file
+
+    def getBuildPath(self):
+        selfPath = os.path.dirname(os.path.realpath(__file__))
+
+        if ("community" in selfPath):
+            projPath = selfPath[:selfPath.find("community")]
+        else:
+            projPath = selfPath[:selfPath.find("tests")]
+
+        for root, dirs, files in os.walk(projPath):
+            if ("taosd" in files):
+                rootRealPath = os.path.dirname(os.path.realpath(root))
+                if ("packaging" not in rootRealPath):
+                    buildPath = root[:len(root) - len("/build/bin")]
+                    break
+        return buildPath
+
+    def create_tables(self,dbName,vgroups,stbName,ctbNum,rowsPerTbl):
+        tdSql.execute("create database if not exists %s vgroups %d"%(dbName, vgroups))        
+        tdSql.execute("use %s" %dbName)
+        tdSql.execute("create table %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"%stbName)
+        pre_create = "create table"
+        sql = pre_create
+        #tdLog.debug("doing create one  stable %s and %d  child table in %s  ..." %(stbname, count ,dbname))
+        for i in range(ctbNum):
+            sql += " %s_%d using %s tags(%d)"%(stbName,i,stbName,i+1)
+            if (i > 0) and (i%100 == 0):
+                tdSql.execute(sql)
+                sql = pre_create
+        if sql != pre_create:
+            tdSql.execute(sql)
+            
+        tdLog.debug("complete to create database[%s], stable[%s] and %d child tables" %(dbName, stbName, ctbNum))
+        return
+
+    def insert_data(self,dbName,stbName,ctbNum,rowsPerTbl,startTs):
+        tdLog.debug("start to insert data ............")
+        tdSql.execute("use %s" %dbName)
+        pre_insert = "insert into "
+        sql = pre_insert
+
+        #tdLog.debug("doing insert data into stable:%s rows:%d ..."%(stbName, allRows))
+        for i in range(ctbNum):
+            sql += " %s_%d values "%(stbName,i)
+            for j in range(rowsPerTbl):
+                sql += "(%d, %d, 'tmqrow_%d') "%(startTs + j, j, j)
+                if (j > 0) and (j%2000 == 0):
+                    tdSql.execute(sql)
+                    sql = "insert into %s_%d values " %(stbName,i)
+        #end sql
+        if sql != pre_insert:
+            # print(sql)
+            print("sql:%s"%sql)
+            tdSql.execute(sql)
+        tdLog.debug("insert data ............ [OK]")
+        return
+        
+    def prepareEnv(self, **parameterDict):
+        print ("input parameters:")
+        print (parameterDict)
+        self.create_tables(parameterDict["dbName"],\
+                           parameterDict["vgroups"],\
+                           parameterDict["stbName"],\
+                           parameterDict["ctbNum"],\
+                           parameterDict["rowsPerTbl"])
+
+        self.insert_data(parameterDict["dbName"],\
+                           parameterDict["stbName"],\
+                           parameterDict["ctbNum"],\
+                           parameterDict["rowsPerTbl"],\
+                           parameterDict["startTs"])                           
+        return                   
+        
+    def run(self):
+        tdSql.prepare()
+
+        buildPath = self.getBuildPath()
+        if (buildPath == ""):
+            tdLog.exit("taosd not found!")
+        else:
+            tdLog.info("taosd found in %s" % buildPath)
+        cfgPath = buildPath + "/../sim/psim/cfg"
+        tdLog.info("cfgPath: %s" % cfgPath)
+
+        tdLog.printNoPrefix("======== test scenario 1: ")
+        tdLog.info("step 1: create database, stb, ctb and insert data")
+        # create and start thread
+        parameterDict = {'dbName':     'db',     \
+                         'vgroups':    1,        \
+                         'stbName':    'stb',    \
+                         'ctbNum':     10,       \
+                         'rowsPerTbl': 10,    \
+                         'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
+        prepareEnvThread.start()
+        
+        # wait for data ready
+        prepareEnvThread.join()        
+        
+        tdLog.printNoPrefix("======== test scenario 2: ")
+
+
+        tdLog.printNoPrefix("======== test scenario 3: ")
+
+        #os.system('pkill tmq_sim')
+
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())