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

example/src/tmq.c

@@ -107,7 +107,7 @@ int32_t create_topic() {
   taos_free_result(pRes);
 
   /*pRes = taos_query(pConn, "create topic topic_ctb_column as abc1");*/
-  pRes = taos_query(pConn, "create topic topic_ctb_column with table as select ts, c1, c2, c3 from st1");
+  pRes = taos_query(pConn, "create topic topic_ctb_column as select ts, c1, c2, c3 from st1");
   if (taos_errno(pRes) != 0) {
     printf("failed to create topic topic_ctb_column, reason:%s\n", taos_errstr(pRes));
     return -1;
@@ -166,6 +166,7 @@ tmq_t* build_consumer() {
   tmq_conf_set(conf, "td.connect.user", "root");
   tmq_conf_set(conf, "td.connect.pass", "taosdata");
   /*tmq_conf_set(conf, "td.connect.db", "abc1");*/
+  tmq_conf_set(conf, "msg.with.table.name", "true");
   tmq_conf_set_offset_commit_cb(conf, tmq_commit_cb_print, NULL);
   tmq_t* tmq = tmq_consumer_new(conf, NULL, 0);
   assert(tmq);

include/common/tdataformat.h

@@ -391,7 +391,7 @@ static FORCE_INLINE int32_t comparTagId(const void *key1, const void *key2) {
   }
 }
 
-static FORCE_INLINE void *tdGetKVRowValOfCol(SKVRow row, int16_t colId) {
+static FORCE_INLINE void *tdGetKVRowValOfCol(const SKVRow row, int16_t colId) {
   void *ret = taosbsearch(&colId, kvRowColIdx(row), kvRowNCols(row), sizeof(SColIdx), comparTagId, TD_EQ);
   if (ret == NULL) return NULL;
   return kvRowColVal(row, (SColIdx *)ret);

include/common/tglobal.h

@@ -43,7 +43,7 @@ extern int32_t tsMaxNumOfDistinctResults;
 extern int32_t tsCompatibleModel;
 extern bool    tsEnableSlaveQuery;
 extern bool    tsPrintAuth;
-extern int64_t tsTickPerDay[3];
+extern int64_t tsTickPerMin[3];
 
 // multi-process
 extern bool    tsMultiProcess;
@@ -122,7 +122,7 @@ extern int32_t  tsDiskCfgNum;
 extern SDiskCfg tsDiskCfg[];
 
 // udf
-extern bool  tsStartUdfd;
+extern bool tsStartUdfd;
 
 // internal
 extern int32_t tsTransPullupInterval;

include/common/tmsg.h

@@ -252,6 +252,7 @@ STSRow* tGetSubmitBlkNext(SSubmitBlkIter* pIter);
 int32_t tPrintFixedSchemaSubmitReq(const SSubmitReq* pReq, STSchema* pSchema);
 
 typedef struct {
+  int32_t code;
   int8_t  hashMeta;
   int64_t uid;
   char*   tblFName;
@@ -2383,6 +2384,7 @@ typedef struct {
 typedef struct {
   SMsgHead head;
   char     subKey[TSDB_SUBSCRIBE_KEY_LEN];
+  int8_t   withTbName;
   int32_t  epoch;
   uint64_t reqId;
   int64_t  consumerId;

include/common/tmsgdef.h

@@ -217,6 +217,7 @@ enum {
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_UNKNOWN, "vnode-sync-unknown", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_COMMON_RESPONSE, "vnode-sync-common-response", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_APPLY_MSG, "vnode-sync-apply-msg", NULL, NULL)
+  TD_DEF_MSG_TYPE(TDMT_VND_SYNC_CONFIG_CHANGE, "vnode-sync-config-change", NULL, NULL)
   
   TD_DEF_MSG_TYPE(TDMT_VND_SYNC_VNODE, "vnode-sync-vnode", NULL, NULL)
   TD_DEF_MSG_TYPE(TDMT_VND_ALTER_VNODE, "vnode-alter-vnode", NULL, NULL)

include/libs/function/tudf.h

@@ -39,15 +39,6 @@ extern "C" {
 //======================================================================================
 //begin API to taosd and qworker
 
-enum {
-  UDFC_CODE_STOPPING = -1,
-  UDFC_CODE_PIPE_READ_ERR = -2,
-  UDFC_CODE_CONNECT_PIPE_ERR = -3,
-  UDFC_CODE_LOAD_UDF_FAILURE = -4,
-  UDFC_CODE_INVALID_STATE = -5,
-  UDFC_CODE_NO_PIPE = -6,
-};
-
 typedef void *UdfcFuncHandle;
 
 /**
@@ -89,6 +80,7 @@ typedef struct SUdfColumnData {
 
 typedef struct SUdfColumn {
   SUdfColumnMeta colMeta;
+  bool           hasNull;
   SUdfColumnData colData;
 } SUdfColumn;
 
@@ -232,6 +224,7 @@ static FORCE_INLINE void udfColDataSetNull(SUdfColumn* pColumn, int32_t row) {
   } else {
     udfColDataSetNull_f(pColumn, row);
   }
+  pColumn->hasNull = true;
 }
 
 static FORCE_INLINE int32_t udfColDataSet(SUdfColumn* pColumn, uint32_t currentRow, const char* pData, bool isNull) {

include/libs/nodes/querynodes.h

@@ -232,9 +232,9 @@ typedef struct SSelectStmt {
   char        stmtName[TSDB_TABLE_NAME_LEN];
   uint8_t     precision;
   bool        isEmptyResult;
+  bool        isTimeOrderQuery;
   bool        hasAggFuncs;
   bool        hasRepeatScanFuncs;
-  bool        isTimeOrderQuery;
 } SSelectStmt;
 
 typedef enum ESetOperatorType { SET_OP_TYPE_UNION_ALL = 1, SET_OP_TYPE_UNION } ESetOperatorType;

include/util/taoserror.h

@@ -634,6 +634,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_PAR_INVALID_TAGS_NUM          TAOS_DEF_ERROR_CODE(0, 0x2643)
 #define TSDB_CODE_PAR_PERMISSION_DENIED         TAOS_DEF_ERROR_CODE(0, 0x2644)
 #define TSDB_CODE_PAR_INVALID_STREAM_QUERY      TAOS_DEF_ERROR_CODE(0, 0x2645)
+#define TSDB_CODE_PAR_INVALID_INTERNAL_PK       TAOS_DEF_ERROR_CODE(0, 0x2646)
 
 //planner
 #define TSDB_CODE_PLAN_INTERNAL_ERROR           TAOS_DEF_ERROR_CODE(0, 0x2700)
@@ -645,7 +646,16 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_FUNC_FUNTION_PARA_NUM         TAOS_DEF_ERROR_CODE(0, 0x2801)
 #define TSDB_CODE_FUNC_FUNTION_PARA_TYPE        TAOS_DEF_ERROR_CODE(0, 0x2802)
 #define TSDB_CODE_FUNC_FUNTION_PARA_VALUE       TAOS_DEF_ERROR_CODE(0, 0x2803)
-#define TSDB_CODE_FUNC_INVALID_FUNTION          TAOS_DEF_ERROR_CODE(0, 0x2604)
+#define TSDB_CODE_FUNC_INVALID_FUNTION          TAOS_DEF_ERROR_CODE(0, 0x2804)
+
+//udf
+#define TSDB_CODE_UDF_STOPPING                  TAOS_DEF_ERROR_CODE(0, 0x2901)
+#define TSDB_CODE_UDF_PIPE_READ_ERR             TAOS_DEF_ERROR_CODE(0, 0x2902)
+#define TSDB_CODE_UDF_PIPE_CONNECT_ERR          TAOS_DEF_ERROR_CODE(0, 0x2903)
+#define TSDB_CODE_UDF_PIPE_NO_PIPE              TAOS_DEF_ERROR_CODE(0, 0x2904)
+#define TSDB_CODE_UDF_LOAD_UDF_FAILURE          TAOS_DEF_ERROR_CODE(0, 0x2905)
+#define TSDB_CODE_UDF_INVALID_STATE             TAOS_DEF_ERROR_CODE(0, 0x2906)
+#define TSDB_CODE_UDF_INVALID_INPUT             TAOS_DEF_ERROR_CODE(0, 0x2907)
 
 #define TSDB_CODE_SML_INVALID_PROTOCOL_TYPE     TAOS_DEF_ERROR_CODE(0, 0x3000)
 #define TSDB_CODE_SML_INVALID_PRECISION_TYPE    TAOS_DEF_ERROR_CODE(0, 0x3001)

include/util/tcoding.h

@@ -59,6 +59,21 @@ static FORCE_INLINE void *taosDecodeFixedI8(const void *buf, int8_t *value) {
 
 static FORCE_INLINE void *taosSkipFixedLen(const void *buf, size_t len) { return POINTER_SHIFT(buf, len); }
 
+// --- Bool
+
+static FORCE_INLINE int32_t taosEncodeFixedBool(void **buf, bool value) {
+  if (buf != NULL) {
+    ((int8_t *)(*buf))[0] = value ? 1 : 0;
+    *buf = POINTER_SHIFT(*buf, sizeof(int8_t));
+  }
+  return (int32_t)sizeof(int8_t);
+}
+
+static FORCE_INLINE void *taosDecodeFixedBool(const void *buf, bool *value) {
+  *value = ((int8_t *)buf)[0] == 0 ? false : true;
+  return POINTER_SHIFT(buf, sizeof(int8_t));
+}
+
 // ---- Fixed U16
 static FORCE_INLINE int32_t taosEncodeFixedU16(void **buf, uint16_t value) {
   if (buf != NULL) {

source/client/src/tmq.c

@@ -57,16 +57,17 @@ struct tmq_topic_vgroup_list_t {
 };
 
 struct tmq_conf_t {
-  char           clientId[256];
+  char     clientId[256];
-  char           groupId[TSDB_CGROUP_LEN];
+  char     groupId[TSDB_CGROUP_LEN];
-  int8_t         autoCommit;
+  int8_t   autoCommit;
-  int8_t         resetOffset;
+  int8_t   resetOffset;
-  uint16_t       port;
+  int8_t   withTbName;
-  int32_t        autoCommitInterval;
+  uint16_t port;
-  char*          ip;
+  int32_t  autoCommitInterval;
-  char*          user;
+  char*    ip;
-  char*          pass;
+  char*    user;
-  char*          db;
+  char*    pass;
+  /*char*          db;*/
   tmq_commit_cb* commitCb;
   void*          commitCbUserParam;
 };
@@ -75,6 +76,7 @@ struct tmq_t {
   // conf
   char           groupId[TSDB_CGROUP_LEN];
   char           clientId[256];
+  int8_t         withTbName;
   int8_t         autoCommit;
   int32_t        autoCommitInterval;
   int32_t        resetOffsetCfg;
@@ -187,6 +189,7 @@ typedef struct {
 
 tmq_conf_t* tmq_conf_new() {
   tmq_conf_t* conf = taosMemoryCalloc(1, sizeof(tmq_conf_t));
+  conf->withTbName = -1;
   conf->autoCommit = true;
   conf->autoCommitInterval = 5000;
   conf->resetOffset = TMQ_CONF__RESET_OFFSET__EARLIEAST;
@@ -240,6 +243,18 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
     }
   }
 
+  if (strcmp(key, "msg.with.table.name") == 0) {
+    if (strcmp(value, "true") == 0) {
+      conf->withTbName = 1;
+    } else if (strcmp(value, "false") == 0) {
+      conf->withTbName = 0;
+    } else if (strcmp(value, "none") == 0) {
+      conf->withTbName = -1;
+    } else {
+      return TMQ_CONF_INVALID;
+    }
+  }
+
   if (strcmp(key, "td.connect.ip") == 0) {
     conf->ip = strdup(value);
     return TMQ_CONF_OK;
@@ -257,7 +272,7 @@ tmq_conf_res_t tmq_conf_set(tmq_conf_t* conf, const char* key, const char* value
     return TMQ_CONF_OK;
   }
   if (strcmp(key, "td.connect.db") == 0) {
-    conf->db = strdup(value);
+    /*conf->db = strdup(value);*/
     return TMQ_CONF_OK;
   }
 
@@ -485,6 +500,7 @@ tmq_t* tmq_consumer_new(tmq_conf_t* conf, char* errstr, int32_t errstrLen) {
   // set conf
   strcpy(pTmq->clientId, conf->clientId);
   strcpy(pTmq->groupId, conf->groupId);
+  pTmq->withTbName = conf->withTbName;
   pTmq->autoCommit = conf->autoCommit;
   pTmq->autoCommitInterval = conf->autoCommitInterval;
   pTmq->commitCb = conf->commitCb;
@@ -1104,6 +1120,7 @@ SMqPollReq* tmqBuildConsumeReqImpl(tmq_t* tmq, int64_t waitTime, SMqClientTopic*
   pReq->subKey[tlen] = TMQ_SEPARATOR;
   strcpy(pReq->subKey + tlen + 1, pTopic->topicName);
 
+  pReq->withTbName = tmq->withTbName;
   pReq->waitTime = waitTime;
   pReq->consumerId = tmq->consumerId;
   pReq->epoch = tmq->epoch;

source/common/src/tdatablock.c

@@ -1311,6 +1311,7 @@ int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {
     tlen += taosEncodeFixedI16(buf, pColData->info.colId);
     tlen += taosEncodeFixedI16(buf, pColData->info.type);
     tlen += taosEncodeFixedI32(buf, pColData->info.bytes);
+    tlen += taosEncodeFixedBool(buf, pColData->hasNull);
 
     if (IS_VAR_DATA_TYPE(pColData->info.type)) {
       tlen += taosEncodeBinary(buf, pColData->varmeta.offset, sizeof(int32_t) * rows);
@@ -1340,6 +1341,7 @@ void* tDecodeDataBlock(const void* buf, SSDataBlock* pBlock) {
     buf = taosDecodeFixedI16(buf, &data.info.colId);
     buf = taosDecodeFixedI16(buf, &data.info.type);
     buf = taosDecodeFixedI32(buf, &data.info.bytes);
+    buf = taosDecodeFixedBool(buf, &data.hasNull);
 
     if (IS_VAR_DATA_TYPE(data.info.type)) {
       buf = taosDecodeBinary(buf, (void**)&data.varmeta.offset, pBlock->info.rows * sizeof(int32_t));

source/common/src/tglobal.c

@@ -153,11 +153,11 @@ bool tsStreamSchedV = true;
 
 /*
  * minimum scale for whole system, millisecond by default
- * for TSDB_TIME_PRECISION_MILLI: 86400000L
+ * for TSDB_TIME_PRECISION_MILLI: 60000L
- *     TSDB_TIME_PRECISION_MICRO: 86400000000L
+ *     TSDB_TIME_PRECISION_MICRO: 60000000L
- *     TSDB_TIME_PRECISION_NANO:  86400000000000L
+ *     TSDB_TIME_PRECISION_NANO:  60000000000L
  */
-int64_t tsTickPerDay[] = {86400000L, 86400000000L, 86400000000000L};
+int64_t tsTickPerMin[] = {60000L, 60000000L, 60000000000L};
 
 // lossy compress 6
 char tsLossyColumns[32] = "";  // "float|double" means all float and double columns can be lossy compressed.  set empty
@@ -170,7 +170,7 @@ uint32_t tsCurRange = 100;                      // range
 char     tsCompressor[32] = "ZSTD_COMPRESSOR";  // ZSTD_COMPRESSOR or GZIP_COMPRESSOR
 
 // udf
-bool     tsStartUdfd = true;
+bool tsStartUdfd = true;
 
 // internal
 int32_t tsTransPullupInterval = 6;
@@ -444,7 +444,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "transPullupInterval", tsTransPullupInterval, 1, 10000, 1) != 0) return -1;
   if (cfgAddInt32(pCfg, "mqRebalanceInterval", tsMqRebalanceInterval, 1, 10000, 1) != 0) return -1;
 
-  if (cfgAddBool(pCfg, "startUdfd", tsStartUdfd, 0) != 0) return -1;
+  if (cfgAddBool(pCfg, "udf", tsStartUdfd, 0) != 0) return -1;
   return 0;
 }
 
@@ -585,7 +585,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) {
   tsTransPullupInterval = cfgGetItem(pCfg, "transPullupInterval")->i32;
   tsMqRebalanceInterval = cfgGetItem(pCfg, "mqRebalanceInterval")->i32;
 
-  tsStartUdfd = cfgGetItem(pCfg, "startUdfd")->bval;
+  tsStartUdfd = cfgGetItem(pCfg, "udf")->bval;
 
   if (tsQueryBufferSize >= 0) {
     tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL;

source/common/src/tmsg.c

@@ -4032,6 +4032,7 @@ int32_t tDecodeSVSubmitReq(SDecoder *pCoder, SVSubmitReq *pReq) {
 static int32_t tEncodeSSubmitBlkRsp(SEncoder *pEncoder, const SSubmitBlkRsp *pBlock) {
   if (tStartEncode(pEncoder) < 0) return -1;
 
+  if (tEncodeI32(pEncoder, pBlock->code) < 0) return -1;
   if (tEncodeI8(pEncoder, pBlock->hashMeta) < 0) return -1;
   if (pBlock->hashMeta) {
     if (tEncodeI64(pEncoder, pBlock->uid) < 0) return -1;
@@ -4047,10 +4048,11 @@ static int32_t tEncodeSSubmitBlkRsp(SEncoder *pEncoder, const SSubmitBlkRsp *pBl
 static int32_t tDecodeSSubmitBlkRsp(SDecoder *pDecoder, SSubmitBlkRsp *pBlock) {
   if (tStartDecode(pDecoder) < 0) return -1;
 
+  if (tDecodeI32(pDecoder, &pBlock->code) < 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);
+    pBlock->tblFName = taosMemoryCalloc(TSDB_TABLE_FNAME_LEN, 1);
     if (NULL == pBlock->tblFName) return -1;
     if (tDecodeCStrTo(pDecoder, pBlock->tblFName) < 0) return -1;
   }
@@ -4089,7 +4091,7 @@ int32_t tDecodeSSubmitRsp(SDecoder *pDecoder, SSubmitRsp *pRsp) {
     if (tDecodeSSubmitBlkRsp(pDecoder, pRsp->pBlocks + iBlock) < 0) return -1;
   }
 
-  tEndDecode(pDecoder);  
+  tEndDecode(pDecoder);
   tDecoderClear(pDecoder);
   return 0;
 }
@@ -4108,4 +4110,3 @@ void tFreeSSubmitRsp(SSubmitRsp *pRsp) {
 
   taosMemoryFree(pRsp);
 }
-

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

@@ -144,11 +144,12 @@ static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
   pCfg->szCache = pCreate->pages;
   pCfg->szBuf = pCreate->buffer * 1024 * 1024;
   pCfg->isWeak = true;
+  pCfg->tsdbCfg.compression = pCreate->compression;
   pCfg->tsdbCfg.precision = pCreate->precision;
-  pCfg->tsdbCfg.days = 10;
+  pCfg->tsdbCfg.days = pCreate->daysPerFile;
-  pCfg->tsdbCfg.keep0 = 3650;
+  pCfg->tsdbCfg.keep0 = pCreate->daysToKeep0;
-  pCfg->tsdbCfg.keep1 = 3650;
+  pCfg->tsdbCfg.keep1 = pCreate->daysToKeep1;
-  pCfg->tsdbCfg.keep2 = 3650;
+  pCfg->tsdbCfg.keep2 = pCreate->daysToKeep2;
   pCfg->tsdbCfg.minRows = pCreate->minRows;
   pCfg->tsdbCfg.maxRows = pCreate->maxRows;
   for (size_t i = 0; i < taosArrayGetSize(pCreate->pRetensions); ++i) {

source/dnode/vnode/inc/vnode.h

@@ -72,10 +72,11 @@ typedef struct SMeta       SMeta;  // todo: remove
 typedef struct SMetaReader SMetaReader;
 typedef struct SMetaEntry  SMetaEntry;
 
-void metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags);
+void        metaReaderInit(SMetaReader *pReader, SMeta *pMeta, int32_t flags);
-void metaReaderClear(SMetaReader *pReader);
+void        metaReaderClear(SMetaReader *pReader);
-int  metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid);
+int         metaGetTableEntryByUid(SMetaReader *pReader, tb_uid_t uid);
-int  metaReadNext(SMetaReader *pReader);
+int         metaReadNext(SMetaReader *pReader);
+const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t cid);
 
 #if 1  // refact APIs below (TODO)
 typedef SVCreateTbReq   STbCfg;

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

@@ -518,9 +518,9 @@ void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn);
 
 static FORCE_INLINE int TSDB_KEY_FID(TSKEY key, int32_t days, int8_t precision) {
   if (key < 0) {
-    return (int)((key + 1) / tsTickPerDay[precision] / days - 1);
+    return (int)((key + 1) / tsTickPerMin[precision] / days - 1);
   } else {
-    return (int)((key / tsTickPerDay[precision] / days));
+    return (int)((key / tsTickPerMin[precision] / days));
   }
 }
 
@@ -770,8 +770,8 @@ static FORCE_INLINE int tsdbCopyDFileSet(SDFileSet *pSrc, SDFileSet *pDest) {
 }
 
 static FORCE_INLINE void tsdbGetFidKeyRange(int days, int8_t precision, int fid, TSKEY *minKey, TSKEY *maxKey) {
-  *minKey = fid * days * tsTickPerDay[precision];
+  *minKey = fid * days * tsTickPerMin[precision];
-  *maxKey = *minKey + days * tsTickPerDay[precision] - 1;
+  *maxKey = *minKey + days * tsTickPerMin[precision] - 1;
 }
 
 static FORCE_INLINE bool tsdbFSetIsOk(SDFileSet *pSet) {

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

@@ -466,3 +466,8 @@ void *metaGetSmaInfoByIndex(SMeta *pMeta, int64_t indexUid, bool isDecode) {
 }
 
 #endif
+
+const void *metaGetTableTagVal(SMetaEntry *pEntry, int16_t cid) {
+  ASSERT(pEntry->type == TSDB_CHILD_TABLE);
+  return tdGetKVRowValOfCol((const SKVRow)pEntry->ctbEntry.pTags, cid);
+}

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

@@ -427,13 +427,18 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
   SMqDataBlkRsp rsp = {0};
   rsp.reqOffset = pReq->currentOffset;
   rsp.withSchema = pExec->withSchema;
-  rsp.withTbName = pExec->withTbName;
 
   rsp.blockData = taosArrayInit(0, sizeof(void*));
   rsp.blockDataLen = taosArrayInit(0, sizeof(int32_t));
   rsp.blockSchema = taosArrayInit(0, sizeof(void*));
   rsp.blockTbName = taosArrayInit(0, sizeof(void*));
 
+  int8_t withTbName = pExec->withTbName;
+  if (pReq->withTbName != -1) {
+    withTbName = pReq->withTbName;
+  }
+  rsp.withTbName = withTbName;
+
   while (1) {
     consumerEpoch = atomic_load_32(&pExec->epoch);
     if (consumerEpoch > reqEpoch) {
@@ -538,7 +543,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
             taosArrayPush(rsp.blockSchema, &pSW);
           }
 
-          if (pExec->withTbName) {
+          if (withTbName) {
             SMetaReader mr = {0};
             metaReaderInit(&mr, pTq->pVnode->pMeta, 0);
             int64_t uid = pExec->pExecReader[workerId]->msgIter.uid;
@@ -578,7 +583,7 @@ int32_t tqProcessPollReq(STQ* pTq, SRpcMsg* pMsg, int32_t workerId) {
           ASSERT(actualLen <= dataStrLen);
           taosArrayPush(rsp.blockDataLen, &actualLen);
           taosArrayPush(rsp.blockData, &buf);
-          if (pExec->withTbName) {
+          if (withTbName) {
             SMetaReader mr = {0};
             metaReaderInit(&mr, pTq->pVnode->pMeta, 0);
             if (metaGetTableEntryByUid(&mr, block.info.uid) < 0) {

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

@@ -216,9 +216,9 @@ void tsdbGetRtnSnap(STsdb *pRepo, SRtn *pRtn) {
   TSKEY         minKey, midKey, maxKey, now;
 
   now = taosGetTimestamp(pCfg->precision);
-  minKey = now - pCfg->keep2 * tsTickPerDay[pCfg->precision];
+  minKey = now - pCfg->keep2 * tsTickPerMin[pCfg->precision];
-  midKey = now - pCfg->keep1 * tsTickPerDay[pCfg->precision];
+  midKey = now - pCfg->keep1 * tsTickPerMin[pCfg->precision];
-  maxKey = now - pCfg->keep0 * tsTickPerDay[pCfg->precision];
+  maxKey = now - pCfg->keep0 * tsTickPerMin[pCfg->precision];
 
   pRtn->minKey = minKey;
   pRtn->minFid = (int)(TSDB_KEY_FID(minKey, pCfg->days, pCfg->precision));
@@ -398,7 +398,7 @@ static int tsdbCommitToFile(SCommitH *pCommith, SDFileSet *pSet, int fid) {
       ++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
+      ++mIter;  // skip the table and do nothing
     } else if (pIdx) {
       if (tsdbMoveBlkIdx(pCommith, pIdx) < 0) {
         tsdbCloseCommitFile(pCommith, true);
@@ -948,7 +948,6 @@ static int tsdbMoveBlkIdx(SCommitH *pCommith, SBlockIdx *pIdx) {
 static int tsdbSetCommitTable(SCommitH *pCommith, STable *pTable) {
   STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1);
 
-
   pCommith->pTable = pTable;
 
   if (tdInitDataCols(pCommith->pDataCols, pSchema) < 0) {

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

@@ -310,6 +310,17 @@ int tsdbInsertTableData(STsdb *pTsdb, SSubmitMsgIter *pMsgIter, SSubmitBlk *pBlo
   TSKEY          keyMax;
   SSubmitBlk    *pBlkCopy;
 
+  // check if table exists
+  SMetaReader mr = {0};
+  SMetaEntry  me = {0};
+  metaReaderInit(&mr, pTsdb->pVnode->pMeta, 0);
+  if (metaGetTableEntryByUid(&mr, pMsgIter->uid) < 0) {
+    metaReaderClear(&mr);
+    terrno = TSDB_CODE_PAR_TABLE_NOT_EXIST;
+    return -1;
+  }
+  metaReaderClear(&mr);
+
   // create container is nedd
   tptr = taosHashGet(pMemTable->pHashIdx, &(pMsgIter->uid), sizeof(pMsgIter->uid));
   if (tptr == NULL) {

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

@@ -323,7 +323,7 @@ static int64_t getEarliestValidTimestamp(STsdb* pTsdb) {
   STsdbKeepCfg* pCfg = REPO_KEEP_CFG(pTsdb);
 
   int64_t now = taosGetTimestamp(pCfg->precision);
-  return now - (tsTickPerDay[pCfg->precision] * pCfg->keep2) + 1;  // needs to add one tick
+  return now - (tsTickPerMin[pCfg->precision] * pCfg->keep2) + 1;  // needs to add one tick
 }
 
 static void setQueryTimewindow(STsdbReadHandle* pTsdbReadHandle, SQueryTableDataCond* pCond) {
@@ -1047,10 +1047,10 @@ static int32_t getFileIdFromKey(TSKEY key, int32_t daysPerFile, int32_t precisio
   }
 
   if (key < 0) {
-    key -= (daysPerFile * tsTickPerDay[precision]);
+    key -= (daysPerFile * tsTickPerMin[precision]);
   }
 
-  int64_t fid = (int64_t)(key / (daysPerFile * tsTickPerDay[precision]));  // set the starting fileId
+  int64_t fid = (int64_t)(key / (daysPerFile * tsTickPerMin[precision]));  // set the starting fileId
   if (fid < 0L && llabs(fid) > INT32_MAX) {                                // data value overflow for INT32
     fid = INT32_MIN;
   }

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

@@ -1017,7 +1017,7 @@ static int32_t tsdbGetTSmaDays(STsdb *pTsdb, int64_t interval, int32_t storageLe
   int32_t       daysPerFile = pCfg->days;
 
   if (storageLevel == SMA_STORAGE_LEVEL_TSDB) {
-    int32_t days = SMA_STORAGE_TSDB_TIMES * (interval / tsTickPerDay[pCfg->precision]);
+    int32_t days = SMA_STORAGE_TSDB_TIMES * (interval / tsTickPerMin[pCfg->precision]);
     daysPerFile = days > SMA_STORAGE_TSDB_DAYS ? days : SMA_STORAGE_TSDB_DAYS;
   }
 

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

@@ -63,8 +63,8 @@ static int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq *pMsg) {
   STSRow        *row = NULL;
   STsdbKeepCfg  *pCfg = REPO_KEEP_CFG(pTsdb);
   TSKEY          now = taosGetTimestamp(pCfg->precision);
-  TSKEY          minKey = now - tsTickPerDay[pCfg->precision] * pCfg->keep2;
+  TSKEY          minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep2;
-  TSKEY          maxKey = now + tsTickPerDay[pCfg->precision] * pCfg->days;
+  TSKEY          maxKey = now + tsTickPerMin[pCfg->precision] * pCfg->days;
 
   terrno = TSDB_CODE_SUCCESS;
   // pMsg->length = htonl(pMsg->length);

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

@@ -502,7 +502,7 @@ _exit:
   return 0;
 }
 
-static int vnodeDebugPrintSingleSubmitMsg(SMeta         *pMeta, SSubmitBlk  *pBlock, SSubmitMsgIter *msgIter, const char *tags) {
+static int vnodeDebugPrintSingleSubmitMsg(SMeta *pMeta, SSubmitBlk *pBlock, SSubmitMsgIter *msgIter, const char *tags) {
   SSubmitBlkIter blkIter = {0};
   STSchema      *pSchema = NULL;
   tb_uid_t       suid = 0;
@@ -544,7 +544,7 @@ static int vnodeDebugPrintSubmitMsg(SVnode *pVnode, SSubmitReq *pMsg, const char
   while (true) {
     if (tGetSubmitMsgNext(&msgIter, &pBlock) < 0) return -1;
     if (pBlock == NULL) break;
-    
+
     vnodeDebugPrintSingleSubmitMsg(pMeta, pBlock, &msgIter, tags);
   }
 
@@ -562,6 +562,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
   int32_t        nRows;
   int32_t        tsize, ret;
   SEncoder       encoder = {0};
+  terrno = TSDB_CODE_SUCCESS;
 
   pRsp->code = 0;
 
@@ -576,6 +577,11 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
   }
 
   submitRsp.pArray = taosArrayInit(pSubmitReq->numOfBlocks, sizeof(SSubmitBlkRsp));
+  if (!submitRsp.pArray) {
+    pRsp->code = TSDB_CODE_OUT_OF_MEMORY;
+    goto _exit;
+  }
+
   for (int i = 0;;) {
     tGetSubmitMsgNext(&msgIter, &pBlock);
     if (pBlock == NULL) break;
@@ -595,7 +601,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
 
       if (metaCreateTable(pVnode->pMeta, version, &createTbReq) < 0) {
         if (terrno != TSDB_CODE_TDB_TABLE_ALREADY_EXIST) {
-          pRsp->code = terrno;
+          submitBlkRsp.code = terrno;
           tDecoderClear(&decoder);
           goto _exit;
         }
@@ -617,8 +623,7 @@ static int vnodeProcessSubmitReq(SVnode *pVnode, int64_t version, void *pReq, in
     }
 
     if (tsdbInsertTableData(pVnode->pTsdb, &msgIter, pBlock, &submitBlkRsp) < 0) {
-      pRsp->code = terrno;
+      submitBlkRsp.code = terrno;
-      goto _exit;
     }
 
     submitRsp.numOfRows += submitBlkRsp.numOfRows;
@@ -640,7 +645,12 @@ _exit:
 
   taosArrayDestroy(submitRsp.pArray);
 
-  tsdbTriggerRSma(pVnode->pTsdb, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK);
+  // TODO: the partial success scenario and the error case
+  // TODO: refactor
+  if ((terrno == TSDB_CODE_SUCCESS || terrno == TSDB_CODE_TDB_TABLE_ALREADY_EXIST) &&
+      (pRsp->code == TSDB_CODE_SUCCESS)) {
+    tsdbTriggerRSma(pVnode->pTsdb, pReq, STREAM_DATA_TYPE_SUBMIT_BLOCK);
+  }
 
   return 0;
 }

source/libs/function/src/tudf.c

@@ -695,6 +695,7 @@ int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlo
     udfCol->colMeta.scale = col->info.scale;
     udfCol->colMeta.precision = col->info.precision;
     udfCol->colData.numOfRows = udfBlock->numOfRows;
+    udfCol->hasNull = col->hasNull;
     if (IS_VAR_DATA_TYPE(udfCol->colMeta.type)) {
       udfCol->colData.varLenCol.varOffsetsLen = sizeof(int32_t) * udfBlock->numOfRows;
       udfCol->colData.varLenCol.varOffsets = taosMemoryMalloc(udfCol->colData.varLenCol.varOffsetsLen);
@@ -731,6 +732,7 @@ int32_t convertUdfColumnToDataBlock(SUdfColumn *udfCol, SSDataBlock *block) {
   col->info.bytes = meta->bytes;
   col->info.scale = meta->scale;
   col->info.type = meta->type;
+  col->hasNull = udfCol->hasNull;
   SUdfColumnData *data = &udfCol->colData;
 
   if (!IS_VAR_DATA_TYPE(meta->type)) {
@@ -929,7 +931,7 @@ void udfcUvHandleError(SClientUvConn *conn) {
   while (!QUEUE_EMPTY(&conn->taskQueue)) {
     QUEUE* h = QUEUE_HEAD(&conn->taskQueue);
     SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, connTaskQueue);
-    task->errCode = UDFC_CODE_PIPE_READ_ERR;
+    task->errCode = TSDB_CODE_UDF_PIPE_READ_ERR;
     QUEUE_REMOVE(&task->connTaskQueue);
     QUEUE_REMOVE(&task->procTaskQueue);
     uv_sem_post(&task->taskSem);
@@ -1117,7 +1119,7 @@ void cleanUpUvTasks(SUdfdProxy *udfc) {
     QUEUE_REMOVE(h);
     SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, recvTaskQueue);
     if (udfc->gUdfcState == UDFC_STATE_STOPPING) {
-      task->errCode = UDFC_CODE_STOPPING;
+      task->errCode = TSDB_CODE_UDF_STOPPING;
     }
     uv_sem_post(&task->taskSem);
   }
@@ -1127,7 +1129,7 @@ void cleanUpUvTasks(SUdfdProxy *udfc) {
     QUEUE_REMOVE(h);
     SClientUvTaskNode *task = QUEUE_DATA(h, SClientUvTaskNode, procTaskQueue);
     if (udfc->gUdfcState == UDFC_STATE_STOPPING) {
-      task->errCode = UDFC_CODE_STOPPING;
+      task->errCode = TSDB_CODE_UDF_STOPPING;
     }
     uv_sem_post(&task->taskSem);
   }
@@ -1211,7 +1213,7 @@ int32_t udfcRunUdfUvTask(SClientUdfTask *task, int8_t uvTaskType) {
 int32_t setupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
   fnInfo("udfc setup udf. udfName: %s", udfName);
   if (gUdfdProxy.gUdfcState != UDFC_STATE_READY) {
-    return UDFC_CODE_INVALID_STATE;
+    return TSDB_CODE_UDF_INVALID_STATE;
   }
   SClientUdfTask *task = taosMemoryCalloc(1,sizeof(SClientUdfTask));
   task->errCode = 0;
@@ -1225,7 +1227,7 @@ int32_t setupUdf(char udfName[], UdfcFuncHandle *funcHandle) {
   int32_t errCode = udfcRunUdfUvTask(task, UV_TASK_CONNECT);
   if (errCode != 0) {
     fnError("failed to connect to pipe. udfName: %s, pipe: %s", udfName, (&gUdfdProxy)->udfdPipeName);
-    return UDFC_CODE_CONNECT_PIPE_ERR;
+    return TSDB_CODE_UDF_PIPE_CONNECT_ERR;
   }
 
   udfcRunUdfUvTask(task, UV_TASK_REQ_RSP);
@@ -1252,7 +1254,7 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t callType, SSDataBlock *input, SUdf
   SClientUdfUvSession *session = (SClientUdfUvSession *) handle;
   if (session->udfUvPipe == NULL) {
     fnError("No pipe to udfd");
-    return UDFC_CODE_NO_PIPE;
+    return TSDB_CODE_UDF_PIPE_NO_PIPE;
   }
   SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask));
   task->errCode = 0;
@@ -1372,7 +1374,7 @@ int32_t teardownUdf(UdfcFuncHandle handle) {
   SClientUdfUvSession *session = (SClientUdfUvSession *) handle;
   if (session->udfUvPipe == NULL) {
     fnError("pipe to udfd does not exist");
-    return UDFC_CODE_NO_PIPE;
+    return TSDB_CODE_UDF_PIPE_NO_PIPE;
   }
 
   SClientUdfTask *task = taosMemoryCalloc(1, sizeof(SClientUdfTask));
@@ -1495,7 +1497,7 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
   taosArrayDestroy(tempBlock.pDataBlock);
 
   taosMemoryFree(newState.buf);
-  return TSDB_CODE_SUCCESS;
+  return udfCode;
 }
 
 int32_t udfAggFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock) {

source/libs/function/src/udfd.c

@@ -102,7 +102,7 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
   int err = uv_dlopen(udf->path, &udf->lib);
   if (err != 0) {
     fnError("can not load library %s. error: %s", udf->path, uv_strerror(err));
-    return UDFC_CODE_LOAD_UDF_FAILURE;
+    return TSDB_CODE_UDF_LOAD_UDF_FAILURE;
   }
 
   char initFuncName[TSDB_FUNC_NAME_LEN+5] = {0};
@@ -140,6 +140,182 @@ int32_t udfdLoadUdf(char *udfName, SUdf *udf) {
   return 0;
 }
 
+void udfdProcessSetupRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
+    // TODO: tracable id from client. connect, setup, call, teardown
+    fnInfo("%" PRId64 " setup request. udf name: %s", request->seqNum, request->setup.udfName);
+    SUdfSetupRequest *setup = &request->setup;
+    int32_t code = TSDB_CODE_SUCCESS;
+    SUdf *udf = NULL;
+    uv_mutex_lock(&global.udfsMutex);
+    SUdf **udfInHash = taosHashGet(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName));
+    if (udfInHash) {
+      ++(*udfInHash)->refCount;
+      udf = *udfInHash;
+      uv_mutex_unlock(&global.udfsMutex);
+    } else {
+      SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf));
+      udfNew->refCount = 1;
+      udfNew->state = UDF_STATE_INIT;
+
+      uv_mutex_init(&udfNew->lock);
+      uv_cond_init(&udfNew->condReady);
+      udf = udfNew;
+      taosHashPut(global.udfsHash, request->setup.udfName, strlen(request->setup.udfName), &udfNew, sizeof(&udfNew));
+      uv_mutex_unlock(&global.udfsMutex);
+    }
+
+    uv_mutex_lock(&udf->lock);
+    if (udf->state == UDF_STATE_INIT) {
+      udf->state = UDF_STATE_LOADING;
+      code = udfdLoadUdf(setup->udfName, udf);
+      if (udf->initFunc) {
+        udf->initFunc();
+      }
+      udf->state = UDF_STATE_READY;
+      uv_cond_broadcast(&udf->condReady);
+      uv_mutex_unlock(&udf->lock);
+    } else {
+      while (udf->state != UDF_STATE_READY) {
+        uv_cond_wait(&udf->condReady, &udf->lock);
+      }
+      uv_mutex_unlock(&udf->lock);
+    }
+    SUdfcFuncHandle *handle = taosMemoryMalloc(sizeof(SUdfcFuncHandle));
+    handle->udf = udf;
+
+    SUdfResponse rsp;
+    rsp.seqNum = request->seqNum;
+    rsp.type = request->type;
+    rsp.code = code;
+    rsp.setupRsp.udfHandle = (int64_t)(handle);
+    rsp.setupRsp.outputType = udf->outputType;
+    rsp.setupRsp.outputLen = udf->outputLen;
+    rsp.setupRsp.bufSize = udf->bufSize;
+
+    int32_t len = encodeUdfResponse(NULL, &rsp);
+    rsp.msgLen = len;
+    void *bufBegin = taosMemoryMalloc(len);
+    void *buf = bufBegin;
+    encodeUdfResponse(&buf, &rsp);
+
+    uvUdf->output = uv_buf_init(bufBegin, len);
+
+    taosMemoryFree(uvUdf->input.base);
+    return;
+}
+
+void udfdProcessCallRequest(SUvUdfWork *uvUdf, SUdfRequest *request) {
+  SUdfCallRequest *call = &request->call;
+  fnDebug("%" PRId64 "call request. call type %d, handle: %" PRIx64, request->seqNum, call->callType,
+          call->udfHandle);
+  SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(call->udfHandle);
+  SUdf            *udf = handle->udf;
+  SUdfResponse  response = {0};
+  SUdfResponse *rsp = &response;
+  SUdfCallResponse *subRsp = &rsp->callRsp;
+
+  int32_t code = TSDB_CODE_SUCCESS;
+  switch(call->callType) {
+    case TSDB_UDF_CALL_SCALA_PROC: {
+      SUdfColumn output = {0};
+
+      SUdfDataBlock input = {0};
+      convertDataBlockToUdfDataBlock(&call->block, &input);
+      code = udf->scalarProcFunc(&input, &output);
+
+      convertUdfColumnToDataBlock(&output, &response.callRsp.resultData);
+      freeUdfColumn(&output);
+      break;
+    }
+    case TSDB_UDF_CALL_AGG_INIT: {
+      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
+                             .bufLen= udf->bufSize,
+                             .numOfResult = 0};
+      udf->aggStartFunc(&outBuf);
+      subRsp->resultBuf = outBuf;
+      break;
+    }
+    case TSDB_UDF_CALL_AGG_PROC: {
+      SUdfDataBlock input = {0};
+      convertDataBlockToUdfDataBlock(&call->block, &input);
+      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
+                             .bufLen= udf->bufSize,
+                             .numOfResult = 0};
+      code = udf->aggProcFunc(&input, &call->interBuf, &outBuf);
+      subRsp->resultBuf = outBuf;
+
+      break;
+    }
+    case TSDB_UDF_CALL_AGG_FIN: {
+      SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
+                             .bufLen= udf->bufSize,
+                             .numOfResult = 0};
+      code = udf->aggFinishFunc(&call->interBuf, &outBuf);
+      subRsp->resultBuf = outBuf;
+      break;
+    }
+    default:
+      break;
+  }
+
+  rsp->seqNum = request->seqNum;
+  rsp->type = request->type;
+  rsp->code = code;
+  subRsp->callType = call->callType;
+
+  int32_t len = encodeUdfResponse(NULL, rsp);
+  rsp->msgLen = len;
+  void *bufBegin = taosMemoryMalloc(len);
+  void *buf = bufBegin;
+  encodeUdfResponse(&buf, rsp);
+  uvUdf->output = uv_buf_init(bufBegin, len);
+
+  taosMemoryFree(uvUdf->input.base);
+  return;
+}
+
+void udfdProcessTeardownRequest(SUvUdfWork* uvUdf, SUdfRequest* request) {
+  SUdfTeardownRequest *teardown = &request->teardown;
+  fnInfo("teardown. %" PRId64 "handle:%" PRIx64, request->seqNum, teardown->udfHandle);
+  SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(teardown->udfHandle);
+  SUdf *udf = handle->udf;
+  bool  unloadUdf = false;
+  int32_t code = TSDB_CODE_SUCCESS;
+
+  uv_mutex_lock(&global.udfsMutex);
+  udf->refCount--;
+  if (udf->refCount == 0) {
+    unloadUdf = true;
+    taosHashRemove(global.udfsHash, udf->name, strlen(udf->name));
+  }
+  uv_mutex_unlock(&global.udfsMutex);
+  if (unloadUdf) {
+    uv_cond_destroy(&udf->condReady);
+    uv_mutex_destroy(&udf->lock);
+    if (udf->destroyFunc) {
+      (udf->destroyFunc)();
+    }
+    uv_dlclose(&udf->lib);
+    taosMemoryFree(udf);
+  }
+  taosMemoryFree(handle);
+
+  SUdfResponse  response;
+  SUdfResponse *rsp = &response;
+  rsp->seqNum = request->seqNum;
+  rsp->type = request->type;
+  rsp->code = code;
+  int32_t len = encodeUdfResponse(NULL, rsp);
+  rsp->msgLen = len;
+  void *bufBegin = taosMemoryMalloc(len);
+  void *buf = bufBegin;
+  encodeUdfResponse(&buf, rsp);
+  uvUdf->output = uv_buf_init(bufBegin, len);
+
+  taosMemoryFree(uvUdf->input.base);
+  return;
+}
+
 void udfdProcessRequest(uv_work_t *req) {
   SUvUdfWork *uvUdf = (SUvUdfWork *)(req->data);
   SUdfRequest request = {0};
@@ -147,172 +323,16 @@ void udfdProcessRequest(uv_work_t *req) {
 
   switch (request.type) {
     case UDF_TASK_SETUP: {
-      // TODO: tracable id from client. connect, setup, call, teardown
+      udfdProcessSetupRequest(uvUdf, &request);
-      fnInfo("%" PRId64 " setup request. udf name: %s", request.seqNum, request.setup.udfName);
-      SUdfSetupRequest *setup = &request.setup;
-
-      SUdf *udf = NULL;
-      uv_mutex_lock(&global.udfsMutex);
-      SUdf **udfInHash = taosHashGet(global.udfsHash, request.setup.udfName, strlen(request.setup.udfName));
-      if (udfInHash) {
-        ++(*udfInHash)->refCount;
-        udf = *udfInHash;
-        uv_mutex_unlock(&global.udfsMutex);
-      } else {
-        SUdf *udfNew = taosMemoryCalloc(1, sizeof(SUdf));
-        udfNew->refCount = 1;
-        udfNew->state = UDF_STATE_INIT;
-
-        uv_mutex_init(&udfNew->lock);
-        uv_cond_init(&udfNew->condReady);
-        udf = udfNew;
-        taosHashPut(global.udfsHash, request.setup.udfName, strlen(request.setup.udfName), &udfNew, sizeof(&udfNew));
-        uv_mutex_unlock(&global.udfsMutex);
-      }
-
-      uv_mutex_lock(&udf->lock);
-      if (udf->state == UDF_STATE_INIT) {
-        udf->state = UDF_STATE_LOADING;
-        udfdLoadUdf(setup->udfName, udf);
-        if (udf->initFunc) {
-          udf->initFunc();
-        }
-        udf->state = UDF_STATE_READY;
-        uv_cond_broadcast(&udf->condReady);
-        uv_mutex_unlock(&udf->lock);
-      } else {
-        while (udf->state != UDF_STATE_READY) {
-          uv_cond_wait(&udf->condReady, &udf->lock);
-        }
-        uv_mutex_unlock(&udf->lock);
-      }
-      SUdfcFuncHandle *handle = taosMemoryMalloc(sizeof(SUdfcFuncHandle));
-      handle->udf = udf;
-      SUdfResponse rsp;
-      rsp.seqNum = request.seqNum;
-      rsp.type = request.type;
-      rsp.code = 0;
-      rsp.setupRsp.udfHandle = (int64_t)(handle);
-      rsp.setupRsp.outputType = udf->outputType;
-      rsp.setupRsp.outputLen = udf->outputLen;
-      rsp.setupRsp.bufSize = udf->bufSize;
-      int32_t len = encodeUdfResponse(NULL, &rsp);
-      rsp.msgLen = len;
-      void *bufBegin = taosMemoryMalloc(len);
-      void *buf = bufBegin;
-      encodeUdfResponse(&buf, &rsp);
-
-      uvUdf->output = uv_buf_init(bufBegin, len);
-
-      taosMemoryFree(uvUdf->input.base);
       break;
     }
 
     case UDF_TASK_CALL: {
-      SUdfCallRequest *call = &request.call;
+      udfdProcessCallRequest(uvUdf, &request);
-      fnDebug("%" PRId64 "call request. call type %d, handle: %" PRIx64, request.seqNum, call->callType,
-              call->udfHandle);
-      SUdfcFuncHandle *handle = (SUdfcFuncHandle *)(call->udfHandle);
-      SUdf            *udf = handle->udf;
-      SUdfResponse  response = {0};
-      SUdfResponse *rsp = &response;
-      SUdfCallResponse *subRsp = &rsp->callRsp;
-
-      switch(call->callType) {
-        case TSDB_UDF_CALL_SCALA_PROC: {
-          SUdfColumn output = {0};
-
-          SUdfDataBlock input = {0};
-          convertDataBlockToUdfDataBlock(&call->block, &input);
-          udf->scalarProcFunc(&input, &output);
-
-          convertUdfColumnToDataBlock(&output, &response.callRsp.resultData);
-          freeUdfColumn(&output);
-          break;
-        }
-        case TSDB_UDF_CALL_AGG_INIT: {
-          SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
-                                 .bufLen= udf->bufSize,
-                                 .numOfResult = 0};
-          udf->aggStartFunc(&outBuf);
-          subRsp->resultBuf = outBuf;
-          break;
-        }
-        case TSDB_UDF_CALL_AGG_PROC: {
-          SUdfDataBlock input = {0};
-          convertDataBlockToUdfDataBlock(&call->block, &input);
-          SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
-                                 .bufLen= udf->bufSize,
-                                 .numOfResult = 0};
-          udf->aggProcFunc(&input, &call->interBuf, &outBuf);
-          subRsp->resultBuf = outBuf;
-
-          break;
-        }
-        case TSDB_UDF_CALL_AGG_FIN: {
-          SUdfInterBuf outBuf = {.buf = taosMemoryMalloc(udf->bufSize),
-                                 .bufLen= udf->bufSize,
-                                 .numOfResult = 0};
-          udf->aggFinishFunc(&call->interBuf, &outBuf);
-          subRsp->resultBuf = outBuf;
-          break;
-        }
-        default:
-          break;
-      }
-
-      rsp->seqNum = request.seqNum;
-      rsp->type = request.type;
-      rsp->code = 0;
-      subRsp->callType = call->callType;
-
-      int32_t len = encodeUdfResponse(NULL, rsp);
-      rsp->msgLen = len;
-      void *bufBegin = taosMemoryMalloc(len);
-      void *buf = bufBegin;
-      encodeUdfResponse(&buf, rsp);
-      uvUdf->output = uv_buf_init(bufBegin, len);
-
-      taosMemoryFree(uvUdf->input.base);
       break;
     }
     case UDF_TASK_TEARDOWN: {
-      SUdfTeardownRequest *teardown = &request.teardown;
+      udfdProcessTeardownRequest(uvUdf, &request);
-      fnInfo("teardown. %" PRId64 "handle:%" PRIx64, request.seqNum, teardown->udfHandle) SUdfcFuncHandle *handle =
-          (SUdfcFuncHandle *)(teardown->udfHandle);
-      SUdf *udf = handle->udf;
-      bool  unloadUdf = false;
-      uv_mutex_lock(&global.udfsMutex);
-      udf->refCount--;
-      if (udf->refCount == 0) {
-        unloadUdf = true;
-        taosHashRemove(global.udfsHash, udf->name, strlen(udf->name));
-      }
-      uv_mutex_unlock(&global.udfsMutex);
-      if (unloadUdf) {
-        uv_cond_destroy(&udf->condReady);
-        uv_mutex_destroy(&udf->lock);
-        if (udf->destroyFunc) {
-          (udf->destroyFunc)();
-        }
-        uv_dlclose(&udf->lib);
-        taosMemoryFree(udf);
-      }
-      taosMemoryFree(handle);
-
-      SUdfResponse  response;
-      SUdfResponse *rsp = &response;
-      rsp->seqNum = request.seqNum;
-      rsp->type = request.type;
-      rsp->code = 0;
-      int32_t len = encodeUdfResponse(NULL, rsp);
-      rsp->msgLen = len;
-      void *bufBegin = taosMemoryMalloc(len);
-      void *buf = bufBegin;
-      encodeUdfResponse(&buf, rsp);
-      uvUdf->output = uv_buf_init(bufBegin, len);
-
-      taosMemoryFree(uvUdf->input.base);
       break;
     }
     default: {

source/libs/function/test/udf2.c

@@ -27,6 +27,12 @@ int32_t udf2_start(SUdfInterBuf *buf) {
 int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
   int64_t sumSquares = *(int64_t*)interBuf->buf;
   int8_t numOutput = 0;
+  for (int32_t i = 0; i < block->numOfCols; ++i) {
+    SUdfColumn* col = block->udfCols[i];
+    if (col->colMeta.type != TSDB_DATA_TYPE_INT) {
+      return TSDB_CODE_UDF_INVALID_INPUT;
+    }
+  }
   for (int32_t i = 0; i < block->numOfCols; ++i) {
     for (int32_t j = 0; j < block->numOfRows; ++j) {
       SUdfColumn* col = block->udfCols[i];

source/libs/parser/inc/parUtil.h

@@ -46,6 +46,7 @@ SSchema*      getTableTagSchema(const STableMeta* pTableMeta);
 int32_t       getNumOfColumns(const STableMeta* pTableMeta);
 int32_t       getNumOfTags(const STableMeta* pTableMeta);
 STableComInfo getTableInfo(const STableMeta* pTableMeta);
+STableMeta*   tableMetaDup(const STableMeta* pTableMeta);
 int           parseJsontoTagData(const char* json, SKVRowBuilder* kvRowBuilder, SMsgBuf* errMsg, int16_t startColId);
 
 int32_t trimString(const char* src, int32_t len, char* dst, int32_t dlen);

source/libs/parser/src/parAstCreater.c

@@ -78,7 +78,7 @@ static bool checkUserName(SAstCreateContext* pCxt, SToken* pUserName) {
 static bool checkPassword(SAstCreateContext* pCxt, const SToken* pPasswordToken, char* pPassword) {
   if (NULL == pPasswordToken) {
     pCxt->errCode = TSDB_CODE_PAR_SYNTAX_ERROR;
-  } else if (pPasswordToken->n >= (TSDB_USET_PASSWORD_LEN - 2)) {
+  } else if (pPasswordToken->n >= (TSDB_USET_PASSWORD_LEN + 2)) {
     pCxt->errCode = generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_NAME_OR_PASSWD_TOO_LONG);
   } else {
     strncpy(pPassword, pPasswordToken->z, pPasswordToken->n);

source/libs/parser/src/parInsert.c

@@ -53,6 +53,7 @@ typedef struct SInsertParseContext {
   SHashObj*          pTableBlockHashObj;  // global
   SHashObj*          pSubTableHashObj;    // global
   SArray*            pVgDataBlocks;       // global
+  SHashObj*          pTableNameHashObj;   // global
   int32_t            totalNum;
   SVnodeModifOpStmt* pOutput;
   SStmtCallback*     pStmtCb;
@@ -237,7 +238,7 @@ static int32_t createSName(SName* pName, SToken* pTableName, int32_t acctId, con
   return code;
 }
 
-static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SName* name, char *dbFname, bool isStb) {
+static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SName* name, char* dbFname, bool isStb) {
   SParseContext* pBasicCtx = pCxt->pComCxt;
 
   bool pass = false;
@@ -252,6 +253,7 @@ static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SName* name, char *db
   } else {
     CHECK_CODE(catalogGetTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, name,
                                    &pCxt->pTableMeta));
+    ASSERT(pCxt->pTableMeta->tableInfo.rowSize > 0);
     SVgroupInfo vg;
     CHECK_CODE(
         catalogGetTableHashVgroup(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, name, &vg));
@@ -260,9 +262,13 @@ static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SName* name, char *db
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t getTableMeta(SInsertParseContext* pCxt, SName* name, char *dbFname) { return getTableMetaImpl(pCxt, name, dbFname, false); }
+static int32_t getTableMeta(SInsertParseContext* pCxt, SName* name, char* dbFname) {
+  return getTableMetaImpl(pCxt, name, dbFname, false);
+}
 
-static int32_t getSTableMeta(SInsertParseContext* pCxt, SName* name, char *dbFname) { return getTableMetaImpl(pCxt, name, dbFname, true); }
+static int32_t getSTableMeta(SInsertParseContext* pCxt, SName* name, char* dbFname) {
+  return getTableMetaImpl(pCxt, name, dbFname, true);
+}
 
 static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pSchema) {
   while (start < end) {
@@ -863,7 +869,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SName* name, char* tb
   createSName(&sname, &sToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg);
   char stbFName[TSDB_TABLE_FNAME_LEN];
   tNameExtractFullName(&sname, stbFName);
-  
+
   CHECK_CODE(getSTableMeta(pCxt, &sname, stbFName));
   if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) {
     return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed");
@@ -1063,8 +1069,8 @@ static void destroyInsertParseContext(SInsertParseContext* pCxt) {
 //   [...];
 static int32_t parseInsertBody(SInsertParseContext* pCxt) {
   int32_t tbNum = 0;
-  char tbFName[TSDB_TABLE_FNAME_LEN];
+  char    tbFName[TSDB_TABLE_FNAME_LEN];
-  bool autoCreateTbl = false;
+  bool    autoCreateTbl = false;
 
   // for each table
   while (1) {
@@ -1106,6 +1112,8 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
     createSName(&name, &tbnameToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg);
     tNameExtractFullName(&name, tbFName);
 
+    CHECK_CODE(taosHashPut(pCxt->pTableNameHashObj, tbFName, strlen(tbFName), &name, sizeof(SName)));
+
     // USING cluase
     if (TK_USING == sToken.type) {
       CHECK_CODE(parseUsingClause(pCxt, &name, tbFName));
@@ -1158,7 +1166,8 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
       return TSDB_CODE_TSC_OUT_OF_MEMORY;
     }
     memcpy(tags, &pCxt->tags, sizeof(pCxt->tags));
-    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl, pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
+    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl,
+                                pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
 
     memset(&pCxt->tags, 0, sizeof(pCxt->tags));
     pCxt->pVgroupsHashObj = NULL;
@@ -1187,7 +1196,8 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
       .pSql = (char*)pContext->pSql,
       .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
       .pTableMeta = NULL,
-      .pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, false),
+      .pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
+      .pTableNameHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
       .totalNum = 0,
       .pOutput = (SVnodeModifOpStmt*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT),
       .pStmtCb = pContext->pStmtCb};
@@ -1196,12 +1206,13 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
     (*pContext->pStmtCb->getExecInfoFn)(pContext->pStmtCb->pStmt, &context.pVgroupsHashObj,
                                         &context.pTableBlockHashObj);
   } else {
-    context.pVgroupsHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, false);
+    context.pVgroupsHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
-    context.pTableBlockHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, false);
+    context.pTableBlockHashObj =
+        taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
   }
 
   if (NULL == context.pVgroupsHashObj || NULL == context.pTableBlockHashObj || NULL == context.pSubTableHashObj ||
-      NULL == context.pOutput) {
+      NULL == context.pTableNameHashObj || NULL == context.pOutput) {
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
 
@@ -1214,6 +1225,10 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
     if (NULL == *pQuery) {
       return TSDB_CODE_OUT_OF_MEMORY;
     }
+    (*pQuery)->pTableList = taosArrayInit(taosHashGetSize(context.pTableNameHashObj), sizeof(SName));
+    if (NULL == (*pQuery)->pTableList) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
     (*pQuery)->execMode = QUERY_EXEC_MODE_SCHEDULE;
     (*pQuery)->haveResultSet = false;
     (*pQuery)->msgType = TDMT_VND_SUBMIT;
@@ -1226,6 +1241,13 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
   if (TSDB_CODE_SUCCESS == code) {
     code = parseInsertBody(&context);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    SName* pTable = taosHashIterate(context.pTableNameHashObj, NULL);
+    while (NULL != pTable) {
+      taosArrayPush((*pQuery)->pTableList, pTable);
+      pTable = taosHashIterate(context.pTableNameHashObj, pTable);
+    }
+  }
   destroyInsertParseContext(&context);
   return code;
 }
@@ -1479,7 +1501,6 @@ int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBu
       taosMemoryFree(pSTSchema);
     }
 #endif
-
   }
 
   if (rowEnd) {
@@ -1677,10 +1698,10 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsFormat, SArray* cols
   buildCreateTbReq(&smlHandle->createTblReq, tableName, row, pTableMeta->suid);
 
   STableDataBlocks* pDataBlock = NULL;
-  ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid), TSDB_DEFAULT_PAYLOAD_SIZE,
+  ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid),
-                                  sizeof(SSubmitBlk), getTableInfo(pTableMeta).rowSize, pTableMeta,
+                             TSDB_DEFAULT_PAYLOAD_SIZE, sizeof(SSubmitBlk), getTableInfo(pTableMeta).rowSize,
-                                  &pDataBlock, NULL, &smlHandle->createTblReq);
+                             pTableMeta, &pDataBlock, NULL, &smlHandle->createTblReq);
-  if(ret != TSDB_CODE_SUCCESS){
+  if (ret != TSDB_CODE_SUCCESS) {
     buildInvalidOperationMsg(&pBuf, "create data block error");
     return ret;
   }

source/libs/parser/src/parInsertData.c

@@ -137,7 +137,7 @@ static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t star
   }
   memset(dataBuf->pData, 0, sizeof(SSubmitBlk));
 
-  dataBuf->pTableMeta = pTableMeta;
+  dataBuf->pTableMeta = tableMetaDup(pTableMeta);
 
   SParsedDataColInfo* pColInfo = &dataBuf->boundColumnInfo;
   SSchema*            pSchema = getTableColumnSchema(dataBuf->pTableMeta);
@@ -465,7 +465,7 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** p
         taosMemoryFreeClear(blkKeyInfo.pKeyTuple);
         return ret;
       }
-
+      ASSERT(pOneTableBlock->pTableMeta->tableInfo.rowSize > 0);
       // the maximum expanded size in byte when a row-wise data is converted to SDataRow format
       int32_t expandSize = isRawPayload ? getRowExpandSize(pOneTableBlock->pTableMeta) : 0;
       int64_t destSize = dataBuf->size + pOneTableBlock->size + pBlocks->numOfRows * expandSize +

source/libs/parser/src/parTranslater.c

@@ -368,11 +368,15 @@ static int32_t createColumnsByTable(STranslateContext* pCxt, const STableNode* p
   return TSDB_CODE_SUCCESS;
 }
 
+static bool isInternalPrimaryKey(const SColumnNode* pCol) {
+  return PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && 0 == strcmp(pCol->colName, PK_TS_COL_INTERNAL_NAME);
+}
+
 static bool findAndSetColumn(SColumnNode* pCol, const STableNode* pTable) {
   bool found = false;
   if (QUERY_NODE_REAL_TABLE == nodeType(pTable)) {
     const STableMeta* pMeta = ((SRealTableNode*)pTable)->pMeta;
-    if (PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId && 0 == strcmp(pCol->colName, PK_TS_COL_INTERNAL_NAME)) {
+    if (isInternalPrimaryKey(pCol)) {
       setColumnInfoBySchema((SRealTableNode*)pTable, pMeta->schema, false, pCol);
       return true;
     }
@@ -389,7 +393,9 @@ static bool findAndSetColumn(SColumnNode* pCol, const STableNode* pTable) {
     SNode*     pNode;
     FOREACH(pNode, pProjectList) {
       SExprNode* pExpr = (SExprNode*)pNode;
-      if (0 == strcmp(pCol->colName, pExpr->aliasName)) {
+      if (0 == strcmp(pCol->colName, pExpr->aliasName) ||
+          ((QUERY_NODE_COLUMN == nodeType(pExpr) && PRIMARYKEY_TIMESTAMP_COL_ID == ((SColumnNode*)pExpr)->colId) &&
+           isInternalPrimaryKey(pCol))) {
         setColumnInfoByExpr(pTable, pExpr, pCol);
         found = true;
         break;
@@ -433,7 +439,11 @@ static EDealRes translateColumnWithoutPrefix(STranslateContext* pCxt, SColumnNod
     }
   }
   if (!found) {
-    return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_COLUMN, pCol->colName);
+    if (isInternalPrimaryKey(pCol)) {
+      return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_INTERNAL_PK);
+    } else {
+      return generateDealNodeErrMsg(pCxt, TSDB_CODE_PAR_INVALID_COLUMN, pCol->colName);
+    }
   }
   return DEAL_RES_CONTINUE;
 }
@@ -3655,6 +3665,9 @@ static int32_t buildNormalTableBatchReq(int32_t acctId, const SCreateTableStmt*
     destroyCreateTbReq(&req);
     return TSDB_CODE_OUT_OF_MEMORY;
   }
+  if (pStmt->ignoreExists) {
+    req.flags |= TD_CREATE_IF_NOT_EXISTS;
+  }
   SNode*   pCol;
   col_id_t index = 0;
   FOREACH(pCol, pStmt->pCols) {
@@ -3785,24 +3798,27 @@ static int32_t rewriteCreateTable(STranslateContext* pCxt, SQuery* pQuery) {
   return code;
 }
 
-static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, const char* pDbName,
+static void addCreateTbReqIntoVgroup(int32_t acctId, SHashObj* pVgroupHashmap, SCreateSubTableClause* pStmt, SKVRow row,
-                                     const char* pTableName, SKVRow row, uint64_t suid, SVgroupInfo* pVgInfo) {
+                                     uint64_t suid, SVgroupInfo* pVgInfo) {
   char  dbFName[TSDB_DB_FNAME_LEN] = {0};
   SName name = {.type = TSDB_DB_NAME_T, .acctId = acctId};
-  strcpy(name.dbname, pDbName);
+  strcpy(name.dbname, pStmt->dbName);
   tNameGetFullDbName(&name, dbFName);
 
   struct SVCreateTbReq req = {0};
   req.type = TD_CHILD_TABLE;
-  req.name = strdup(pTableName);
+  req.name = strdup(pStmt->tableName);
   req.ctb.suid = suid;
   req.ctb.pTag = row;
+  if (pStmt->ignoreExists) {
+    req.flags |= TD_CREATE_IF_NOT_EXISTS;
+  }
 
   SVgroupCreateTableBatch* pTableBatch = taosHashGet(pVgroupHashmap, &pVgInfo->vgId, sizeof(pVgInfo->vgId));
   if (pTableBatch == NULL) {
     SVgroupCreateTableBatch tBatch = {0};
     tBatch.info = *pVgInfo;
-    strcpy(tBatch.dbName, pDbName);
+    strcpy(tBatch.dbName, pStmt->dbName);
 
     tBatch.req.pArray = taosArrayInit(4, sizeof(struct SVCreateTbReq));
     taosArrayPush(tBatch.req.pArray, &req);
@@ -3964,8 +3980,7 @@ static int32_t rewriteCreateSubTable(STranslateContext* pCxt, SCreateSubTableCla
     code = getTableHashVgroup(pCxt, pStmt->dbName, pStmt->tableName, &info);
   }
   if (TSDB_CODE_SUCCESS == code) {
-    addCreateTbReqIntoVgroup(pCxt->pParseCxt->acctId, pVgroupHashmap, pStmt->dbName, pStmt->tableName, row,
+    addCreateTbReqIntoVgroup(pCxt->pParseCxt->acctId, pVgroupHashmap, pStmt, row, pSuperTableMeta->uid, &info);
-                             pSuperTableMeta->uid, &info);
   }
 
   taosMemoryFreeClear(pSuperTableMeta);

source/libs/parser/src/parUtil.c

@@ -146,6 +146,8 @@ static char* getSyntaxErrFormat(int32_t errCode) {
       return "Invalid binary/nchar column length";
     case TSDB_CODE_PAR_INVALID_TAGS_NUM:
       return "Invalid number of tag columns";
+    case TSDB_CODE_PAR_INVALID_INTERNAL_PK:
+      return "Invalid _c0 or _rowts expression";
     case TSDB_CODE_OUT_OF_MEMORY:
       return "Out of memory";
     default:
@@ -191,7 +193,7 @@ int32_t buildSyntaxErrMsg(SMsgBuf* pBuf, const char* additionalInfo, const char*
   return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
 }
 
-SSchema *getTableColumnSchema(const STableMeta *pTableMeta) {
+SSchema* getTableColumnSchema(const STableMeta* pTableMeta) {
   assert(pTableMeta != NULL);
   return (SSchema*)pTableMeta->schema;
 }
@@ -226,6 +228,23 @@ STableComInfo getTableInfo(const STableMeta* pTableMeta) {
   return pTableMeta->tableInfo;
 }
 
+static uint32_t getTableMetaSize(const STableMeta* pTableMeta) {
+  int32_t totalCols = 0;
+  if (pTableMeta->tableInfo.numOfColumns >= 0) {
+    totalCols = pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags;
+  }
+
+  return sizeof(STableMeta) + totalCols * sizeof(SSchema);
+}
+
+STableMeta* tableMetaDup(const STableMeta* pTableMeta) {
+  size_t size = getTableMetaSize(pTableMeta);
+
+  STableMeta* p = taosMemoryMalloc(size);
+  memcpy(p, pTableMeta, size);
+  return p;
+}
+
 int32_t trimString(const char* src, int32_t len, char* dst, int32_t dlen) {
   if (len <= 0 || dlen <= 0) return 0;
 

source/libs/planner/src/planOptimizer.c

@@ -151,20 +151,32 @@ static bool needOptimizeDynamicScan(const SFunctionNode* pFunc) {
 
 static int32_t osdGetRelatedFuncs(SScanLogicNode* pScan, SNodeList** pSdrFuncs, SNodeList** pDsoFuncs) {
   SNodeList* pAllFuncs = osdGetAllFuncs(pScan->node.pParent);
+  SNodeList* pTmpSdrFuncs = NULL;
+  SNodeList* pTmpDsoFuncs = NULL;
   SNode*     pFunc = NULL;
+  bool       otherFunc = false;
   FOREACH(pFunc, pAllFuncs) {
     int32_t code = TSDB_CODE_SUCCESS;
     if (needOptimizeDataRequire((SFunctionNode*)pFunc)) {
-      code = nodesListMakeStrictAppend(pSdrFuncs, nodesCloneNode(pFunc));
+      code = nodesListMakeStrictAppend(&pTmpSdrFuncs, nodesCloneNode(pFunc));
     } else if (needOptimizeDynamicScan((SFunctionNode*)pFunc)) {
-      code = nodesListMakeStrictAppend(pDsoFuncs, nodesCloneNode(pFunc));
+      code = nodesListMakeStrictAppend(&pTmpDsoFuncs, nodesCloneNode(pFunc));
+    } else {
+      otherFunc = true;
     }
     if (TSDB_CODE_SUCCESS != code) {
-      nodesDestroyList(*pSdrFuncs);
+      nodesDestroyList(pTmpSdrFuncs);
-      nodesDestroyList(*pDsoFuncs);
+      nodesDestroyList(pTmpDsoFuncs);
       return code;
     }
   }
+  if (otherFunc) {
+    nodesDestroyList(pTmpSdrFuncs);
+    nodesDestroyList(pTmpDsoFuncs);
+  } else {
+    *pSdrFuncs = pTmpSdrFuncs;
+    *pDsoFuncs = pTmpDsoFuncs;
+  }
   return TSDB_CODE_SUCCESS;
 }
 

source/libs/planner/src/planUtil.c

@@ -18,7 +18,7 @@
 static char* getUsageErrFormat(int32_t errCode) {
   switch (errCode) {
     case TSDB_CODE_PLAN_EXPECTED_TS_EQUAL:
-      return "l.ts = r.ts is expected in join expression";
+      return "left.ts = right.ts is expected in join expression";
     case TSDB_CODE_PLAN_NOT_SUPPORT_CROSS_JOIN:
       return "not support cross join";
     default:

source/libs/planner/test/planOptimizeTest.cpp

@@ -28,6 +28,8 @@ TEST_F(PlanOptimizeTest, optimizeScanData) {
   run("SELECT COUNT(c1) FROM t1");
 
   run("SELECT COUNT(CAST(c1 AS BIGINT)) FROM t1");
+
+  run("SELECT PERCENTILE(c1, 40), COUNT(*) FROM t1");
 }
 
 TEST_F(PlanOptimizeTest, orderByPrimaryKey) {

source/libs/qcom/src/queryUtil.c

@@ -153,11 +153,6 @@ int32_t asyncSendMsgToServerExt(void* pTransporter, SEpSet* epSet, int64_t* pTra
                     .handle = pInfo->msgInfo.handle,
                     .persistHandle = persistHandle,
                     .code = 0};
-  if (pInfo->msgType == TDMT_VND_QUERY || pInfo->msgType == TDMT_VND_FETCH ||
-      pInfo->msgType == TDMT_VND_QUERY_CONTINUE) {
-    rpcMsg.persistHandle = 1;
-  }
-
   assert(pInfo->fp != NULL);
 
   rpcSendRequestWithCtx(pTransporter, epSet, &rpcMsg, pTransporterId, rpcCtx);
@@ -168,7 +163,7 @@ int32_t asyncSendMsgToServer(void* pTransporter, SEpSet* epSet, int64_t* pTransp
   return asyncSendMsgToServerExt(pTransporter, epSet, pTransporterId, pInfo, false, NULL);
 }
 
-char *jobTaskStatusStr(int32_t status) {
+char* jobTaskStatusStr(int32_t status) {
   switch (status) {
     case JOB_TASK_STATUS_NULL:
       return "NULL";
@@ -197,13 +192,10 @@ char *jobTaskStatusStr(int32_t status) {
 
 SSchema createSchema(int8_t type, int32_t bytes, col_id_t colId, const char* name) {
   SSchema s = {0};
-  s.type  = type;
+  s.type = type;
   s.bytes = bytes;
   s.colId = colId;
 
   tstrncpy(s.name, name, tListLen(s.name));
   return s;
 }
-
-
-

source/libs/scheduler/src/scheduler.c

@@ -1092,11 +1092,10 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
         if (TSDB_CODE_SUCCESS == code && batchRsp.nRsps > 0) {
           for (int32_t i = 0; i < batchRsp.nRsps; ++i) {
             SVCreateTbRsp *rsp = batchRsp.pRsps + i;
-            if (NEED_CLIENT_HANDLE_ERROR(rsp->code)) {
+            if (TSDB_CODE_SUCCESS != rsp->code) {
-              tDecoderClear(&coder);
-              SCH_ERR_JRET(rsp->code);
-            } else if (TSDB_CODE_SUCCESS != rsp->code) {
               code = rsp->code;
+              tDecoderClear(&coder);
+              SCH_ERR_JRET(code);
             }
           }
         }
@@ -1117,11 +1116,10 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
         if (TSDB_CODE_SUCCESS == code && batchRsp.nRsps > 0) {
           for (int32_t i = 0; i < batchRsp.nRsps; ++i) {
             SVDropTbRsp *rsp = batchRsp.pRsps + i;
-            if (NEED_CLIENT_HANDLE_ERROR(rsp->code)) {
+            if (TSDB_CODE_SUCCESS != rsp->code) {
-              tDecoderClear(&coder);
-              SCH_ERR_JRET(rsp->code);
-            } else if (TSDB_CODE_SUCCESS != rsp->code) {
               code = rsp->code;
+              tDecoderClear(&coder);
+              SCH_ERR_JRET(code);
             }
           }
         }
@@ -1137,7 +1135,7 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
       SCH_ERR_JRET(rspCode);
 
       if (msg) {
-        SDecoder coder = {0};
+        SDecoder    coder = {0};
         SSubmitRsp *rsp = taosMemoryMalloc(sizeof(*rsp));
         tDecoderInit(&coder, msg, msgSize);
         code = tDecodeSSubmitRsp(&coder, rsp);
@@ -1146,10 +1144,21 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
           tFreeSSubmitRsp(rsp);
           SCH_ERR_JRET(code);
         }
-        
+
+        if (rsp->nBlocks > 0) {
+          for (int32_t i = 0; i < rsp->nBlocks; ++i) {
+            SSubmitBlkRsp *blk = rsp->pBlocks + i;
+            if (TSDB_CODE_SUCCESS != blk->code) {
+              code = blk->code;
+              tFreeSSubmitRsp(rsp);
+              SCH_ERR_JRET(code);
+            }
+          }
+        }
+
         atomic_add_fetch_32(&pJob->resNumOfRows, rsp->affectedRows);
         SCH_TASK_DLOG("submit succeed, affectedRows:%d", rsp->affectedRows);
-        
+
         if (pJob->attr.needRes) {
           SCH_LOCK(SCH_WRITE, &pJob->resLock);
           if (pJob->resData) {
@@ -1160,7 +1169,7 @@ int32_t schHandleResponseMsg(SSchJob *pJob, SSchTask *pTask, int32_t msgType, ch
             memcpy(sum->pBlocks + sum->nBlocks - rsp->nBlocks, rsp->pBlocks, rsp->nBlocks * sizeof(*sum->pBlocks));
             taosMemoryFree(rsp->pBlocks);
             taosMemoryFree(rsp);
-          } else {        
+          } else {
             pJob->resData = rsp;
           }
           SCH_UNLOCK(SCH_WRITE, &pJob->resLock);

source/libs/sync/inc/syncInt.h

@@ -271,6 +271,7 @@ int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, S
 cJSON*  syncNode2Json(const SSyncNode* pSyncNode);
 char*   syncNode2Str(const SSyncNode* pSyncNode);
 char*   syncNode2SimpleStr(const SSyncNode* pSyncNode);
+void    syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg *newConfig);
 
 SSyncNode* syncNodeAcquire(int64_t rid);
 void       syncNodeRelease(SSyncNode* pNode);

source/libs/sync/inc/syncUtil.h

@@ -57,6 +57,11 @@ SyncIndex   syncUtilMinIndex(SyncIndex a, SyncIndex b);
 SyncIndex   syncUtilMaxIndex(SyncIndex a, SyncIndex b);
 void        syncUtilMsgHtoN(void* msg);
 void        syncUtilMsgNtoH(void* msg);
+bool        syncUtilIsData(tmsg_t msgType);
+bool        syncUtilUserPreCommit(tmsg_t msgType);
+bool        syncUtilUserCommit(tmsg_t msgType);
+bool        syncUtilUserRollback(tmsg_t msgType);
+
 
 #ifdef __cplusplus
 }

source/libs/sync/src/syncAppendEntries.c

@@ -19,6 +19,7 @@
 #include "syncRaftStore.h"
 #include "syncUtil.h"
 #include "syncVoteMgr.h"
+#include "syncRaftCfg.h"
 
 // TLA+ Spec
 // HandleAppendEntriesRequest(i, j, m) ==
@@ -199,7 +200,8 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
             SSyncRaftEntry* pRollBackEntry = logStoreGetEntry(ths->pLogStore, index);
             assert(pRollBackEntry != NULL);
 
-            if (pRollBackEntry->msgType != TDMT_VND_SYNC_NOOP) {
+            //if (pRollBackEntry->msgType != TDMT_VND_SYNC_NOOP) {
+            if (syncUtilUserRollback(pRollBackEntry->msgType)) {
               SRpcMsg rpcMsg;
               syncEntry2OriginalRpc(pRollBackEntry, &rpcMsg);
 
@@ -227,7 +229,8 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
         SRpcMsg rpcMsg;
         syncEntry2OriginalRpc(pAppendEntry, &rpcMsg);
         if (ths->pFsm != NULL) {
-          if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+          //if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+          if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pAppendEntry->originalRpcType)) {
             SFsmCbMeta cbMeta;
             cbMeta.index = pAppendEntry->index;
             cbMeta.isWeak = pAppendEntry->isWeak;
@@ -258,7 +261,8 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
       SRpcMsg rpcMsg;
       syncEntry2OriginalRpc(pAppendEntry, &rpcMsg);
       if (ths->pFsm != NULL) {
-        if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+        //if (ths->pFsm->FpPreCommitCb != NULL && pAppendEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+        if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pAppendEntry->originalRpcType)) {
           SFsmCbMeta cbMeta;
           cbMeta.index = pAppendEntry->index;
           cbMeta.isWeak = pAppendEntry->isWeak;
@@ -320,7 +324,8 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
               SRpcMsg rpcMsg;
               syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
-              if (ths->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+              //if (ths->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+              if (ths->pFsm->FpCommitCb != NULL && syncUtilUserCommit(pEntry->originalRpcType)) {
                 SFsmCbMeta cbMeta;
                 cbMeta.index = pEntry->index;
                 cbMeta.isWeak = pEntry->isWeak;
@@ -330,6 +335,15 @@ int32_t syncNodeOnAppendEntriesCb(SSyncNode* ths, SyncAppendEntries* pMsg) {
                 ths->pFsm->FpCommitCb(ths->pFsm, &rpcMsg, cbMeta);
               }
 
+              // config change
+              if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) {
+                SSyncCfg newSyncCfg;
+                int32_t ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg);
+                ASSERT(ret == 0);
+
+                syncNodeUpdateConfig(ths, &newSyncCfg);
+              }
+
               rpcFreeCont(rpcMsg.pCont);
               syncEntryDestory(pEntry);
             }

source/libs/sync/src/syncCommit.c

@@ -19,6 +19,7 @@
 #include "syncRaftLog.h"
 #include "syncRaftStore.h"
 #include "syncUtil.h"
+#include "syncRaftCfg.h"
 
 // \* Leader i advances its commitIndex.
 // \* This is done as a separate step from handling AppendEntries responses,
@@ -101,7 +102,8 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
           SRpcMsg rpcMsg;
           syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
-          if (pSyncNode->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+          //if (pSyncNode->pFsm->FpCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+          if (pSyncNode->pFsm->FpCommitCb != NULL && syncUtilUserCommit(pEntry->originalRpcType)) {
             SFsmCbMeta cbMeta;
             cbMeta.index = pEntry->index;
             cbMeta.isWeak = pEntry->isWeak;
@@ -111,6 +113,15 @@ void syncMaybeAdvanceCommitIndex(SSyncNode* pSyncNode) {
             pSyncNode->pFsm->FpCommitCb(pSyncNode->pFsm, &rpcMsg, cbMeta);
           }
 
+          // config change
+            if (pEntry->originalRpcType == TDMT_VND_SYNC_CONFIG_CHANGE) {
+              SSyncCfg newSyncCfg;
+              int32_t ret = syncCfgFromStr(rpcMsg.pCont, &newSyncCfg);
+              ASSERT(ret == 0);
+
+              syncNodeUpdateConfig(pSyncNode, &newSyncCfg);
+          }
+
           rpcFreeCont(rpcMsg.pCont);
           syncEntryDestory(pEntry);
         }

source/libs/sync/src/syncMain.c

@@ -116,6 +116,15 @@ void syncStop(int64_t rid) {
 
 int32_t syncReconfig(int64_t rid, const SSyncCfg* pSyncCfg) {
   int32_t ret = 0;
+  char *configChange = syncCfg2Str((SSyncCfg*)pSyncCfg);
+  SRpcMsg rpcMsg = {0};
+  rpcMsg.msgType = TDMT_VND_SYNC_CONFIG_CHANGE;
+  rpcMsg.noResp = 1;
+  rpcMsg.contLen = strlen(configChange) + 1;
+  rpcMsg.pCont = rpcMallocCont(rpcMsg.contLen);
+  snprintf(rpcMsg.pCont, rpcMsg.contLen, "%s", configChange);
+  taosMemoryFree(configChange);
+  ret = syncPropose(rid, &rpcMsg, false);
   return ret;
 }
 
@@ -849,6 +858,35 @@ char* syncNode2SimpleStr(const SSyncNode* pSyncNode) {
   return s;
 }
 
+void syncNodeUpdateConfig(SSyncNode* pSyncNode, SSyncCfg *newConfig) {
+  pSyncNode->pRaftCfg->cfg = *newConfig;
+  int32_t ret = raftCfgPersist(pSyncNode->pRaftCfg);
+  ASSERT(ret == 0);
+
+  // init internal
+  pSyncNode->myNodeInfo = pSyncNode->pRaftCfg->cfg.nodeInfo[pSyncNode->pRaftCfg->cfg.myIndex];
+  syncUtilnodeInfo2raftId(&pSyncNode->myNodeInfo, pSyncNode->vgId, &pSyncNode->myRaftId);
+
+  // init peersNum, peers, peersId
+  pSyncNode->peersNum = pSyncNode->pRaftCfg->cfg.replicaNum - 1;
+  int j = 0;
+  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
+    if (i != pSyncNode->pRaftCfg->cfg.myIndex) {
+      pSyncNode->peersNodeInfo[j] = pSyncNode->pRaftCfg->cfg.nodeInfo[i];
+      j++;
+    }
+  }
+  for (int i = 0; i < pSyncNode->peersNum; ++i) {
+    syncUtilnodeInfo2raftId(&pSyncNode->peersNodeInfo[i], pSyncNode->vgId, &pSyncNode->peersId[i]);
+  }
+
+  // init replicaNum, replicasId
+  pSyncNode->replicaNum = pSyncNode->pRaftCfg->cfg.replicaNum;
+  for (int i = 0; i < pSyncNode->pRaftCfg->cfg.replicaNum; ++i) {
+    syncUtilnodeInfo2raftId(&pSyncNode->pRaftCfg->cfg.nodeInfo[i], pSyncNode->vgId, &pSyncNode->replicasId[i]);
+  }
+}
+
 SSyncNode* syncNodeAcquire(int64_t rid) {
   SSyncNode* pNode = taosAcquireRef(tsNodeRefId, rid);
   if (pNode == NULL) {
@@ -1207,7 +1245,8 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) {
     syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
     if (ths->pFsm != NULL) {
-      if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+      //if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+      if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pEntry->originalRpcType)) {
         SFsmCbMeta cbMeta;
         cbMeta.index = pEntry->index;
         cbMeta.isWeak = pEntry->isWeak;
@@ -1228,7 +1267,8 @@ int32_t syncNodeOnClientRequestCb(SSyncNode* ths, SyncClientRequest* pMsg) {
     syncEntry2OriginalRpc(pEntry, &rpcMsg);
 
     if (ths->pFsm != NULL) {
-      if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+      //if (ths->pFsm->FpPreCommitCb != NULL && pEntry->originalRpcType != TDMT_VND_SYNC_NOOP) {
+      if (ths->pFsm->FpPreCommitCb != NULL && syncUtilUserPreCommit(pEntry->originalRpcType)) {
         SFsmCbMeta cbMeta;
         cbMeta.index = pEntry->index;
         cbMeta.isWeak = pEntry->isWeak;

source/libs/sync/src/syncUtil.c

@@ -212,4 +212,32 @@ void syncUtilMsgNtoH(void* msg) {
   SMsgHead* pHead = msg;
   pHead->contLen = ntohl(pHead->contLen);
   pHead->vgId = ntohl(pHead->vgId);
+}
+
+bool syncUtilIsData(tmsg_t msgType) {
+  if (msgType == TDMT_VND_SYNC_NOOP || msgType == TDMT_VND_SYNC_CONFIG_CHANGE) {
+    return false;
+  }
+  return true;
+}
+
+bool syncUtilUserPreCommit(tmsg_t msgType) {
+  if (msgType != TDMT_VND_SYNC_NOOP && msgType != TDMT_VND_SYNC_CONFIG_CHANGE) {
+    return true;
+  }
+  return false;
+}
+
+bool syncUtilUserCommit(tmsg_t msgType) {
+  if (msgType != TDMT_VND_SYNC_NOOP && msgType != TDMT_VND_SYNC_CONFIG_CHANGE) {
+    return true;
+  }
+  return false;
+}
+
+bool syncUtilUserRollback(tmsg_t msgType) {
+  if (msgType != TDMT_VND_SYNC_NOOP && msgType != TDMT_VND_SYNC_CONFIG_CHANGE) {
+    return true;
+  }
+  return false;
 }

source/libs/wal/src/walRead.c

@@ -65,11 +65,18 @@ static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, i
   ret = taosLSeekFile(pIdxTFile, offset, SEEK_SET);
   if (ret < 0) {
     terrno = TAOS_SYSTEM_ERROR(errno);
+    wError("failed to seek idx file, ver %ld, pos: %ld, since %s", ver, offset, terrstr());
     return -1;
   }
   SWalIdxEntry entry;
-  if (taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry)) != sizeof(SWalIdxEntry)) {
+  if ((ret = taosReadFile(pIdxTFile, &entry, sizeof(SWalIdxEntry))) != sizeof(SWalIdxEntry)) {
-    terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
+    if (ret < 0) {
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      wError("failed to read idx file, since %s", terrstr());
+    } else {
+      terrno = TSDB_CODE_WAL_FILE_CORRUPTED;
+      wError("read idx file incompletely, read bytes %d, bytes should be %lu", ret, sizeof(SWalIdxEntry));
+    }
     return -1;
   }
 
@@ -77,6 +84,7 @@ static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, i
   ret = taosLSeekFile(pLogTFile, entry.offset, SEEK_SET);
   if (ret < 0) {
     terrno = TAOS_SYSTEM_ERROR(errno);
+    wError("failed to seek log file, ver %ld, pos: %ld, since %s", ver, entry.offset, terrstr());
     return -1;
   }
   return ret;
@@ -92,6 +100,8 @@ static int32_t walReadChangeFile(SWalReadHandle *pRead, int64_t fileFirstVer) {
   TdFilePtr pLogTFile = taosOpenFile(fnameStr, TD_FILE_READ);
   if (pLogTFile == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
+    terrno = TSDB_CODE_WAL_INVALID_VER;
+    wError("cannot open file %s, since %s", fnameStr, terrstr());
     return -1;
   }
 
@@ -99,6 +109,7 @@ static int32_t walReadChangeFile(SWalReadHandle *pRead, int64_t fileFirstVer) {
   TdFilePtr pIdxTFile = taosOpenFile(fnameStr, TD_FILE_READ);
   if (pIdxTFile == NULL) {
     terrno = TAOS_SYSTEM_ERROR(errno);
+    wError("cannot open file %s, since %s", fnameStr, terrstr());
     return -1;
   }
 
@@ -113,6 +124,7 @@ static int32_t walReadSeekVer(SWalReadHandle *pRead, int64_t ver) {
     return 0;
   }
   if (ver > pWal->vers.lastVer || ver < pWal->vers.firstVer) {
+    wError("invalid version: % " PRId64 ", first ver %ld, last ver %ld", ver, pWal->vers.firstVer, pWal->vers.lastVer);
     terrno = TSDB_CODE_WAL_INVALID_VER;
     return -1;
   }
@@ -125,12 +137,13 @@ static int32_t walReadSeekVer(SWalReadHandle *pRead, int64_t ver) {
   SWalFileInfo *pRet = taosArraySearch(pWal->fileInfoSet, &tmpInfo, compareWalFileInfo, TD_LE);
   ASSERT(pRet != NULL);
   if (pRead->curFileFirstVer != pRet->firstVer) {
+    // error code set inner
     if (walReadChangeFile(pRead, pRet->firstVer) < 0) {
       return -1;
     }
   }
 
-  // code set inner
+  // error code set inner
   if (walReadSeekFilePos(pRead, pRet->firstVer, ver) < 0) {
     return -1;
   }
@@ -155,9 +168,7 @@ int32_t walFetchHead(SWalReadHandle *pRead, int64_t ver, SWalHead *pHead) {
     if (code < 0) return -1;
   }
 
-  if (!taosValidFile(pRead->pReadLogTFile)) {
+  ASSERT(taosValidFile(pRead->pReadLogTFile) == true);
-    return -1;
-  }
 
   code = taosReadFile(pRead->pReadLogTFile, pHead, sizeof(SWalHead));
   if (code != sizeof(SWalHead)) {
@@ -250,15 +261,12 @@ int32_t walReadWithHandle(SWalReadHandle *pRead, int64_t ver) {
   // TODO: check wal life
   if (pRead->curVersion != ver) {
     if (walReadSeekVer(pRead, ver) < 0) {
-      terrno = TSDB_CODE_WAL_INVALID_VER;
+      wError("unexpected wal log version: % " PRId64 ", since %s", ver, terrstr());
-      wError("unexpected wal log version: % " PRId64 ", since seek error", ver);
       return -1;
     }
   }
 
-  /*if (!taosValidFile(pRead->pReadLogTFile)) {*/
+  ASSERT(taosValidFile(pRead->pReadLogTFile) == true);
-  /*return -1;*/
-  /*}*/
 
   code = taosReadFile(pRead->pReadLogTFile, pRead->pHead, sizeof(SWalHead));
   if (code != sizeof(SWalHead)) {

source/util/src/terror.c

@@ -454,6 +454,15 @@ TAOS_DEFINE_ERROR(TSDB_CODE_PAR_PERMISSION_DENIED,        "Permission denied")
 //planner
 TAOS_DEFINE_ERROR(TSDB_CODE_PLAN_INTERNAL_ERROR,          "planner internal error")
 
+//udf
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_STOPPING,                 "udf is stopping")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_PIPE_READ_ERR,            "udf pipe read error")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_PIPE_CONNECT_ERR,         "udf pipe connect error")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_PIPE_NO_PIPE,             "udf no pipe")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_LOAD_UDF_FAILURE,         "udf load failure")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_INVALID_STATE,            "udf invalid state")
+TAOS_DEFINE_ERROR(TSDB_CODE_UDF_INVALID_INPUT,            "udf invalid function input")
+
 //schemaless
 TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_PROTOCOL_TYPE,    "Invalid line protocol type")
 TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_PRECISION_TYPE,   "Invalid timestamp precision type")

tests/pytest/util/sql.py

@@ -150,6 +150,9 @@ class TDSql:
             raise Exception(repr(e))
         return (self.queryRows, timeout)
 
+    def getRows(self):
+        return self.queryRows
+
     def checkRows(self, expectRows):
         if self.queryRows == expectRows:
             tdLog.info("sql:%s, queryRows:%d == expect:%d" % (self.sql, self.queryRows, expectRows))

tests/system-test/1-insert/insertWithMoreVgroup.py

@@ -346,8 +346,10 @@ class TDTestCase:
         return
 
     def test_case3(self):
+        self.taosBenchCreate("127.0.0.1","no","db1", "stb1", 1, 8, 1*10000)
+        # self.taosBenchCreate("test209","no","db2", "stb2", 1, 8, 1*10000)
+
         # self.taosBenchCreate("chenhaoran02","no","db1", "stb1", 1, 8, 1*10000)
-        self.taosBenchCreate("chenhaoran02","no","db1", "stb1", 1, 8, 1*1000)
 
         # self.taosBenchCreate("db1", "stb1", 4, 5, 100*10000)
         # self.taosBenchCreate("db1", "stb1", 1, 5, 100*10000)

tests/system-test/1-insert/manyVgroups.json

@@ -29,8 +29,8 @@
                     "batch_create_tbl_num": 50000,
                     "data_source": "rand",
                     "insert_mode": "taosc",
-                    "insert_rows": 0,
+                    "insert_rows": 10,
-                    "interlace_rows": 0,
+                    "interlace_rows": 100000,
                     "insert_interval": 0,
                     "max_sql_len": 10000000,
                     "disorder_ratio": 0,

tests/system-test/2-query/query_cols_tags_and_or.py

@@ -93,104 +93,124 @@ class TDTestCase:
             res = tdSql.query(query_sql.replace('*', 'last(*)'), True)
         return int(res[0][-4])
 
-    def queryTsCol(self, tb_name):
+    def queryTsCol(self, tb_name, check_elm=None):
+        select_elm = "*" if check_elm is None else check_elm
         # ts and ts
-        query_sql = f'select * from {tb_name} where ts > "2021-01-11 12:00:00" or ts < "2021-01-13 12:00:00"'
+        query_sql = f'select {select_elm} from {tb_name} where ts > "2021-01-11 12:00:00" or ts < "2021-01-13 12:00:00"'
         tdSql.query(query_sql)
         tdSql.checkRows(11)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 11)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts >= "2021-01-11 12:00:00" and ts <= "2021-01-13 12:00:00"'
+        query_sql = f'select {select_elm} from {tb_name} where ts >= "2021-01-11 12:00:00" and ts <= "2021-01-13 12:00:00"'
         tdSql.query(query_sql)
-        # tdSql.checkRows(2)
+        tdSql.checkRows(2)
-        # tdSql.checkEqual(self.queryLastC10(query_sql), 6)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 6) if select_elm == "*" else False
 
         ## ts or and tinyint col
-        query_sql = f'select * from {tb_name} where ts > "2021-01-11 12:00:00" or c1 = 2'
+        query_sql = f'select {select_elm} from {tb_name} where ts > "2021-01-11 12:00:00" or c1 = 2'
-        tdSql.error(query_sql)
+        tdSql.query(query_sql)
+        tdSql.checkRows(7)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts <= "2021-01-11 12:00:00" and c1 != 2'
+        query_sql = f'select {select_elm} from {tb_name} where ts <= "2021-01-11 12:00:00" and c1 != 2'
         tdSql.query(query_sql)
         tdSql.checkRows(4)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 5)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 5) if select_elm == "*" else False
 
         ## ts or and smallint col
-        query_sql = f'select * from {tb_name} where ts <> "2021-01-11 12:00:00" or c2 = 10'
+        query_sql = f'select {select_elm} from {tb_name} where ts <> "2021-01-11 12:00:00" or c2 = 10'
-        tdSql.error(query_sql)
+        tdSql.query(query_sql)
+        tdSql.checkRows(10)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts <= "2021-01-11 12:00:00" and c2 <= 1'
+
+        query_sql = f'select {select_elm} from {tb_name} where ts <= "2021-01-11 12:00:00" and c2 <= 1'
         tdSql.query(query_sql)
         tdSql.checkRows(1)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 1)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 1) if select_elm == "*" else False
 
         ## ts or and int col
-        query_sql = f'select * from {tb_name} where ts >= "2021-01-11 12:00:00" or c3 = 4'
+        query_sql = f'select {select_elm} from {tb_name} where ts >= "2021-01-11 12:00:00" or c3 = 4'
-        tdSql.error(query_sql)
+        tdSql.query(query_sql)
+        tdSql.checkRows(8)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts < "2021-01-11 12:00:00" and c3 = 4'
+        query_sql = f'select {select_elm} from {tb_name} where ts < "2021-01-11 12:00:00" and c3 = 4'
         tdSql.query(query_sql)
         tdSql.checkRows(1)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 4)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 4) if select_elm == "*" else False
 
         ## ts or and big col
-        query_sql = f'select * from {tb_name} where ts is Null or c4 = 5'
+        query_sql = f'select {select_elm} from {tb_name} where ts is Null or c4 = 5'
-        tdSql.error(query_sql)
-
-        query_sql = f'select * from {tb_name} where ts is not Null and c4 = 2'
         tdSql.query(query_sql)
         tdSql.checkRows(1)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 3)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 5) if select_elm == "*" else False
+
+        query_sql = f'select {select_elm} from {tb_name} where ts is not Null and c4 = 2'
+        tdSql.query(query_sql)
+        tdSql.checkRows(1)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 3) if select_elm == "*" else False
 
         ## ts or and float col
-        query_sql = f'select * from {tb_name} where ts between "2021-01-17 12:00:00" and "2021-01-23 12:00:00" or c5 = 6.6'
+        query_sql = f'select {select_elm} from {tb_name} where ts between "2021-01-17 12:00:00" and "2021-01-23 12:00:00" or c5 = 6.6'
-        tdSql.error(query_sql)
+        tdSql.query(query_sql)
+        tdSql.checkRows(5)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts < "2021-01-11 12:00:00" and c5 = 1.1'
+        query_sql = f'select {select_elm} from {tb_name} where ts < "2021-01-11 12:00:00" and c5 = 1.1'
         tdSql.query(query_sql)
         tdSql.checkRows(4)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 4)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 4) if select_elm == "*" else False
 
         ## ts or and double col
-        query_sql = f'select * from {tb_name} where ts between "2021-01-17 12:00:00" and "2021-01-23 12:00:00" or c6 = 7.7'
+        query_sql = f'select {select_elm} from {tb_name} where ts between "2021-01-17 12:00:00" and "2021-01-23 12:00:00" or c6 = 7.7'
-        tdSql.error(query_sql) 
+        tdSql.query(query_sql)
+        tdSql.checkRows(5)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts < "2021-01-11 12:00:00" and c6 = 1.1'
+        query_sql = f'select {select_elm} from {tb_name} where ts < "2021-01-11 12:00:00" and c6 = 1.1'
         tdSql.query(query_sql)
         tdSql.checkRows(4)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 4)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 4) if select_elm == "*" else False
 
         ## ts or and binary col
-        query_sql = f'select * from {tb_name} where ts < "2021-01-11 12:00:00" or c7 like "binary_"'
+        query_sql = f'select {select_elm} from {tb_name} where ts < "2021-01-11 12:00:00" or c7 like "binary_"'
-        tdSql.error(query_sql)
-
-        query_sql = f'select * from {tb_name} where ts <= "2021-01-11 12:00:00" and c7 in ("binary")'
         tdSql.query(query_sql)
         tdSql.checkRows(5)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 5)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 8) if select_elm == "*" else False
+
+        query_sql = f'select {select_elm} from {tb_name} where ts <= "2021-01-11 12:00:00" and c7 in ("binary")'
+        tdSql.query(query_sql)
+        tdSql.checkRows(5)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 5) if select_elm == "*" else False
 
         ## ts or and nchar col
-        query_sql = f'select * from {tb_name} where ts < "2021-01-11 12:00:00" or c8 like "nchar%"'
+        query_sql = f'select {select_elm} from {tb_name} where ts < "2021-01-11 12:00:00" or c8 like "nchar%"'
-        tdSql.error(query_sql)
+        tdSql.query(query_sql)
+        tdSql.checkRows(10)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 10) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts >= "2021-01-11 12:00:00" and c8 is Null'
+        query_sql = f'select {select_elm} from {tb_name} where ts >= "2021-01-11 12:00:00" and c8 is Null'
         tdSql.query(query_sql)
         tdSql.checkRows(1)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 11)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
         ## ts or and bool col
-        query_sql = f'select * from {tb_name} where ts < "2021-01-11 12:00:00" or c9=false'
+        query_sql = f'select {select_elm} from {tb_name} where ts < "2021-01-11 12:00:00" or c9=false'
-        tdSql.error(query_sql)
+        tdSql.query(query_sql)
+        tdSql.checkRows(6)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 11) if select_elm == "*" else False
 
-        query_sql = f'select * from {tb_name} where ts >= "2021-01-11 12:00:00" and c9=true'
+        query_sql = f'select {select_elm} from {tb_name} where ts >= "2021-01-11 12:00:00" and c9=true'
         tdSql.query(query_sql)
         tdSql.checkRows(5)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 9)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 9) if select_elm == "*" else False
 
         ## multi cols
-        query_sql = f'select * from {tb_name} where ts > "2021-01-03 12:00:00" and c1 != 2 and c2 >= 2 and c3 <> 4 and c4 < 4 and c5 > 1 and c6 >= 1.1 and c7 is not Null and c8 = "nchar" and c9=false'
+        query_sql = f'select {select_elm} from {tb_name} where ts > "2021-01-03 12:00:00" and c1 != 2 and c2 >= 2 and c3 <> 4 and c4 < 4 and c5 > 1 and c6 >= 1.1 and c7 is not Null and c8 = "nchar" and c9=false'
         tdSql.query(query_sql)
         tdSql.checkRows(1)
-        tdSql.checkEqual(self.queryLastC10(query_sql), 10)
+        tdSql.checkEqual(self.queryLastC10(query_sql), 10) if select_elm == "*" else False
 
     def queryTsTag(self, tb_name):
         ## ts and tinyint col
@@ -2029,12 +2049,12 @@ class TDTestCase:
         tb_name = self.initStb()
         self.queryFullTagType(tb_name)
 
-    def checkTbTsCol(self):
+    def checkTbTsCol(self, check_elm):
         '''
             Ordinary table ts and col check
         '''
         tb_name = self.initTb()
-        self.queryTsCol(tb_name)
+        self.queryTsCol(tb_name, check_elm)
 
     def checkStbTsTol(self):
         tb_name = self.initStb()
@@ -2112,8 +2132,8 @@ class TDTestCase:
         for check_elm in [None, column_name]:
             self.checkTbColTypeOperator(check_elm)
             self.checkStbColTypeOperator(check_elm)
+            self.checkTbTsCol(check_elm)
         # self.checkStbTagTypeOperator()
-        # self.checkTbTsCol()
         # self.checkStbTsTol()
         # self.checkStbTsTag()
         # self.checkStbTsColTag()

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

@@ -13,14 +13,12 @@ from util.dnodes import *
 
 class TDTestCase:
     hostname = socket.gethostname()
-    rpcDebugFlagVal = '143'
+    #rpcDebugFlagVal = '143'
-    clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    #clientCfgDict = {'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
-    clientCfgDict["rpcDebugFlag"]  = rpcDebugFlagVal
+    #clientCfgDict["rpcDebugFlag"]  = rpcDebugFlagVal
-
+    #updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
-    updatecfgDict = {'clientCfg': {}, 'serverPort': '', 'firstEp': '', 'secondEp':'', 'rpcDebugFlag':'135', 'fqdn':''}
+    #updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal
-    updatecfgDict["rpcDebugFlag"] = rpcDebugFlagVal
+    #print ("===================: ", updatecfgDict)
-
-    print ("===================: ", updatecfgDict)
 
     def init(self, conn, logSql):
         tdLog.debug(f"start to excute {__file__}")
@@ -43,27 +41,35 @@ class TDTestCase:
                     break
         return buildPath
 
-    def create_tables(self,dbName,vgroups,stbName,ctbNum,rowsPerTbl):
+    def newcur(self,cfg,host,port):
-        tdSql.execute("create database if not exists %s vgroups %d"%(dbName, vgroups))        
+        user = "root"
-        tdSql.execute("use %s" %dbName)
+        password = "taosdata"
-        tdSql.execute("create table %s (ts timestamp, c1 bigint, c2 binary(16)) tags(t1 int)"%stbName)
+        con=taos.connect(host=host, user=user, password=password, config=cfg ,port=port)
+        cur=con.cursor()
+        print(cur)
+        return cur
+
+    def create_tables(self,tsql, dbName,vgroups,stbName,ctbNum,rowsPerTbl):
+        tsql.execute("create database if not exists %s vgroups %d"%(dbName, vgroups))        
+        tsql.execute("use %s" %dbName)
+        tsql.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)
+                tsql.execute(sql)
                 sql = pre_create
         if sql != pre_create:
-            tdSql.execute(sql)
+            tsql.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):
+    def insert_data(self,tsql,dbName,stbName,ctbNum,rowsPerTbl,batchNum,startTs):
         tdLog.debug("start to insert data ............")
-        tdSql.execute("use %s" %dbName)
+        tsql.execute("use %s" %dbName)
         pre_insert = "insert into "
         sql = pre_insert
 
@@ -72,31 +78,38 @@ class TDTestCase:
             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):
+                if (j > 0) and ((j%batchNum == 0) or (j == rowsPerTbl - 1)):
-                    tdSql.execute(sql)
+                    tsql.execute(sql)
-                    sql = "insert into %s_%d values " %(stbName,i)
+                    if j < rowsPerTbl - 1:
+                        sql = "insert into %s_%d values " %(stbName,i)
+                    else:
+                        sql = "insert into "
         #end sql
         if sql != pre_insert:
-            # print(sql)
+            #print("insert sql:%s"%sql)
-            print("sql:%s"%sql)
+            tsql.execute(sql)
-            tdSql.execute(sql)
         tdLog.debug("insert data ............ [OK]")
         return
         
     def prepareEnv(self, **parameterDict):
         print ("input parameters:")
         print (parameterDict)
-        self.create_tables(parameterDict["dbName"],\
+        # create new connector for my thread
+        tsql=self.newcur(parameterDict['cfg'], 'localhost', 6030)
+        self.create_tables(tsql,\
+                           parameterDict["dbName"],\
                            parameterDict["vgroups"],\
                            parameterDict["stbName"],\
                            parameterDict["ctbNum"],\
                            parameterDict["rowsPerTbl"])
 
-        self.insert_data(parameterDict["dbName"],\
+        self.insert_data(tsql,\
-                           parameterDict["stbName"],\
+                         parameterDict["dbName"],\
-                           parameterDict["ctbNum"],\
+                         parameterDict["stbName"],\
-                           parameterDict["rowsPerTbl"],\
+                         parameterDict["ctbNum"],\
-                           parameterDict["startTs"])                           
+                         parameterDict["rowsPerTbl"],\
+                         parameterDict["batchNum"],\
+                         parameterDict["startTs"])                           
         return                   
         
     def run(self):
@@ -113,17 +126,114 @@ class TDTestCase:
         tdLog.printNoPrefix("======== test scenario 1: ")
         tdLog.info("step 1: create database, stb, ctb and insert data")
         # create and start thread
-        parameterDict = {'dbName':     'db',     \
+        parameterDict = {'cfg':        '',       \
+                         'dbName':     'db',     \
                          'vgroups':    1,        \
                          'stbName':    'stb',    \
                          'ctbNum':     10,       \
-                         'rowsPerTbl': 10,    \
+                         'rowsPerTbl': 10000,   \
+                         'batchNum':   10,       \
                          'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        parameterDict['cfg'] = cfgPath
         prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
         prepareEnvThread.start()
+        time.sleep(2)
         
+        # wait stb ready
+        while 1:
+            #tdSql.query("show %s.stables"%parameterDict['dbName'])
+            tdSql.query("show db.stables")
+            #print (self.queryResult)
+            #print (tdSql.getRows())
+            if tdSql.getRows() == 1:            
+                break
+            else:
+                time.sleep(1)
+
+        tdLog.info("create topics from super table")
+        topicFromStb = 'topic_stb_column'
+        topicFromCtb = 'topic_ctb_column'        
+        
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb, parameterDict['dbName'], parameterDict['stbName']))
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s_0" %(topicFromCtb, parameterDict['dbName'], parameterDict['stbName']))
+        
+        time.sleep(1)
+        tdSql.query("show topics")
+        print ("======================================")
+        #print (self.queryResult)
+        #tdSql.checkRows(2)
+        topic1 = tdSql.getData(0 , 0)
+        topic2 = tdSql.getData(1 , 0)
+        print (topic1)
+        print (topic2)
+        
+        print (topicFromStb)
+        print (topicFromCtb)
+        #tdLog.info("show topics: %s, %s"%topic1, topic2)
+        #if topic1 != topicFromStb or topic1 != topicFromCtb:
+        #    tdLog.exit("topic error1") 
+        #if topic2 != topicFromStb or topic2 != topicFromCtb:
+        #    tdLog.exit("topic error2") 
+         
+        tdLog.info("create consume info table and consume result table")
+        cdbName = parameterDict["dbName"]
+        #tdSql.query("create database %s"%cdbName)
+        #tdSql.query("use %s"%cdbName)
+        tdSql.query("create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)")
+        tdSql.query("create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)")
+
+        consumerId   = 0
+        expectmsgcnt = (parameterDict["rowsPerTbl"] / parameterDict["batchNum"] ) * parameterDict["ctbNum"]
+        expectmsgcnt1 = expectmsgcnt + parameterDict["ctbNum"]
+        topicList    = topicFromStb
+        ifcheckdata  = 0
+        keyList      = 'group.id:cgrp1,\
+                        enable.auto.commit:false,\
+                        auto.commit.interval.ms:6000,\
+                        auto.offset.reset:earliest'
+        sql = "insert into consumeinfo values "
+        sql += "(now, %d, '%s', '%s', %d, %d)"%(consumerId, topicList, keyList, expectmsgcnt1, ifcheckdata)
+        tdSql.query(sql)
+        
+        tdLog.info("check stb if there are data")
+        while 1:
+            tdSql.query("select count(*) from %s"%parameterDict["stbName"])
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            countOfStb = tdSql.getData(0, 0)
+            if countOfStb != 0:
+                tdLog.info("count from stb: %d"%countOfStb)
+                break
+            else:
+                time.sleep(1)
+        
+        tdLog.info("start consume processor")
+        pollDelay = 5
+        showMsg   = 1
+        showRow   = 1
+        
+        shellCmd = 'nohup ' + buildPath + '/build/bin/tmq_sim -c ' + cfgPath
+        shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, parameterDict["dbName"], showMsg, showRow, cdbName) 
+        shellCmd += "> /dev/null 2>&1 &"        
+        tdLog.info(shellCmd)
+        os.system(shellCmd)        
+
         # wait for data ready
-        prepareEnvThread.join()        
+        prepareEnvThread.join()
+        
+        tdLog.info("insert process end, and start to check consume result")
+        while 1:
+            tdSql.query("select * from consumeresult")
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            if tdSql.getRows() == 1:
+                break
+            else:
+                time.sleep(5)
+
+        expectrowcnt = parameterDict["rowsPerTbl"]  * parameterDict["ctbNum"]
+                    
+        tdSql.checkData(0 , 1, consumerId)
+        tdSql.checkData(0 , 2, expectmsgcnt)
+        tdSql.checkData(0 , 3, expectrowcnt)
         
         tdLog.printNoPrefix("======== test scenario 2: ")
 

tests/system-test/99-TDcase/TD-15517.py

@@ -0,0 +1,372 @@
+
+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 newcur(self,cfg,host,port):
+        user = "root"
+        password = "taosdata"
+        con=taos.connect(host=host, user=user, password=password, config=cfg ,port=port)
+        cur=con.cursor()
+        print(cur)
+        return cur
+
+    def create_tables(self,tsql, dbName,vgroups,stbName,ctbNum,rowsPerTbl):
+        tsql.execute("create database if not exists %s vgroups %d"%(dbName, vgroups))        
+        tsql.execute("use %s" %dbName)
+        tsql.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):
+                tsql.execute(sql)
+                sql = pre_create
+        if sql != pre_create:
+            tsql.execute(sql)
+            
+        tdLog.debug("complete to create database[%s], stable[%s] and %d child tables" %(dbName, stbName, ctbNum))
+        return
+
+    def insert_data(self,tsql,dbName,stbName,ctbNum,rowsPerTbl,batchNum,startTs):
+        tdLog.debug("start to insert data ............")
+        tsql.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%batchNum == 0) or (j == rowsPerTbl - 1)):
+                    tsql.execute(sql)
+                    if j < rowsPerTbl - 1:
+                        sql = "insert into %s_%d values " %(stbName,i)
+                    else:
+                        sql = "insert into "
+        #end sql
+        if sql != pre_insert:
+            #print("insert sql:%s"%sql)
+            tsql.execute(sql)
+        tdLog.debug("insert data ............ [OK]")
+        return
+        
+    def prepareEnv(self, **parameterDict):
+        print ("input parameters:")
+        print (parameterDict)
+        # create new connector for my thread
+        tsql=self.newcur(parameterDict['cfg'], 'localhost', 6030)
+        self.create_tables(tsql,\
+                           parameterDict["dbName"],\
+                           parameterDict["vgroups"],\
+                           parameterDict["stbName"],\
+                           parameterDict["ctbNum"],\
+                           parameterDict["rowsPerTbl"])
+
+        self.insert_data(tsql,\
+                         parameterDict["dbName"],\
+                         parameterDict["stbName"],\
+                         parameterDict["ctbNum"],\
+                         parameterDict["rowsPerTbl"],\
+                         parameterDict["batchNum"],\
+                         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 = {'cfg':        '',       \
+                         'dbName':     'db',     \
+                         'vgroups':    1,        \
+                         'stbName':    'stb',    \
+                         'ctbNum':     10,       \
+                         'rowsPerTbl': 100,   \
+                         'batchNum':   10,       \
+                         'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        parameterDict['cfg'] = cfgPath
+        prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
+        prepareEnvThread.start()
+        time.sleep(2)
+        
+        # wait stb ready
+        while 1:
+            tdSql.query("show %s.stables"%parameterDict['dbName'])            
+            if tdSql.getRows() == 1:            
+                break
+            else:
+                time.sleep(1)
+
+        tdLog.info("create topics from super table")
+        topicFromStb = 'topic_stb_column'
+        topicFromCtb = 'topic_ctb_column'        
+        
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb, parameterDict['dbName'], parameterDict['stbName']))
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s_0" %(topicFromCtb, parameterDict['dbName'], parameterDict['stbName']))
+        
+        time.sleep(1)
+        tdSql.query("show topics")
+        #tdSql.checkRows(2)
+        topic1 = tdSql.getData(0 , 0)
+        topic2 = tdSql.getData(1 , 0)
+        print (topic1)
+        print (topic2)
+        
+        print (topicFromStb)
+        print (topicFromCtb)
+        #tdLog.info("show topics: %s, %s"%topic1, topic2)
+        #if topic1 != topicFromStb or topic1 != topicFromCtb:
+        #    tdLog.exit("topic error1") 
+        #if topic2 != topicFromStb or topic2 != topicFromCtb:
+        #    tdLog.exit("topic error2") 
+         
+        tdLog.info("create consume info table and consume result table")
+        cdbName = parameterDict["dbName"]
+        tdSql.query("create table consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)")
+        tdSql.query("create table consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)")
+
+        consumerId   = 0
+        expectmsgcnt = (parameterDict["rowsPerTbl"] / parameterDict["batchNum"] ) * parameterDict["ctbNum"]
+        expectmsgcnt1 = expectmsgcnt + parameterDict["ctbNum"]
+        topicList    = topicFromStb
+        ifcheckdata  = 0
+        keyList      = 'group.id:cgrp1,\
+                        enable.auto.commit:false,\
+                        auto.commit.interval.ms:6000,\
+                        auto.offset.reset:earliest'
+        sql = "insert into consumeinfo values "
+        sql += "(now, %d, '%s', '%s', %d, %d)"%(consumerId, topicList, keyList, expectmsgcnt1, ifcheckdata)
+        tdSql.query(sql)
+        
+        tdLog.info("check stb if there are data")
+        while 1:
+            tdSql.query("select count(*) from %s"%parameterDict["stbName"])
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            countOfStb = tdSql.getData(0, 0)
+            if countOfStb != 0:
+                tdLog.info("count from stb: %d"%countOfStb)
+                break
+            else:
+                time.sleep(1)
+        
+        tdLog.info("start consume processor")
+        pollDelay = 5
+        showMsg   = 1
+        showRow   = 1
+        
+        shellCmd = 'nohup ' + buildPath + '/build/bin/tmq_sim -c ' + cfgPath
+        shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, parameterDict["dbName"], showMsg, showRow, cdbName) 
+        shellCmd += "> /dev/null 2>&1 &"        
+        tdLog.info(shellCmd)
+        os.system(shellCmd)        
+
+        # wait for data ready
+        prepareEnvThread.join()
+        
+        tdLog.info("insert process end, and start to check consume result")
+        while 1:
+            tdSql.query("select * from consumeresult")
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            if tdSql.getRows() == 1:
+                break
+            else:
+                time.sleep(5)
+
+        expectrowcnt = parameterDict["rowsPerTbl"]  * parameterDict["ctbNum"]
+                    
+        tdSql.checkData(0 , 1, consumerId)
+        tdSql.checkData(0 , 2, expectmsgcnt)
+        tdSql.checkData(0 , 3, expectrowcnt)
+
+        tdSql.query("drop topic %s"%topicFromStb)
+        tdSql.query("drop topic %s"%topicFromCtb)
+        
+        # ==============================================================================
+        tdLog.printNoPrefix("======== test scenario 2: add child table with consuming ")
+        tdLog.info(" clean database")        
+        # create and start thread
+        parameterDict = {'cfg':        '',       \
+                         'dbName':     'db2',    \
+                         'vgroups':    1,        \
+                         'stbName':    'stb',    \
+                         'ctbNum':     10,       \
+                         'rowsPerTbl': 10000,   \
+                         'batchNum':   100,       \
+                         'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
+        parameterDict['cfg'] = cfgPath
+
+        prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict)
+        prepareEnvThread.start()
+        
+        # wait db ready
+        while 1:
+            tdSql.query("show databases")
+            if tdSql.getRows() == 4: 
+                print (tdSql.getData(0,0), tdSql.getData(1,0),tdSql.getData(2,0),)           
+                break
+            else:
+                time.sleep(1)
+        
+        tdSql.query("use %s"%parameterDict['dbName'])
+        # wait stb ready
+        while 1:
+            tdSql.query("show %s.stables"%parameterDict['dbName'])
+            if tdSql.getRows() == 1:            
+                break
+            else:
+                time.sleep(1)
+
+        tdLog.info("create topics from super table")
+        topicFromStb = 'topic_stb_column2'
+        topicFromCtb = 'topic_ctb_column2'
+        
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(topicFromStb, parameterDict['dbName'], parameterDict['stbName']))
+        tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s_0" %(topicFromCtb, parameterDict['dbName'], parameterDict['stbName']))
+        
+        time.sleep(1)
+        tdSql.query("show topics")
+        topic1 = tdSql.getData(0 , 0)
+        topic2 = tdSql.getData(1 , 0)
+        print (topic1)
+        print (topic2)
+        
+        print (topicFromStb)
+        print (topicFromCtb)
+        #tdLog.info("show topics: %s, %s"%topic1, topic2)
+        #if topic1 != topicFromStb or topic1 != topicFromCtb:
+        #    tdLog.exit("topic error1") 
+        #if topic2 != topicFromStb or topic2 != topicFromCtb:
+        #    tdLog.exit("topic error2") 
+         
+        tdLog.info("create consume info table and consume result table")
+        cdbName = parameterDict["dbName"]
+        tdSql.query("create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int)"%cdbName)
+        tdSql.query("create table %s.consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"%cdbName)
+
+        consumerId   = 0
+        expectmsgcnt = (parameterDict["rowsPerTbl"] / parameterDict["batchNum"] ) * parameterDict["ctbNum"]
+        expectmsgcnt1 = expectmsgcnt + parameterDict["ctbNum"]
+        topicList    = topicFromStb
+        ifcheckdata  = 0
+        keyList      = 'group.id:cgrp1,\
+                        enable.auto.commit:false,\
+                        auto.commit.interval.ms:6000,\
+                        auto.offset.reset:earliest'
+        sql = "insert into consumeinfo values "
+        sql += "(now, %d, '%s', '%s', %d, %d)"%(consumerId, topicList, keyList, expectmsgcnt1, ifcheckdata)
+        tdSql.query(sql)
+        
+        tdLog.info("check stb if there are data")
+        while 1:
+            tdSql.query("select count(*) from %s"%parameterDict["stbName"])
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            countOfStb = tdSql.getData(0, 0)
+            if countOfStb != 0:
+                tdLog.info("count from stb: %d"%countOfStb)
+                break
+            else:
+                time.sleep(1)
+        
+        tdLog.info("start consume processor")
+        pollDelay = 5
+        showMsg   = 1
+        showRow   = 1
+        
+        shellCmd = 'nohup ' + buildPath + '/build/bin/tmq_sim -c ' + cfgPath
+        shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, parameterDict["dbName"], showMsg, showRow, cdbName) 
+        shellCmd += "> /dev/null 2>&1 &"        
+        tdLog.info(shellCmd)
+        os.system(shellCmd)        
+
+        # create new child table and insert data
+        newCtbName = 'newctb'
+        rowsOfNewCtb = 1000
+        tdSql.query("create table %s.%s using %s.%s tags(9999)"%(parameterDict["dbName"], newCtbName, parameterDict["dbName"], parameterDict["stbName"]))
+        startTs = parameterDict["startTs"]
+        for j in range(rowsOfNewCtb):
+            sql = "insert into %s.%s values (%d, %d, 'tmqrow_%d') "%(parameterDict["dbName"], newCtbName, startTs + j, j, j)
+            tdSql.execute(sql)
+        tdLog.debug("insert data into new child table ............ [OK]")
+
+        # wait for data ready
+        prepareEnvThread.join()
+        
+        tdLog.info("insert process end, and start to check consume result")
+        while 1:
+            tdSql.query("select * from consumeresult")
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            if tdSql.getRows() == 1:
+                break
+            else:
+                time.sleep(5)
+
+        expectmsgcnt += rowsOfNewCtb
+        expectrowcnt = parameterDict["rowsPerTbl"]  * parameterDict["ctbNum"] + rowsOfNewCtb
+                    
+        tdSql.checkData(0 , 1, consumerId)
+        tdSql.checkData(0 , 2, expectmsgcnt)
+        tdSql.checkData(0 , 3, expectrowcnt)
+
+        
+        # ==============================================================================
+        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())

tests/system-test/fulltest.sh

@@ -51,3 +51,7 @@ python3 ./test.py -f 2-query/arcsin.py
 python3 ./test.py -f 2-query/arccos.py
 python3 ./test.py -f 2-query/arctan.py
 # python3 ./test.py -f 2-query/query_cols_tags_and_or.py
+
+python3 ./test.py -f 7-tmq/basic5.py
+
+