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

include/common/tcommon.h

@@ -341,7 +341,7 @@ typedef struct SExprInfo {
 
 typedef struct {
   const char* key;
-  int32_t     keyLen;
+  size_t     keyLen;
   uint8_t     type;
   union {
     const char* value;
@@ -350,7 +350,7 @@ typedef struct {
     double      d;
     float       f;
   };
-  int32_t length;
+  size_t length;
 } SSmlKv;
 
 #define QUERY_ASC_FORWARD_STEP  1

include/common/tdatablock.h

@@ -265,7 +265,7 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag);
 // for debug
 char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf);
 
-int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataBlocks, STSchema* pTSchema, int32_t vgId,
+int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, const STSchema* pTSchema, int64_t uid, int32_t vgId,
                                     tb_uid_t suid);
 
 char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);

include/common/tdataformat.h

@@ -44,18 +44,38 @@ typedef struct SColData SColData;
 #define HAS_VALUE ((uint8_t)0x4)
 
 // bitmap ================================
-const static uint8_t BIT2_MAP[4][4] = {{0b00000000, 0b00000001, 0b00000010, 0},
-                                       {0b00000000, 0b00000100, 0b00001000, 2},
-                                       {0b00000000, 0b00010000, 0b00100000, 4},
-                                       {0b00000000, 0b01000000, 0b10000000, 6}};
-
-#define N1(n)             ((((uint8_t)1) << (n)) - 1)
-#define BIT1_SIZE(n)      ((((n)-1) >> 3) + 1)
-#define BIT2_SIZE(n)      ((((n)-1) >> 2) + 1)
-#define SET_BIT1(p, i, v) ((p)[(i) >> 3] = (p)[(i) >> 3] & N1((i)&7) | (((uint8_t)(v)) << ((i)&7)))
-#define GET_BIT1(p, i)    (((p)[(i) >> 3] >> ((i)&7)) & ((uint8_t)1))
-#define SET_BIT2(p, i, v) ((p)[(i) >> 2] = (p)[(i) >> 2] & N1(BIT2_MAP[(i)&3][3]) | BIT2_MAP[(i)&3][(v)])
-#define GET_BIT2(p, i)    (((p)[(i) >> 2] >> BIT2_MAP[(i)&3][3]) & ((uint8_t)3))
+const static uint8_t BIT1_MAP[8] = {0b11111110, 0b11111101, 0b11111011, 0b11110111,
+                                    0b11101111, 0b11011111, 0b10111111, 0b01111111};
+
+const static uint8_t BIT2_MAP[4] = {0b11111100, 0b11110011, 0b11001111, 0b00111111};
+
+#define ONE               ((uint8_t)1)
+#define THREE             ((uint8_t)3)
+#define DIV_8(i)          ((i) >> 3)
+#define MOD_8(i)          ((i)&7)
+#define DIV_4(i)          ((i) >> 2)
+#define MOD_4(i)          ((i)&3)
+#define MOD_4_TIME_2(i)   (MOD_4(i) << 1)
+#define BIT1_SIZE(n)      (DIV_8((n)-1) + 1)
+#define BIT2_SIZE(n)      (DIV_4((n)-1) + 1)
+#define SET_BIT1(p, i, v) ((p)[DIV_8(i)] = (p)[DIV_8(i)] & BIT1_MAP[MOD_8(i)] | ((v) << MOD_8(i)))
+#define SET_BIT1_EX(p, i, v) \
+  do {                       \
+    if (MOD_8(i) == 0) {     \
+      (p)[DIV_8(i)] = 0;     \
+    }                        \
+    SET_BIT1(p, i, v);       \
+  } while (0)
+#define GET_BIT1(p, i)    (((p)[DIV_8(i)] >> MOD_8(i)) & ONE)
+#define SET_BIT2(p, i, v) ((p)[DIV_4(i)] = (p)[DIV_4(i)] & BIT2_MAP[MOD_4(i)] | ((v) << MOD_4_TIME_2(i)))
+#define SET_BIT2_EX(p, i, v) \
+  do {                       \
+    if (MOD_4(i) == 0) {     \
+      (p)[DIV_4(i)] = 0;     \
+    }                        \
+    SET_BIT2(p, i, v);       \
+  } while (0)
+#define GET_BIT2(p, i) (((p)[DIV_4(i)] >> MOD_4_TIME_2(i)) & THREE)
 
 // SBuffer ================================
 struct SBuffer {
@@ -70,9 +90,6 @@ int32_t tBufferInit(SBuffer *pBuffer, int64_t size);
 int32_t tBufferPut(SBuffer *pBuffer, const void *pData, int64_t nData);
 int32_t tBufferReserve(SBuffer *pBuffer, int64_t nData, void **ppData);
 
-// STSchema ================================
-void tDestroyTSchema(STSchema *pTSchema);
-
 // SColVal ================================
 #define CV_FLAG_VALUE ((int8_t)0x0)
 #define CV_FLAG_NONE  ((int8_t)0x1)
@@ -87,8 +104,12 @@ void tDestroyTSchema(STSchema *pTSchema);
 #define COL_VAL_IS_VALUE(CV) ((CV)->flag == CV_FLAG_VALUE)
 
 // SRow ================================
-int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SBuffer *pBuffer);
+int32_t tRowBuild(SArray *aColVal, const STSchema *pTSchema, SRow **ppRow);
 void    tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
+void    tRowDestroy(SRow *pRow);
+void    tRowSort(SArray *aRowP);
+int32_t tRowMerge(SArray *aRowP, STSchema *pTSchema, int8_t flag);
+int32_t tRowAppendToColData(SRow *pRow, STSchema *pTSchema, SColData *aColData, int32_t nColData);
 
 // SRowIter ================================
 int32_t  tRowIterOpen(SRow *pRow, STSchema *pTSchema, SRowIter **ppIter);
@@ -110,15 +131,28 @@ void    debugPrintSTag(STag *pTag, const char *tag, int32_t ln);  // TODO: remov
 int32_t parseJsontoTagData(const char *json, SArray *pTagVals, STag **ppTag, void *pMsgBuf);
 
 // SColData ================================
+typedef void *(*xMallocFn)(void *, int32_t);
 void    tColDataDestroy(void *ph);
 void    tColDataInit(SColData *pColData, int16_t cid, int8_t type, int8_t smaOn);
 void    tColDataClear(SColData *pColData);
+void    tColDataDeepClear(SColData *pColData);
 int32_t tColDataAppendValue(SColData *pColData, SColVal *pColVal);
 void    tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal);
 uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal);
-int32_t tColDataCopy(SColData *pColDataSrc, SColData *pColDataDest);
+int32_t tColDataCopy(SColData *pColDataFrom, SColData *pColData, xMallocFn xMalloc, void *arg);
 extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_t *min, int16_t *numOfNull);
 
+// for stmt bind
+int32_t tColDataAddValueByBind(SColData *pColData, TAOS_MULTI_BIND *pBind);
+void    tColDataSortMerge(SArray *colDataArr);
+
+//for raw block
+int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes,
+                                    int32_t nRows, char* lengthOrbitmap, char *data);
+// for encode/decode
+int32_t tPutColData(uint8_t *pBuf, SColData *pColData);
+int32_t tGetColData(uint8_t *pBuf, SColData *pColData);
+
 // STRUCT ================================
 struct STColumn {
   col_id_t colId;
@@ -225,23 +259,9 @@ struct STag {
     memcpy(varDataVal(x), (str), (_size));      \
   } while (0);
 
-// ----------------- SCHEMA BUILDER DEFINITION
-typedef struct {
-  int32_t      tCols;
-  int32_t      nCols;
-  schema_ver_t version;
-  uint16_t     flen;
-  int32_t      tlen;
-  STColumn    *columns;
-} STSchemaBuilder;
-
-int32_t   tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
-void      tdDestroyTSchemaBuilder(STSchemaBuilder *pBuilder);
-void      tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
-int32_t   tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags, col_id_t colId, col_bytes_t bytes);
-STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
-
+// STSchema ================================
 STSchema *tBuildTSchema(SSchema *aSchema, int32_t numOfCols, int32_t version);
+void      tDestroyTSchema(STSchema *pTSchema);
 
 #endif
 

include/common/tmsg.h

@@ -482,8 +482,6 @@ static FORCE_INLINE int32_t tDecodeSSchemaWrapperEx(SDecoder* pDecoder, SSchemaW
   return 0;
 }
 
-STSchema* tdGetSTSChemaFromSSChema(SSchema* pSchema, int32_t nCols, int32_t sver);
-
 typedef struct {
   char     name[TSDB_TABLE_FNAME_LEN];
   int8_t   igExists;
@@ -1734,6 +1732,8 @@ typedef struct {
   int32_t  execId;
 } STaskDropReq;
 
+int32_t tSerializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
+int32_t tDeserializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
 int32_t tSerializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
 int32_t tDeserializeSTaskDropReq(void* buf, int32_t bufLen, STaskDropReq* pReq);
 
@@ -2083,8 +2083,13 @@ typedef struct SVCreateTbReq {
 
 int  tEncodeSVCreateTbReq(SEncoder* pCoder, const SVCreateTbReq* pReq);
 int  tDecodeSVCreateTbReq(SDecoder* pCoder, SVCreateTbReq* pReq);
+void tDestroySVCreateTbReq(SVCreateTbReq* pReq, int32_t flags);
 
 static FORCE_INLINE void tdDestroySVCreateTbReq(SVCreateTbReq* req) {
+  if (NULL == req) {
+    return;
+  }
+
   taosMemoryFreeClear(req->name);
   taosMemoryFreeClear(req->comment);
   if (req->type == TSDB_CHILD_TABLE) {
@@ -3232,6 +3237,57 @@ int32_t tSerializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
 int32_t tDeserializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
 int32_t tSerializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
 int32_t tDeserializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
+int32_t tSerializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
+int32_t tDeserializeSMqAskEpReq(void* buf, int32_t bufLen, SMqAskEpReq* pReq);
+int32_t tSerializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
+int32_t tDeserializeSMqHbReq(void* buf, int32_t bufLen, SMqHbReq* pReq);
+
+#define SUBMIT_REQ_AUTO_CREATE_TABLE  0x1
+#define SUBMIT_REQ_COLUMN_DATA_FORMAT 0x2
+
+typedef struct {
+  int32_t        flags;
+  SVCreateTbReq* pCreateTbReq;
+  int64_t        suid;
+  int64_t        uid;
+  int32_t        sver;
+  union {
+    SArray* aRowP;
+    SArray* aCol;
+  };
+} SSubmitTbData;
+
+typedef struct {
+  SArray* aSubmitTbData;  // SArray<SSubmitTbData>
+} SSubmitReq2;
+
+int32_t tEncodeSSubmitReq2(SEncoder* pCoder, const SSubmitReq2* pReq);
+int32_t tDecodeSSubmitReq2(SDecoder* pCoder, SSubmitReq2* pReq);
+void    tDestroySSubmitTbData(SSubmitTbData* pTbData, int32_t flag);
+void    tDestroySSubmitReq2(SSubmitReq2* pReq, int32_t flag);
+
+typedef struct {
+  int32_t affectedRows;
+  SArray* aCreateTbRsp;  // SArray<SVCreateTbRsp>
+} SSubmitRsp2;
+
+int32_t tEncodeSSubmitRsp2(SEncoder* pCoder, const SSubmitRsp2* pRsp);
+int32_t tDecodeSSubmitRsp2(SDecoder* pCoder, SSubmitRsp2* pRsp);
+void    tDestroySSubmitRsp2(SSubmitRsp2* pRsp, int32_t flag);
+
+#define TSDB_MSG_FLG_ENCODE 0x1
+#define TSDB_MSG_FLG_DECODE 0x2
+
+typedef struct {
+  union {
+    struct {
+      void*   msgStr;
+      int32_t msgLen;
+      int64_t ver;
+    };
+    void* pDataBlock;
+  };
+} SPackedData;
 
 #pragma pack(pop)
 

include/libs/executor/executor.h

@@ -190,7 +190,9 @@ int32_t qStreamPrepareTsdbScan(qTaskInfo_t tinfo, uint64_t uid, int64_t ts);
 
 int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subType);
 
-int32_t qStreamScanMemData(qTaskInfo_t tinfo, const SSubmitReq* pReq);
+// int32_t qStreamScanMemData(qTaskInfo_t tinfo, const SSubmitReq* pReq, int64_t ver);
+//
+int32_t qStreamSetScanMemData(qTaskInfo_t tinfo, SPackedData submit);
 
 int32_t qStreamExtractOffset(qTaskInfo_t tinfo, STqOffsetVal* pOffset);
 

include/libs/nodes/querynodes.h

@@ -361,8 +361,8 @@ typedef struct SVgDataBlocks {
   void*       pData;  // SSubmitReq + SSubmitBlk + ...
 } SVgDataBlocks;
 
-typedef void (*FFreeDataBlockHash)(SHashObj*);
-typedef void (*FFreeDataBlockArray)(SArray*);
+typedef void (*FFreeTableBlockHash)(SHashObj*);
+typedef void (*FFreeVgourpBlockArray)(SArray*);
 
 typedef struct SVnodeModifyOpStmt {
   ENodeType             nodeType;
@@ -377,15 +377,15 @@ typedef struct SVnodeModifyOpStmt {
   const char*           pBoundCols;
   struct STableMeta*    pTableMeta;
   SHashObj*             pVgroupsHashObj;
-  SHashObj*           pTableBlockHashObj;
+  SHashObj*             pTableBlockHashObj;  // SHashObj<tuid, STableDataCxt*>
   SHashObj*             pSubTableHashObj;
   SHashObj*             pTableNameHashObj;
   SHashObj*             pDbFNameHashObj;
-  SArray*             pVgDataBlocks;
-  SVCreateTbReq       createTblReq;
+  SArray*               pVgDataBlocks;  // SArray<SVgroupDataCxt*>
+  SVCreateTbReq*        pCreateTblReq;
   TdFilePtr             fp;
-  FFreeDataBlockHash  freeHashFunc;
-  FFreeDataBlockArray freeArrayFunc;
+  FFreeTableBlockHash   freeHashFunc;
+  FFreeVgourpBlockArray freeArrayFunc;
   bool                  usingTableProcessing;
   bool                  fileProcessing;
 } SVnodeModifyOpStmt;

include/libs/parser/parser.h

@@ -58,7 +58,6 @@ typedef struct SParseContext {
   bool             isSuperUser;
   bool             enableSysInfo;
   bool             async;
-  int8_t           schemalessType;
   const char*      svrVer;
   bool             nodeOffline;
   SArray*          pTableMetaPos;    // sql table pos => catalog data pos
@@ -85,12 +84,12 @@ int32_t qSetSTableIdForRsma(SNode* pStmt, int64_t uid);
 void    qCleanupKeywordsTable();
 
 int32_t     qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash);
-int32_t     qResetStmtDataBlock(void* block, bool keepBuf);
-int32_t     qCloneStmtDataBlock(void** pDst, void* pSrc);
-void        qFreeStmtDataBlock(void* pDataBlock);
-int32_t     qRebuildStmtDataBlock(void** pDst, void* pSrc, uint64_t uid, int32_t vgId);
-void        qDestroyStmtDataBlock(void* pBlock);
-STableMeta* qGetTableMetaInDataBlock(void* pDataBlock);
+int32_t     qResetStmtDataBlock(STableDataCxt* block, bool keepBuf);
+int32_t     qCloneStmtDataBlock(STableDataCxt** pDst, STableDataCxt* pSrc, bool reset);
+int32_t     qRebuildStmtDataBlock(STableDataCxt** pDst, STableDataCxt* pSrc, uint64_t uid, uint64_t suid, int32_t vgId, bool rebuildCreateTb);
+void        qDestroyStmtDataBlock(STableDataCxt* pBlock);
+STableMeta* qGetTableMetaInDataBlock(STableDataCxt* pDataBlock);
+int32_t qCloneCurrentTbData(STableDataCxt*       pDataBlock, SSubmitTbData **pData);
 
 int32_t qStmtBindParams(SQuery* pQuery, TAOS_MULTI_BIND* pParams, int32_t colIdx);
 int32_t qStmtParseQuerySql(SParseContext* pCxt, SQuery* pQuery);
@@ -105,11 +104,18 @@ void    destroyBoundColumnInfo(void* pBoundInfo);
 int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,
                      int32_t msgBufLen);
 
-void*   smlInitHandle(SQuery* pQuery);
-void    smlDestroyHandle(void* pHandle);
-int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
+void    qDestroyBoundColInfo(void* pInfo);
+
+SQuery* smlInitHandle();
+int32_t smlBuildRow(STableDataCxt* pTableCxt);
+int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void *kv, int32_t index);
+STableDataCxt* smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta);
+
+int32_t smlBindData(SQuery* handle, bool dataFormat, SArray* tags, SArray* colsSchema, SArray* cols, STableMeta* pTableMeta,
                     char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl, char* msgBuf, int16_t msgBufLen);
-int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
+int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash);
+
+int     rawBlockBindData(SQuery *query, STableMeta* pTableMeta, void* data, SVCreateTbReq* pCreateTb, TAOS_FIELD *fields, int numFields);
 
 int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray);
 SArray* serializeVgroupsCreateTableBatch(SHashObj* pVgroupHashmap);

include/libs/qcom/query.h

@@ -163,6 +163,23 @@ typedef struct STargetInfo {
   int32_t     vgId;
 } STargetInfo;
 
+typedef struct SBoundColInfo {
+  int16_t* pColIndex;  // bound index => schema index
+  int32_t  numOfCols;
+  int32_t  numOfBound;
+} SBoundColInfo;
+
+typedef struct STableDataCxt {
+  STableMeta*    pMeta;
+  STSchema*      pSchema;
+  SBoundColInfo  boundColsInfo;
+  SArray*        pValues;
+  SSubmitTbData* pData;
+  TSKEY          lastTs;
+  bool           ordered;
+  bool           duplicateTs;
+} STableDataCxt;
+
 typedef int32_t (*__async_send_cb_fn_t)(void* param, SDataBuf* pMsg, int32_t code);
 typedef int32_t (*__async_exec_fn_t)(void* param);
 
@@ -238,6 +255,7 @@ int32_t dataConverToStr(char* str, int type, void* buf, int32_t bufSize, int32_t
 char*   parseTagDatatoJson(void* p);
 int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst);
 int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst);
+int32_t cloneSVreateTbReq(SVCreateTbReq* pSrc, SVCreateTbReq** pDst);
 void    freeVgInfo(SDBVgInfo* vgInfo);
 
 extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char** msg, int32_t msgSize, int32_t* msgLen,

include/libs/stream/tstream.h

@@ -103,6 +103,7 @@ typedef struct {
   int8_t type;
 } SStreamQueueItem;
 
+#if 0
 typedef struct {
   int8_t      type;
   int64_t     ver;
@@ -116,6 +117,21 @@ typedef struct {
   SArray* dataRefs;  // SArray<int32_t*>
   SArray* reqs;      // SArray<SSubmitReq*>
 } SStreamMergedSubmit;
+#endif
+
+typedef struct {
+  int8_t      type;
+  int64_t     ver;
+  int32_t*    dataRef;
+  SPackedData submit;
+} SStreamDataSubmit2;
+
+typedef struct {
+  int8_t  type;
+  int64_t ver;
+  SArray* dataRefs;  // SArray<int32_t*>
+  SArray* submits;   // SArray<SPackedSubmit>
+} SStreamMergedSubmit2;
 
 typedef struct {
   int8_t type;
@@ -219,11 +235,11 @@ static FORCE_INLINE void* streamQueueNextItem(SStreamQueue* queue) {
   }
 }
 
-SStreamDataSubmit* streamDataSubmitNew(SSubmitReq* pReq);
+SStreamDataSubmit2* streamDataSubmitNew(SPackedData submit);
 
-void streamDataSubmitRefDec(SStreamDataSubmit* pDataSubmit);
+void streamDataSubmitRefDec(SStreamDataSubmit2* pDataSubmit);
 
-SStreamDataSubmit* streamSubmitRefClone(SStreamDataSubmit* pSubmit);
+SStreamDataSubmit2* streamSubmitRefClone(SStreamDataSubmit2* pSubmit);
 
 typedef struct {
   char* qmsg;
@@ -355,14 +371,15 @@ void         tFreeSStreamTask(SStreamTask* pTask);
 
 static FORCE_INLINE int32_t streamTaskInput(SStreamTask* pTask, SStreamQueueItem* pItem) {
   if (pItem->type == STREAM_INPUT__DATA_SUBMIT) {
-    SStreamDataSubmit* pSubmitClone = streamSubmitRefClone((SStreamDataSubmit*)pItem);
+    SStreamDataSubmit2* pSubmitClone = streamSubmitRefClone((SStreamDataSubmit2*)pItem);
     if (pSubmitClone == NULL) {
       qDebug("task %d %p submit enqueue failed since out of memory", pTask->taskId, pTask);
       terrno = TSDB_CODE_OUT_OF_MEMORY;
       atomic_store_8(&pTask->inputStatus, TASK_INPUT_STATUS__FAILED);
       return -1;
     }
-    qDebug("task %d %p submit enqueue %p %p %p", pTask->taskId, pTask, pItem, pSubmitClone, pSubmitClone->data);
+    qDebug("task %d %p submit enqueue %p %p %p %d %" PRId64, pTask->taskId, pTask, pItem, pSubmitClone,
+           pSubmitClone->submit.msgStr, pSubmitClone->submit.msgLen, pSubmitClone->submit.ver);
     taosWriteQitem(pTask->inputQueue->queue, pSubmitClone);
     // qStreamInput(pTask->exec.executor, pSubmitClone);
   } else if (pItem->type == STREAM_INPUT__DATA_BLOCK || pItem->type == STREAM_INPUT__DATA_RETRIEVE ||

include/util/tRealloc.h

@@ -52,11 +52,13 @@ _exit:
   return code;
 }
 
-static FORCE_INLINE void tFree(uint8_t *pBuf) {
-  if (pBuf) {
-    taosMemoryFree(pBuf - sizeof(int64_t));
-  }
-}
+#define tFree(BUF)                                        \
+  do {                                                    \
+    if (BUF) {                                            \
+      taosMemoryFree((uint8_t *)(BUF) - sizeof(int64_t)); \
+      (BUF) = NULL;                                       \
+    }                                                     \
+  } while (0)
 
 #ifdef __cplusplus
 }

include/util/taoserror.h

@@ -156,6 +156,7 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_TSC_QUERY_KILLED              TAOS_DEF_ERROR_CODE(0, 0X022D)
 #define TSDB_CODE_TSC_NO_EXEC_NODE              TAOS_DEF_ERROR_CODE(0, 0X022E)
 #define TSDB_CODE_TSC_NOT_STABLE_ERROR          TAOS_DEF_ERROR_CODE(0, 0X022F)
+#define TSDB_CODE_TSC_STMT_CACHE_ERROR          TAOS_DEF_ERROR_CODE(0, 0X0230)
 
 // mnode-common
 // #define TSDB_CODE_MND_MSG_NOT_PROCESSED      TAOS_DEF_ERROR_CODE(0, 0x0300) // 2.x

include/util/tarray.h

@@ -22,19 +22,6 @@
 extern "C" {
 #endif
 
-#if 0
-#define TARRAY(TYPE)             \
-  struct {                       \
-    int32_t      tarray_size_;   \
-    int32_t      tarray_neles_;  \
-    struct TYPE* td_array_data_; \
-  }
-
-#define TARRAY_SIZE(ARRAY)        (ARRAY)->tarray_size_
-#define TARRAY_NELES(ARRAY)       (ARRAY)->tarray_neles_
-#define TARRAY_ELE_AT(ARRAY, IDX) ((ARRAY)->td_array_data_ + idx)
-#endif
-
 #define TARRAY_MIN_SIZE               8
 #define TARRAY_GET_ELEM(array, index) ((void*)((char*)((array)->pData) + (index) * (array)->elemSize))
 #define TARRAY_ELEM_IDX(array, ele)   (POINTER_DISTANCE(ele, (array)->pData) / (array)->elemSize)
@@ -46,6 +33,9 @@ typedef struct SArray {
   void*    pData;
 } SArray;
 
+#define TARRAY_SIZE(array) ((array)->size)
+#define TARRAY_DATA(array) ((array)->pData)
+
 /**
  *
  * @param size
@@ -194,6 +184,13 @@ void taosArrayPopTailBatch(SArray* pArray, size_t cnt);
  */
 void taosArrayRemove(SArray* pArray, size_t index);
 
+/**
+ * remove batch entry from the given index
+ * @param pArray
+ * @param index
+ */
+void taosArrayRemoveBatch(SArray* pArray, size_t index, size_t num, FDelete fp);
+
 /**
  * copy the whole array from source to destination
  * @param pDst

include/util/tdef.h

@@ -193,6 +193,7 @@ typedef enum ELogicConditionType {
 #define TSDB_TABLE_NAME_LEN  193                                // it is a null-terminated string
 #define TSDB_TOPIC_NAME_LEN  193                                // it is a null-terminated string
 #define TSDB_CGROUP_LEN      193                                // it is a null-terminated string
+#define TSDB_USER_CGROUP_LEN (TSDB_USER_LEN + TSDB_CGROUP_LEN)  // it is a null-terminated string
 #define TSDB_DB_NAME_LEN     65
 #define TSDB_DB_FNAME_LEN    (TSDB_ACCT_ID_LEN + TSDB_DB_NAME_LEN + TSDB_NAME_DELIMITER_LEN)
 

include/util/tencode.h

@@ -116,6 +116,7 @@ static int32_t tEncodeI64v(SEncoder* pCoder, int64_t val);
 static int32_t tEncodeFloat(SEncoder* pCoder, float val);
 static int32_t tEncodeDouble(SEncoder* pCoder, double val);
 static int32_t tEncodeBinary(SEncoder* pCoder, const uint8_t* val, uint32_t len);
+static int32_t tEncodeBinaryEx(SEncoder* pCoder, const uint8_t* val, uint32_t len);
 static int32_t tEncodeCStrWithLen(SEncoder* pCoder, const char* val, uint32_t len);
 static int32_t tEncodeCStr(SEncoder* pCoder, const char* val);
 

source/client/inc/clientInt.h

@@ -149,7 +149,6 @@ typedef struct STscObj {
   int32_t       numOfReqs;  // number of sqlObj bound to this connection
   SAppInstInfo* pAppInfo;
   SHashObj*     pRequests;
-  int8_t        schemalessType;  // todo remove it, this attribute should be move to request
 } STscObj;
 
 typedef struct SResultColumn {

source/client/inc/clientLog.h

@@ -30,7 +30,7 @@ extern "C" {
 #define tscDebug(...)  do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLog("TSC ", DEBUG_DEBUG, cDebugFlag, __VA_ARGS__); }} while(0)
 #define tscTrace(...)  do { if (cDebugFlag & DEBUG_TRACE) { taosPrintLog("TSC ", DEBUG_TRACE, cDebugFlag, __VA_ARGS__); }} while(0)
 #define tscDebugL(...) do { if (cDebugFlag & DEBUG_DEBUG) { taosPrintLongString("TSC ", DEBUG_DEBUG, cDebugFlag, __VA_ARGS__); }} while(0)
-//#define tscPerf(...)   do { if (cDebugFlag & DEBUG_INFO)  { taosPrintLog("TSC ", DEBUG_INFO, cDebugFlag, __VA_ARGS__); }} while(0)
+#define tscPerf(...)   do { if (cDebugFlag & DEBUG_INFO)  { taosPrintLog("TSC ", 0, cDebugFlag, __VA_ARGS__); }} while(0)
 // clang-format on
 
 #ifdef __cplusplus

source/client/inc/clientSml.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+#ifndef TDENGINE_CLIENTSML_H
+#define TDENGINE_CLIENTSML_H
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "catalog.h"
+#include "clientInt.h"
+#include "osThread.h"
+#include "query.h"
+#include "taos.h"
+#include "taoserror.h"
+#include "tcommon.h"
+#include "tdef.h"
+#include "tglobal.h"
+#include "tlog.h"
+#include "tmsg.h"
+#include "tname.h"
+#include "ttime.h"
+#include "ttypes.h"
+#include "cJSON.h"
+
+#if (defined(__GNUC__) && (__GNUC__ >= 3)) || (defined(__INTEL_COMPILER) && (__INTEL_COMPILER >= 800)) || defined(__clang__)
+#  define expect(expr,value)    (__builtin_expect ((expr),(value)) )
+#else
+#  define expect(expr,value)    (expr)
+#endif
+
+#ifndef likely
+#define likely(expr)     expect((expr) != 0, 1)
+#endif
+#ifndef unlikely
+#define unlikely(expr)   expect((expr) != 0, 0)
+#endif
+
+#define SPACE ' '
+#define COMMA ','
+#define EQUAL '='
+#define QUOTE '"'
+#define SLASH '\\'
+
+#define JUMP_SPACE(sql, sqlEnd) \
+  while (sql < sqlEnd) {        \
+    if (unlikely(*sql == SPACE))          \
+      sql++;                    \
+    else                        \
+      break;                    \
+  }
+
+#define IS_INVALID_COL_LEN(len)   ((len) <= 0 || (len) >= TSDB_COL_NAME_LEN)
+#define IS_INVALID_TABLE_LEN(len) ((len) <= 0 || (len) >= TSDB_TABLE_NAME_LEN)
+
+#define TS        "_ts"
+#define TS_LEN    3
+#define VALUE     "_value"
+#define VALUE_LEN 6
+
+#define MAX_RETRY_TIMES 5
+typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;
+
+typedef enum {
+  SCHEMA_ACTION_NULL,
+  SCHEMA_ACTION_CREATE_STABLE,
+  SCHEMA_ACTION_ADD_COLUMN,
+  SCHEMA_ACTION_ADD_TAG,
+  SCHEMA_ACTION_CHANGE_COLUMN_SIZE,
+  SCHEMA_ACTION_CHANGE_TAG_SIZE,
+} ESchemaAction;
+
+typedef struct {
+  const void  *key;
+  int32_t      keyLen;
+  void        *value;
+  bool         used;
+}Node;
+
+typedef struct NodeList{
+  Node             data;
+  struct NodeList* next;
+}NodeList;
+
+typedef struct {
+  char *measure;
+  char *tags;
+  char *cols;
+  char *timestamp;
+
+  int32_t measureLen;
+  int32_t measureTagsLen;
+  int32_t tagsLen;
+  int32_t colsLen;
+  int32_t timestampLen;
+
+  SArray *colArray;
+} SSmlLineInfo;
+
+typedef struct {
+  const char *sTableName;  // super table name
+  int32_t     sTableNameLen;
+  char        childTableName[TSDB_TABLE_NAME_LEN];
+  uint64_t    uid;
+  void       *key;        // for openTsdb
+
+  SArray *tags;
+
+  // elements are SHashObj<cols key string, SSmlKv*> for find by key quickly
+  SArray *cols;
+  STableDataCxt *tableDataCtx;
+} SSmlTableInfo;
+
+typedef struct {
+  SArray   *tags;     // save the origin order to create table
+  SHashObj *tagHash;  // elements are <key, index in tags>
+
+  SArray   *cols;
+  SHashObj *colHash;
+
+  STableMeta *tableMeta;
+} SSmlSTableMeta;
+
+typedef struct {
+  int32_t len;
+  char   *buf;
+} SSmlMsgBuf;
+
+typedef struct {
+  int32_t code;
+  int32_t lineNum;
+
+  int32_t numOfSTables;
+  int32_t numOfCTables;
+  int32_t numOfCreateSTables;
+  int32_t numOfAlterColSTables;
+  int32_t numOfAlterTagSTables;
+
+  int64_t parseTime;
+  int64_t schemaTime;
+  int64_t insertBindTime;
+  int64_t insertRpcTime;
+  int64_t endTime;
+} SSmlCostInfo;
+
+typedef struct {
+  int64_t id;
+
+  SMLProtocolType protocol;
+  int8_t          precision;
+  bool            reRun;
+  bool            dataFormat;  // true means that the name and order of keys in each line are the same(only for influx protocol)
+  bool            isRawLine;
+  int32_t         ttl;
+
+  NodeList *childTables;
+  NodeList *superTables;
+  SHashObj *pVgHash;
+
+  STscObj     *taos;
+  SCatalog    *pCatalog;
+  SRequestObj *pRequest;
+  SQuery      *pQuery;
+
+  SSmlCostInfo cost;
+  int32_t      lineNum;
+  SSmlMsgBuf   msgBuf;
+
+//  cJSON       *root;  // for parse json
+  int8_t             offset[4];
+  SSmlLineInfo      *lines; // element is SSmlLineInfo
+
+  //
+  SArray      *preLineTagKV;
+  SArray      *preLineColKV;
+
+  SSmlLineInfo preLine;
+  STableMeta  *currSTableMeta;
+  STableDataCxt *currTableDataCtx;
+  bool         needModifySchema;
+} SSmlHandle;
+
+#define IS_SAME_CHILD_TABLE (elements->measureTagsLen == info->preLine.measureTagsLen \
+&& memcmp(elements->measure, info->preLine.measure, elements->measureTagsLen) == 0)
+
+#define IS_SAME_SUPER_TABLE (elements->measureLen == info->preLine.measureLen \
+&& memcmp(elements->measure, info->preLine.measure, elements->measureLen) == 0)
+
+#define IS_SAME_KEY (preKV->keyLen == kv.keyLen && memcmp(preKV->key, kv.key, kv.keyLen) == 0)
+
+extern int64_t smlFactorNS[3];
+extern int64_t smlFactorS[3];
+
+typedef int32_t (*_equal_fn_sml)(const void *, const void *);
+
+SSmlHandle   *smlBuildSmlInfo(TAOS *taos);
+void          smlDestroyInfo(SSmlHandle *info);
+void          smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset);
+void          smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset);
+SArray       *smlJsonParseTags(char *start, char *end);
+bool          smlParseNumberOld(SSmlKv *kvVal, SSmlMsgBuf *msg);
+void*         nodeListGet(NodeList* list, const void *key, int32_t len, _equal_fn_sml fn);
+int           nodeListSet(NodeList** list, const void *key, int32_t len, void* value, _equal_fn_sml fn);
+int           nodeListSize(NodeList* list);
+bool          smlDoubleToInt64OverFlow(double num);
+int32_t       smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2);
+bool          smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg);
+int64_t       smlGetTimeValue(const char *value, int32_t len, uint8_t fromPrecision, uint8_t toPrecision);
+int8_t        smlGetTsTypeByLen(int32_t len);
+SSmlTableInfo*    smlBuildTableInfo(int numRows, const char* measure, int32_t measureLen);
+SSmlSTableMeta*   smlBuildSTableMeta(bool isDataFormat);
+int32_t           smlSetCTableName(SSmlTableInfo *oneTable);
+STableMeta*       smlGetMeta(SSmlHandle *info, const void* measure, int32_t measureLen);
+int32_t           is_same_child_table_telnet(const void *a, const void *b);
+int64_t           smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len);
+int32_t           smlClearForRerun(SSmlHandle *info);
+int32_t           smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg);
+
+int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
+int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
+int32_t smlParseJSON(SSmlHandle *info, char *payload);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif  // TDENGINE_CLIENTSML_H

source/client/inc/clientStmt.h

@@ -21,8 +21,6 @@ extern "C" {
 #endif
 #include "catalog.h"
 
-typedef void STableDataBlocks;
-
 typedef enum {
   STMT_TYPE_INSERT = 1,
   STMT_TYPE_MULTI_INSERT,
@@ -43,7 +41,7 @@ typedef enum {
 } STMT_STATUS;
 
 typedef struct SStmtTableCache {
-  STableDataBlocks *pDataBlock;
+  STableDataCxt *pDataCtx;
   void          *boundTags;
 } SStmtTableCache;
 
@@ -74,7 +72,8 @@ typedef struct SStmtExecInfo {
   int32_t        affectedRows;
   SRequestObj   *pRequest;
   SHashObj      *pBlockHash;
-  bool         autoCreateTbl;
+  STableDataCxt *pCurrBlock;
+  SSubmitTbData *pCurrTbData;
 } SStmtExecInfo;
 
 typedef struct SStmtSQLInfo {

source/client/src/clientEnv.c

@@ -76,13 +76,19 @@ static void deregisterRequest(SRequestObj *pRequest) {
            "current:%d, app current:%d",
            pRequest->self, pTscObj->id, pRequest->requestId, duration / 1000.0, num, currentInst);
 
+  tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
+          "us, exec:%" PRId64 "us, stmtType:%d",
+          duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
+          pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd - pRequest->metric.ctgEnd,
+          pRequest->metric.execEnd - pRequest->metric.semanticEnd, pRequest->stmtType);
+
   if (QUERY_NODE_VNODE_MODIFY_STMT == pRequest->stmtType) {
     //        tscPerf("insert duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
     //                "us, exec:%" PRId64 "us",
     //                duration, pRequest->metric.syntaxEnd - pRequest->metric.syntaxStart,
     //                pRequest->metric.ctgEnd - pRequest->metric.ctgStart, pRequest->metric.semanticEnd -
     //                pRequest->metric.ctgEnd, pRequest->metric.execEnd - pRequest->metric.semanticEnd);
-    atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
+    //    atomic_add_fetch_64((int64_t *)&pActivity->insertElapsedTime, duration);
   } else if (QUERY_NODE_SELECT_STMT == pRequest->stmtType) {
     //    tscPerf("select duration %" PRId64 "us: syntax:%" PRId64 "us, ctg:%" PRId64 "us, semantic:%" PRId64
     //            "us, planner:%" PRId64 "us, exec:%" PRId64 "us, reqId:0x%" PRIx64,
@@ -264,7 +270,6 @@ void *createTscObj(const char *user, const char *auth, const char *db, int32_t c
 
   taosThreadMutexInit(&pObj->mutex, NULL);
   pObj->id = taosAddRef(clientConnRefPool, pObj);
-  pObj->schemalessType = 1;
 
   atomic_add_fetch_64(&pObj->pAppInfo->numOfConns, 1);
 

source/client/src/clientImpl.c

@@ -239,7 +239,6 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
                        .pTransporter = pTscObj->pAppInfo->pTransporter,
                        .pStmtCb = pStmtCb,
                        .pUser = pTscObj->user,
-                       .schemalessType = pTscObj->schemalessType,
                        .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
                        .enableSysInfo = pTscObj->sysInfo,
                        .svrVer = pTscObj->sVer,
@@ -741,47 +740,21 @@ int32_t scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList
 }
 
 int32_t handleSubmitExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {
-  int32_t     code = 0;
   SArray*      pArray = NULL;
-  SSubmitRsp* pRsp = (SSubmitRsp*)res;
-  if (pRsp->nBlocks <= 0) {
-    taosMemoryFreeClear(pRsp->pBlocks);
+  SSubmitRsp2* pRsp = (SSubmitRsp2*)res;
+  if (NULL == pRsp->aCreateTbRsp) {
     return TSDB_CODE_SUCCESS;
   }
 
-  pArray = taosArrayInit(pRsp->nBlocks, sizeof(STbSVersion));
-  if (NULL == pArray) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return TSDB_CODE_OUT_OF_MEMORY;
-  }
-
-  for (int32_t i = 0; i < pRsp->nBlocks; ++i) {
-    SSubmitBlkRsp* blk = pRsp->pBlocks + i;
-    if (blk->pMeta) {
-      handleCreateTbExecRes(blk->pMeta, pCatalog);
-      tFreeSTableMetaRsp(blk->pMeta);
-      taosMemoryFreeClear(blk->pMeta);
-    }
-
-    if (NULL == blk->tblFName || 0 == blk->tblFName[0]) {
-      continue;
+  int32_t tbNum = taosArrayGetSize(pRsp->aCreateTbRsp);
+  for (int32_t i = 0; i < tbNum; ++i) {
+    SVCreateTbRsp* pTbRsp = (SVCreateTbRsp*)taosArrayGet(pRsp->aCreateTbRsp, i);
+    if (pTbRsp->pMeta) {
+      handleCreateTbExecRes(pTbRsp->pMeta, pCatalog);
     }
-
-    STbSVersion tbSver = {.tbFName = blk->tblFName, .sver = blk->sver};
-    taosArrayPush(pArray, &tbSver);
   }
 
-  SRequestConnInfo conn = {.pTrans = pRequest->pTscObj->pAppInfo->pTransporter,
-                           .requestId = pRequest->requestId,
-                           .requestObjRefId = pRequest->self,
-                           .mgmtEps = *epset};
-
-  code = catalogChkTbMetaVersion(pCatalog, &conn, pArray);
-
-_return:
-
-  taosArrayDestroy(pArray);
-  return code;
+  return TSDB_CODE_SUCCESS;
 }
 
 int32_t handleQueryExecRes(SRequestObj* pRequest, void* res, SCatalog* pCatalog, SEpSet* epset) {

source/client/src/clientMain.c

@@ -866,7 +866,6 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt) {
                            .pTransporter = pTscObj->pAppInfo->pTransporter,
                            .pStmtCb = NULL,
                            .pUser = pTscObj->user,
-                           .schemalessType = pTscObj->schemalessType,
                            .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER)),
                            .enableSysInfo = pTscObj->sysInfo,
                            .async = true,

source/client/src/clientSmlJson.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <ctype.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include "clientSml.h"
+
+#define JUMP_JSON_SPACE(start) \
+while(*(start)){\
+  if(unlikely(*(start) > 32))\
+    break;\
+  else\
+    (start)++;\
+  }
+
+SArray *smlJsonParseTags(char *start, char *end){
+  SArray *tags = taosArrayInit(4, sizeof(SSmlKv));
+  while(start < end){
+    SSmlKv kv = {0};
+    kv.type = TSDB_DATA_TYPE_NCHAR;
+    bool isInQuote = false;
+    while(start < end){
+      if(unlikely(!isInQuote && *start == '"')){
+        start++;
+        kv.key = start;
+        isInQuote = true;
+        continue;
+      }
+      if(unlikely(isInQuote && *start == '"')){
+        kv.keyLen = start - kv.key;
+        start++;
+        break;
+      }
+      start++;
+    }
+    bool hasColon = false;
+    while(start < end){
+      if(unlikely(!hasColon && *start == ':')){
+        start++;
+        hasColon = true;
+        continue;
+      }
+      if(unlikely(hasColon && kv.value == NULL && (*start > 32 && *start != '"'))){
+        kv.value = start;
+        start++;
+        continue;
+      }
+
+      if(unlikely(hasColon && kv.value != NULL && (*start == '"' || *start == ',' || *start == '}'))){
+        kv.length = start - kv.value;
+        taosArrayPush(tags, &kv);
+        start++;
+        break;
+      }
+      start++;
+    }
+  }
+  return tags;
+}
+
+static int32_t smlParseTagsFromJSON(SSmlHandle *info, SSmlLineInfo *elements) {
+  int32_t ret = TSDB_CODE_SUCCESS;
+
+  if(is_same_child_table_telnet(elements, &info->preLine) == 0){
+    return TSDB_CODE_SUCCESS;
+  }
+
+  bool isSameMeasure = IS_SAME_SUPER_TABLE;
+
+  int     cnt = 0;
+  SArray *preLineKV = info->preLineTagKV;
+  bool    isSuperKVInit = true;
+  SArray *superKV = NULL;
+  if(info->dataFormat){
+    if(unlikely(!isSameMeasure)){
+      SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
+
+      if(unlikely(sMeta == NULL)){
+        sMeta = smlBuildSTableMeta(info->dataFormat);
+        STableMeta * pTableMeta = smlGetMeta(info, elements->measure, elements->measureLen);
+        sMeta->tableMeta = pTableMeta;
+        if(pTableMeta == NULL){
+          info->dataFormat = false;
+          info->reRun      = true;
+          return TSDB_CODE_SUCCESS;
+        }
+        nodeListSet(&info->superTables, elements->measure, elements->measureLen, sMeta, NULL);
+      }
+      info->currSTableMeta = sMeta->tableMeta;
+      superKV = sMeta->tags;
+
+      if(unlikely(taosArrayGetSize(superKV) == 0)){
+        isSuperKVInit = false;
+      }
+      taosArraySetSize(preLineKV, 0);
+    }
+  }else{
+    taosArraySetSize(preLineKV, 0);
+  }
+
+  SArray *tags = smlJsonParseTags(elements->tags, elements->tags + elements->tagsLen);
+  int32_t tagNum = taosArrayGetSize(tags);
+  for (int32_t i = 0; i < tagNum; ++i) {
+    SSmlKv kv = *(SSmlKv*)taosArrayGet(tags, i);
+
+    if(info->dataFormat){
+      if(unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)){
+        info->dataFormat = false;
+        info->reRun      = true;
+        taosArrayDestroy(tags);
+        return TSDB_CODE_SUCCESS;
+      }
+
+      if(isSameMeasure){
+        if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
+          info->dataFormat = false;
+          info->reRun      = true;
+          taosArrayDestroy(tags);
+          return TSDB_CODE_SUCCESS;
+        }
+        SSmlKv *preKV = (SSmlKv *)taosArrayGet(preLineKV, cnt);
+        if(unlikely(kv.length > preKV->length)){
+          preKV->length = kv.length;
+          SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
+          ASSERT(tableMeta != NULL);
+
+          SSmlKv *oldKV = (SSmlKv *)taosArrayGet(tableMeta->tags, cnt);
+          oldKV->length = kv.length;
+          info->needModifySchema = true;
+        }
+        if(unlikely(!IS_SAME_KEY)){
+          info->dataFormat = false;
+          info->reRun      = true;
+          taosArrayDestroy(tags);
+          return TSDB_CODE_SUCCESS;
+        }
+      }else{
+        if(isSuperKVInit){
+          if(unlikely(cnt >= taosArrayGetSize(superKV))) {
+            info->dataFormat = false;
+            info->reRun      = true;
+            taosArrayDestroy(tags);
+            return TSDB_CODE_SUCCESS;
+          }
+          SSmlKv *preKV = (SSmlKv *)taosArrayGet(superKV, cnt);
+          if(unlikely(kv.length > preKV->length)) {
+            preKV->length = kv.length;
+          }else{
+            kv.length = preKV->length;
+          }
+          info->needModifySchema = true;
+
+          if(unlikely(!IS_SAME_KEY)){
+            info->dataFormat = false;
+            info->reRun      = true;
+            taosArrayDestroy(tags);
+            return TSDB_CODE_SUCCESS;
+          }
+        }else{
+          taosArrayPush(superKV, &kv);
+        }
+        taosArrayPush(preLineKV, &kv);
+      }
+    }else{
+      taosArrayPush(preLineKV, &kv);
+    }
+    cnt++;
+  }
+  taosArrayDestroy(tags);
+
+  SSmlTableInfo *tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements, POINTER_BYTES, is_same_child_table_telnet);
+  if (unlikely(tinfo == NULL)) {
+    tinfo = smlBuildTableInfo(1, elements->measure, elements->measureLen);
+    if (unlikely(!tinfo)) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    tinfo->tags = taosArrayDup(preLineKV, NULL);
+
+    smlSetCTableName(tinfo);
+    if (info->dataFormat) {
+      info->currSTableMeta->uid = tinfo->uid;
+      tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
+      if (tinfo->tableDataCtx == NULL) {
+        smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
+        return TSDB_CODE_SML_INVALID_DATA;
+      }
+    }
+
+    SSmlLineInfo *key = (SSmlLineInfo *)taosMemoryMalloc(sizeof(SSmlLineInfo));
+    *key = *elements;
+    tinfo->key = key;
+    nodeListSet(&info->childTables, key, POINTER_BYTES, tinfo, is_same_child_table_telnet);
+  }
+  if (info->dataFormat) info->currTableDataCtx = tinfo->tableDataCtx;
+
+  return ret;
+}
+
+static char* smlJsonGetObj(char *payload){
+  int   leftBracketCnt = 0;
+  while(*payload) {
+    if (unlikely(*payload == '{')) {
+      leftBracketCnt++;
+      payload++;
+      continue;
+    }
+    if (unlikely(*payload == '}')) {
+      leftBracketCnt--;
+      payload++;
+      if (leftBracketCnt == 0) {
+        return payload;
+      } else if (leftBracketCnt < 0) {
+        return NULL;
+      }
+      continue;
+    }
+    payload++;
+  }
+  return NULL;
+}
+
+void smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset){
+  int index = 0;
+  while(*(*start)){
+    if((*start)[0] != '"'){
+      (*start)++;
+      continue;
+    }
+
+    if(unlikely(index >= 4)) {
+      uError("index >= 4, %s", *start)
+      break;
+    }
+    char *sTmp = *start;
+    if((*start)[1] == 'm' && (*start)[2] == 'e' && (*start)[3] == 't'
+       && (*start)[4] == 'r' &&  (*start)[5] == 'i' && (*start)[6] == 'c' && (*start)[7] == '"'){
+
+      (*start) += 8;
+      bool isInQuote = false;
+      while(*(*start)){
+        if(unlikely(!isInQuote && *(*start) == '"')){
+          (*start)++;
+          offset[index++] = *start - sTmp;
+          element->measure = (*start);
+          isInQuote = true;
+          continue;
+        }
+        if(unlikely(isInQuote && *(*start) == '"')){
+          element->measureLen = (*start) - element->measure;
+          break;
+        }
+        (*start)++;
+      }
+    }else if((*start)[1] == 't' && (*start)[2] == 'i' && (*start)[3] == 'm'
+             && (*start)[4] == 'e' &&  (*start)[5] == 's' && (*start)[6] == 't'
+             && (*start)[7] == 'a' &&  (*start)[8] == 'm' && (*start)[9] == 'p' && (*start)[10] == '"'){
+
+      (*start) += 11;
+      bool hasColon = false;
+      while(*(*start)){
+        if(unlikely(!hasColon && *(*start) == ':')){
+          (*start)++;
+          JUMP_JSON_SPACE((*start))
+          offset[index++] = *start - sTmp;
+          element->timestamp = (*start);
+          hasColon = true;
+          continue;
+        }
+        if(unlikely(hasColon && (*(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32))){
+          element->timestampLen = (*start) - element->timestamp;
+          break;
+        }
+        (*start)++;
+      }
+    }else if((*start)[1] == 'v' && (*start)[2] == 'a' && (*start)[3] == 'l'
+             && (*start)[4] == 'u' &&  (*start)[5] == 'e' && (*start)[6] == '"'){
+
+      (*start) += 7;
+
+      bool hasColon = false;
+      while(*(*start)){
+        if(unlikely(!hasColon && *(*start) == ':')){
+          (*start)++;
+          JUMP_JSON_SPACE((*start))
+          offset[index++] = *start - sTmp;
+          element->cols = (*start);
+          hasColon = true;
+          continue;
+        }
+        if(unlikely(hasColon && (*(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32))){
+          element->colsLen = (*start) - element->cols;
+          break;
+        }
+        (*start)++;
+      }
+    }else if((*start)[1] == 't' && (*start)[2] == 'a' && (*start)[3] == 'g'
+             && (*start)[4] == 's' && (*start)[5] == '"'){
+      (*start) += 6;
+
+      while(*(*start)){
+        if(unlikely(*(*start) == ':')){
+          (*start)++;
+          JUMP_JSON_SPACE((*start))
+          offset[index++] = *start - sTmp;
+          element->tags = (*start);
+          char* tmp = smlJsonGetObj((*start));
+          if(tmp){
+            element->tagsLen = tmp - (*start);
+            *start = tmp;
+          }
+          break;
+        }
+        (*start)++;
+      }
+    }
+    if(*(*start) == '}'){
+      (*start)++;
+      break;
+    }
+    (*start)++;
+  }
+}
+
+void smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset){
+  int index = 0;
+  while(*(*start)){
+    if((*start)[0] != '"'){
+      (*start)++;
+      continue;
+    }
+
+    if(unlikely(index >= 4)) {
+      uError("index >= 4, %s", *start)
+      break;
+    }
+    if((*start)[1] == 'm'){
+      (*start) += offset[index++];
+      element->measure = *start;
+      while(*(*start)){
+        if(unlikely(*(*start) == '"')){
+          element->measureLen = (*start) - element->measure;
+          break;
+        }
+        (*start)++;
+      }
+    }else if((*start)[1] == 't' && (*start)[2] == 'i'){
+      (*start) += offset[index++];
+      element->timestamp = *start;
+      while(*(*start)){
+        if(unlikely(*(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32)){
+          element->timestampLen = (*start) - element->timestamp;
+          break;
+        }
+        (*start)++;
+      }
+    }else if((*start)[1] == 'v'){
+      (*start) += offset[index++];
+      element->cols = *start;
+      while(*(*start)){
+        if(unlikely( *(*start) == ',' || *(*start) == '}' || (*(*start)) <= 32)){
+          element->colsLen = (*start) - element->cols;
+          break;
+        }
+        (*start)++;
+      }
+    }else if((*start)[1] == 't' && (*start)[2] == 'a'){
+      (*start) += offset[index++];
+      element->tags = (*start);
+      char* tmp = smlJsonGetObj((*start));
+      if(tmp){
+        element->tagsLen = tmp - (*start);
+        *start = tmp;
+      }
+      break;
+    }
+    if(*(*start) == '}'){
+      (*start)++;
+      break;
+    }
+    (*start)++;
+  }
+}
+
+static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *elements) {
+  int32_t ret = TSDB_CODE_SUCCESS;
+
+  if(info->offset[0] == 0){
+    smlJsonParseObjFirst(start, elements, info->offset);
+  }else{
+    smlJsonParseObj(start, elements, info->offset);
+  }
+  if(**start == '\0') return TSDB_CODE_SUCCESS;
+
+  SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN, .value = elements->cols, .length = (size_t)elements->colsLen};
+  if (smlParseNumber(&kv, &info->msgBuf)) {
+    kv.length = (int16_t)tDataTypes[kv.type].bytes;
+  }else{
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  // Parse tags
+  ret = smlParseTagsFromJSON(info, elements);
+  if (unlikely(ret)) {
+    uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
+    return ret;
+  }
+
+  if(unlikely(info->reRun)){
+    return TSDB_CODE_SUCCESS;
+  }
+
+  // Parse timestamp
+  // notice!!! put ts back to tag to ensure get meta->precision
+  int64_t ts = smlParseOpenTsdbTime(info, elements->timestamp, elements->timestampLen);
+  if (unlikely(ts < 0)) {
+    uError("OTD:0x%" PRIx64 " Unable to parse timestamp from JSON payload", info->id);
+    return TSDB_CODE_INVALID_TIMESTAMP;
+  }
+  SSmlKv kvTs = { .key = TS, .keyLen = TS_LEN, .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
+
+  if(info->dataFormat){
+    ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kvTs, 0);
+    if(ret == TSDB_CODE_SUCCESS){
+      ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 1);
+    }
+    if(ret == TSDB_CODE_SUCCESS){
+      ret = smlBuildRow(info->currTableDataCtx);
+    }
+    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
+      smlBuildInvalidDataMsg(&info->msgBuf, "smlBuildCol error", NULL);
+      return ret;
+    }
+  }else{
+    if(elements->colArray == NULL){
+      elements->colArray = taosArrayInit(16, sizeof(SSmlKv));
+    }
+    taosArrayPush(elements->colArray, &kvTs);
+    taosArrayPush(elements->colArray, &kv);
+  }
+  info->preLine = *elements;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t smlParseJSON(SSmlHandle *info, char *payload) {
+  int32_t payloadNum = 1 << 15;
+  int32_t ret = TSDB_CODE_SUCCESS;
+
+  int cnt = 0;
+  char *dataPointStart = payload;
+  while (1) {
+    if(info->dataFormat) {
+      SSmlLineInfo element = {0};
+      ret = smlParseJSONString(info, &dataPointStart, &element);
+    }else{
+      if(cnt >= payloadNum){
+        payloadNum = payloadNum << 1;
+        void* tmp = taosMemoryRealloc(info->lines, payloadNum * sizeof(SSmlLineInfo));
+        if(tmp != NULL){
+          info->lines = (SSmlLineInfo*)tmp;
+        }
+      }
+      ret = smlParseJSONString(info, &dataPointStart, info->lines + cnt);
+    }
+    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
+      uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
+      return ret;
+    }
+
+    if(*dataPointStart == '\0') break;
+
+    if(unlikely(info->reRun)){
+      cnt = 0;
+      dataPointStart = payload;
+      info->lineNum = payloadNum;
+      ret = smlClearForRerun(info);
+      if(ret != TSDB_CODE_SUCCESS){
+        return ret;
+      }
+      continue;
+    }
+    cnt++;
+  }
+  info->lineNum = cnt;
+
+  return TSDB_CODE_SUCCESS;
+}

source/client/src/clientSmlTelnet.c

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include <ctype.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include "clientSml.h"
+
+int32_t is_same_child_table_telnet(const void *a, const void *b){
+  SSmlLineInfo *t1 = (SSmlLineInfo *)a;
+  SSmlLineInfo *t2 = (SSmlLineInfo *)b;
+  return (((t1->measureLen == t2->measureLen) && memcmp(t1->measure, t2->measure, t1->measureLen) == 0)
+          && ((t1->tagsLen == t2->tagsLen) && memcmp(t1->tags, t2->tags, t1->tagsLen) == 0)) ? 0 : 1;
+}
+
+int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len) {
+  uint8_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;
+
+  if (unlikely(!data)) {
+    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
+    return -1;
+  }
+  if (unlikely(len == 1 && data[0] == '0')) {
+    return taosGetTimestampNs()/smlFactorNS[toPrecision];
+  }
+  int8_t fromPrecision = smlGetTsTypeByLen(len);
+  if (unlikely(fromPrecision == -1)) {
+    smlBuildInvalidDataMsg(&info->msgBuf,
+                           "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
+    return -1;
+  }
+  int64_t ts = smlGetTimeValue(data, len, fromPrecision, toPrecision);
+  if (unlikely(ts == -1)) {
+    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
+    return -1;
+  }
+  return ts;
+}
+
+
+static void smlParseTelnetElement(char **sql, char *sqlEnd, char **data, int32_t *len) {
+  while (*sql < sqlEnd) {
+    if (unlikely((**sql != SPACE && !(*data)))) {
+      *data = *sql;
+    } else if (unlikely(**sql == SPACE && *data)) {
+      *len = *sql - *data;
+      break;
+    }
+    (*sql)++;
+  }
+}
+
+static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
+  if(is_same_child_table_telnet(elements, &info->preLine) == 0){
+    return TSDB_CODE_SUCCESS;
+  }
+
+  bool isSameMeasure = IS_SAME_SUPER_TABLE;
+
+  int     cnt = 0;
+  SArray *preLineKV = info->preLineTagKV;
+  bool    isSuperKVInit = true;
+  SArray *superKV = NULL;
+  if(info->dataFormat){
+    if(!isSameMeasure){
+      SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
+
+      if(unlikely(sMeta == NULL)){
+        sMeta = smlBuildSTableMeta(info->dataFormat);
+        STableMeta * pTableMeta = smlGetMeta(info, elements->measure, elements->measureLen);
+        sMeta->tableMeta = pTableMeta;
+        if(pTableMeta == NULL){
+          info->dataFormat = false;
+          info->reRun      = true;
+          return TSDB_CODE_SUCCESS;
+        }
+        nodeListSet(&info->superTables, elements->measure, elements->measureLen, sMeta, NULL);
+      }
+      info->currSTableMeta = sMeta->tableMeta;
+      superKV = sMeta->tags;
+
+      if(unlikely(taosArrayGetSize(superKV) == 0)){
+        isSuperKVInit = false;
+      }
+      taosArraySetSize(preLineKV, 0);
+    }
+  }else{
+    taosArraySetSize(preLineKV, 0);
+  }
+
+  const char *sql = data;
+  while (sql < sqlEnd) {
+    JUMP_SPACE(sql, sqlEnd)
+    if (unlikely(*sql == '\0')) break;
+
+    const char *key = sql;
+    size_t     keyLen = 0;
+
+    // parse key
+    while (sql < sqlEnd) {
+      if (unlikely(*sql == SPACE)) {
+        smlBuildInvalidDataMsg(msg, "invalid data", sql);
+        return TSDB_CODE_SML_INVALID_DATA;
+      }
+      if (unlikely(*sql == EQUAL)) {
+        keyLen = sql - key;
+        sql++;
+        break;
+      }
+      sql++;
+    }
+
+    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
+      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
+      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
+    }
+//    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
+//      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
+//      return TSDB_CODE_TSC_DUP_NAMES;
+//    }
+
+    // parse value
+    const char *value = sql;
+    size_t     valueLen = 0;
+    while (sql < sqlEnd) {
+      // parse value
+      if (unlikely(*sql == SPACE)) {
+        break;
+      }
+      if (unlikely(*sql == EQUAL)) {
+        smlBuildInvalidDataMsg(msg, "invalid data", sql);
+        return TSDB_CODE_SML_INVALID_DATA;
+      }
+      sql++;
+    }
+    valueLen = sql - value;
+
+    if (unlikely(valueLen == 0)) {
+      smlBuildInvalidDataMsg(msg, "invalid value", value);
+      return TSDB_CODE_TSC_INVALID_VALUE;
+    }
+
+    if (unlikely(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
+      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
+    }
+
+    SSmlKv kv = {.key = key, .keyLen = keyLen, .type = TSDB_DATA_TYPE_NCHAR, .value = value, .length = valueLen};
+
+    if(info->dataFormat){
+      if(unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)){
+        info->dataFormat = false;
+        info->reRun      = true;
+        return TSDB_CODE_SUCCESS;
+      }
+
+      if(isSameMeasure){
+        if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
+          info->dataFormat = false;
+          info->reRun      = true;
+          return TSDB_CODE_SUCCESS;
+        }
+        SSmlKv *preKV = (SSmlKv *)taosArrayGet(preLineKV, cnt);
+        if(unlikely(kv.length > preKV->length)){
+          preKV->length = kv.length;
+          SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
+          ASSERT(tableMeta != NULL);
+
+          SSmlKv *oldKV = (SSmlKv *)taosArrayGet(tableMeta->tags, cnt);
+          oldKV->length = kv.length;
+          info->needModifySchema = true;
+        }
+        if(unlikely(!IS_SAME_KEY)){
+          info->dataFormat = false;
+          info->reRun      = true;
+          return TSDB_CODE_SUCCESS;
+        }
+      }else{
+        if(isSuperKVInit){
+          if(unlikely(cnt >= taosArrayGetSize(superKV))) {
+            info->dataFormat = false;
+            info->reRun      = true;
+            return TSDB_CODE_SUCCESS;
+          }
+          SSmlKv *preKV = (SSmlKv *)taosArrayGet(superKV, cnt);
+          if(unlikely(kv.length > preKV->length)) {
+            preKV->length = kv.length;
+          }else{
+            kv.length = preKV->length;
+          }
+          info->needModifySchema = true;
+
+          if(unlikely(!IS_SAME_KEY)){
+            info->dataFormat = false;
+            info->reRun      = true;
+            return TSDB_CODE_SUCCESS;
+          }
+        }else{
+          taosArrayPush(superKV, &kv);
+        }
+        taosArrayPush(preLineKV, &kv);
+      }
+    }else{
+      taosArrayPush(preLineKV, &kv);
+    }
+    cnt++;
+  }
+  SSmlTableInfo *tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements, POINTER_BYTES, is_same_child_table_telnet);
+  if (unlikely(tinfo == NULL)) {
+    tinfo = smlBuildTableInfo(1, elements->measure, elements->measureLen);
+    if (!tinfo) {
+      return TSDB_CODE_OUT_OF_MEMORY;
+    }
+    tinfo->tags = taosArrayDup(preLineKV, NULL);
+
+    smlSetCTableName(tinfo);
+    if (info->dataFormat) {
+      info->currSTableMeta->uid = tinfo->uid;
+      tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
+      if (tinfo->tableDataCtx == NULL) {
+        smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
+        return TSDB_CODE_SML_INVALID_DATA;
+      }
+    }
+
+    SSmlLineInfo *key = (SSmlLineInfo *)taosMemoryMalloc(sizeof(SSmlLineInfo));
+    *key = *elements;
+    tinfo->key = key;
+    nodeListSet(&info->childTables, key, POINTER_BYTES, tinfo, is_same_child_table_telnet);
+  }
+  if (info->dataFormat) info->currTableDataCtx = tinfo->tableDataCtx;
+
+  return TSDB_CODE_SUCCESS;
+}
+
+// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
+int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements) {
+  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
+
+  // parse metric
+  smlParseTelnetElement(&sql, sqlEnd, &elements->measure, &elements->measureLen);
+  if (unlikely((!(elements->measure) || IS_INVALID_TABLE_LEN(elements->measureLen)))) {
+    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
+    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
+  }
+
+  // parse timestamp
+  smlParseTelnetElement(&sql, sqlEnd, &elements->timestamp, &elements->timestampLen);
+  if (unlikely(!elements->timestamp || elements->timestampLen == 0)) {
+    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
+    return TSDB_CODE_SML_INVALID_DATA;
+  }
+
+  bool needConverTime = false;  // get TS before parse tag(get meta), so need conver time
+  if(info->dataFormat && info->currSTableMeta == NULL){
+    needConverTime = true;
+  }
+  int64_t ts = smlParseOpenTsdbTime(info, elements->timestamp, elements->timestampLen);
+  if (unlikely(ts < 0)) {
+    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
+    return TSDB_CODE_INVALID_TIMESTAMP;
+  }
+  SSmlKv kvTs = { .key = TS, .keyLen = TS_LEN, .type = TSDB_DATA_TYPE_TIMESTAMP, .i = ts, .length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
+
+  // parse value
+  smlParseTelnetElement(&sql, sqlEnd, &elements->cols, &elements->colsLen);
+  if (unlikely(!elements->cols || elements->colsLen == 0)) {
+    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN, .value = elements->cols, .length = (size_t)elements->colsLen};
+  if (smlParseValue(&kv, &info->msgBuf) != TSDB_CODE_SUCCESS) {
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  JUMP_SPACE(sql, sqlEnd)
+
+  elements->tags = sql;
+  elements->tagsLen = sqlEnd - sql;
+  if (unlikely(!elements->tags || elements->tagsLen == 0)) {
+    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
+    return TSDB_CODE_TSC_INVALID_VALUE;
+  }
+
+  int ret = smlParseTelnetTags(info, sql, sqlEnd, elements, &info->msgBuf);
+  if (unlikely(ret != TSDB_CODE_SUCCESS)) {
+    return ret;
+  }
+
+  if(unlikely(info->reRun)){
+    return TSDB_CODE_SUCCESS;
+  }
+
+  if(info->dataFormat){
+    if(needConverTime) {
+      kvTs.i = convertTimePrecision(kvTs.i, TSDB_TIME_PRECISION_NANO, info->currSTableMeta->tableInfo.precision);
+    }
+    ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kvTs, 0);
+    if(ret == TSDB_CODE_SUCCESS){
+      ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 1);
+    }
+    if(ret == TSDB_CODE_SUCCESS){
+      ret = smlBuildRow(info->currTableDataCtx);
+    }
+    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
+      smlBuildInvalidDataMsg(&info->msgBuf, "smlBuildCol error", NULL);
+      return ret;
+    }
+  }else{
+    if(elements->colArray == NULL){
+      elements->colArray = taosArrayInit(16, sizeof(SSmlKv));
+    }
+    taosArrayPush(elements->colArray, &kvTs);
+    taosArrayPush(elements->colArray, &kv);
+  }
+  info->preLine = *elements;
+
+  return TSDB_CODE_SUCCESS;
+}
\ No newline at end of file

source/client/src/clientStmt.c

@@ -152,7 +152,8 @@ int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName, bool autoCreateTbl) {
+int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName,
+                           bool autoCreateTbl) {
   STscStmt* pStmt = (STscStmt*)stmt;
   char      tbFName[TSDB_TABLE_FNAME_LEN];
   tNameExtractFullName(tbName, tbFName);
@@ -171,12 +172,11 @@ int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags,
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash, bool autoCreateTbl) {
+int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockHash) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
   pStmt->sql.pVgHash = pVgHash;
   pStmt->exec.pBlockHash = pBlockHash;
-  pStmt->exec.autoCreateTbl = autoCreateTbl;
 
   return TSDB_CODE_SUCCESS;
 }
@@ -186,7 +186,7 @@ int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SNam
   STscStmt* pStmt = (STscStmt*)stmt;
 
   STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));
-  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash, autoCreateTbl));
+  STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
 
   pStmt->sql.autoCreateTbl = autoCreateTbl;
 
@@ -214,16 +214,16 @@ int32_t stmtCacheBlock(STscStmt* pStmt) {
     return TSDB_CODE_SUCCESS;
   }
 
-  STableDataBlocks** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
+  STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
   if (!pSrc) {
     return TSDB_CODE_OUT_OF_MEMORY;
   }
-  STableDataBlocks* pDst = NULL;
+  STableDataCxt* pDst = NULL;
 
-  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc));
+  STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
 
   SStmtTableCache cache = {
-      .pDataBlock = pDst,
+      .pDataCtx = pDst,
       .boundTags = pStmt->bInfo.boundTags,
   };
 
@@ -241,6 +241,8 @@ int32_t stmtCacheBlock(STscStmt* pStmt) {
 }
 
 int32_t stmtParseSql(STscStmt* pStmt) {
+  pStmt->exec.pCurrBlock = NULL;
+
   SStmtCallback stmtCb = {
       .pStmt = pStmt,
       .getTbNameFn = stmtGetTbName,
@@ -273,7 +275,7 @@ int32_t stmtCleanBindInfo(STscStmt* pStmt) {
   pStmt->bInfo.tbName[0] = 0;
   pStmt->bInfo.tbFName[0] = 0;
   if (!pStmt->bInfo.tagsCached) {
-    destroyBoundColumnInfo(pStmt->bInfo.boundTags);
+    qDestroyBoundColInfo(pStmt->bInfo.boundTags);
     taosMemoryFreeClear(pStmt->bInfo.boundTags);
   }
   memset(pStmt->bInfo.stbFName, 0, TSDB_TABLE_FNAME_LEN);
@@ -289,29 +291,25 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
   size_t keyLen = 0;
   void*  pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
   while (pIter) {
-    STableDataBlocks* pBlocks = *(STableDataBlocks**)pIter;
+    STableDataCxt* pBlocks = *(STableDataCxt**)pIter;
     char*          key = taosHashGetKey(pIter, &keyLen);
     STableMeta*    pMeta = qGetTableMetaInDataBlock(pBlocks);
 
-    if (keepTable && (strlen(pStmt->bInfo.tbFName) == keyLen) && strncmp(pStmt->bInfo.tbFName, key, keyLen) == 0) {
-      STMT_ERR_RET(qResetStmtDataBlock(pBlocks, true));
+    if (keepTable && pBlocks == pStmt->exec.pCurrBlock) {
+      ASSERT(NULL == pBlocks->pData);
+      TSWAP(pBlocks->pData, pStmt->exec.pCurrTbData);
+      STMT_ERR_RET(qResetStmtDataBlock(pBlocks, false));
 
       pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
       continue;
     }
 
-    if (STMT_TYPE_MULTI_INSERT == pStmt->sql.type) {
-      qFreeStmtDataBlock(pBlocks);
-    } else {
     qDestroyStmtDataBlock(pBlocks);
-    }
     taosHashRemove(pStmt->exec.pBlockHash, key, keyLen);
 
     pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
   }
 
-  pStmt->exec.autoCreateTbl = false;
-
   if (keepTable) {
     return TSDB_CODE_SUCCESS;
   }
@@ -319,6 +317,9 @@ int32_t stmtCleanExecInfo(STscStmt* pStmt, bool keepTable, bool deepClean) {
   taosHashCleanup(pStmt->exec.pBlockHash);
   pStmt->exec.pBlockHash = NULL;
 
+  tDestroySSubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
+  taosMemoryFreeClear(pStmt->exec.pCurrTbData);
+
   STMT_ERR_RET(stmtCleanBindInfo(pStmt));
 
   return TSDB_CODE_SUCCESS;
@@ -337,8 +338,8 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
   while (pIter) {
     SStmtTableCache* pCache = (SStmtTableCache*)pIter;
 
-    qDestroyStmtDataBlock(pCache->pDataBlock);
-    destroyBoundColumnInfo(pCache->boundTags);
+    qDestroyStmtDataBlock(pCache->pDataCtx);
+    qDestroyBoundColInfo(pCache->boundTags);
     taosMemoryFreeClear(pCache->boundTags);
 
     pIter = taosHashIterate(pStmt->sql.pTableCache, pIter);
@@ -354,7 +355,8 @@ int32_t stmtCleanSQLInfo(STscStmt* pStmt) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks* pDataBlock, STableDataBlocks** newBlock, uint64_t uid) {
+int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataCxt* pDataBlock, STableDataCxt** newBlock, uint64_t uid,
+                             uint64_t suid) {
   SEpSet           ep = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
   SVgroupInfo      vgInfo = {0};
   SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
@@ -366,7 +368,9 @@ int32_t stmtRebuildDataBlock(STscStmt* pStmt, STableDataBlocks* pDataBlock, STab
   STMT_ERR_RET(
       taosHashPut(pStmt->sql.pVgHash, (const char*)&vgInfo.vgId, sizeof(vgInfo.vgId), (char*)&vgInfo, sizeof(vgInfo)));
 
-  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, vgInfo.vgId));
+  STMT_ERR_RET(qRebuildStmtDataBlock(newBlock, pDataBlock, uid, suid, vgInfo.vgId, pStmt->sql.autoCreateTbl));
+
+  STMT_DLOG("tableDataCxt rebuilt, uid:%" PRId64 ", vgId:%d", uid, vgInfo.vgId);
 
   return TSDB_CODE_SUCCESS;
 }
@@ -375,12 +379,13 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
   pStmt->bInfo.needParse = true;
   pStmt->bInfo.inExecCache = false;
 
-  STableDataBlocks* pBlockInExec =
-      taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
-  if (pBlockInExec) {
+  STableDataCxt** pCxtInExec = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
+  if (pCxtInExec) {
     pStmt->bInfo.needParse = false;
     pStmt->bInfo.inExecCache = true;
 
+    pStmt->exec.pCurrBlock = *pCxtInExec;
+
     if (pStmt->sql.autoCreateTbl) {
       tscDebug("reuse stmt block for tb %s in execBlock", pStmt->bInfo.tbFName);
       return TSDB_CODE_SUCCESS;
@@ -407,18 +412,18 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     SStmtTableCache* pCache = taosHashGet(pStmt->sql.pTableCache, &pStmt->bInfo.tbSuid, sizeof(pStmt->bInfo.tbSuid));
     if (pCache) {
       pStmt->bInfo.needParse = false;
-      pStmt->exec.autoCreateTbl = true;
-
       pStmt->bInfo.tbUid = 0;
 
-      STableDataBlocks* pNewBlock = NULL;
-      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, 0));
+      STableDataCxt* pNewBlock = NULL;
+      STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, 0, pStmt->bInfo.tbSuid));
 
       if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
                       POINTER_BYTES)) {
         STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
       }
 
+      pStmt->exec.pCurrBlock = pNewBlock;
+
       tscDebug("reuse stmt block for tb %s in sqlBlock, suid:0x%" PRIx64, pStmt->bInfo.tbFName, pStmt->bInfo.tbSuid);
 
       return TSDB_CODE_SUCCESS;
@@ -489,14 +494,16 @@ int32_t stmtGetFromCache(STscStmt* pStmt) {
     pStmt->bInfo.boundTags = pCache->boundTags;
     pStmt->bInfo.tagsCached = true;
 
-    STableDataBlocks* pNewBlock = NULL;
-    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataBlock, &pNewBlock, uid));
+    STableDataCxt* pNewBlock = NULL;
+    STMT_ERR_RET(stmtRebuildDataBlock(pStmt, pCache->pDataCtx, &pNewBlock, uid, suid));
 
     if (taosHashPut(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName), &pNewBlock,
                     POINTER_BYTES)) {
       STMT_ERR_RET(TSDB_CODE_OUT_OF_MEMORY);
     }
 
+    pStmt->exec.pCurrBlock = pNewBlock;
+
     tscDebug("tb %s in sqlBlock list, set to current", pStmt->bInfo.tbFName);
 
     return TSDB_CODE_SUCCESS;
@@ -614,8 +621,8 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
     return TSDB_CODE_SUCCESS;
   }
 
-  STableDataBlocks** pDataBlock =
-      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
+  STableDataCxt** pDataBlock =
+      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
   if (NULL == pDataBlock) {
     tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
     STMT_ERR_RET(TSDB_CODE_APP_ERROR);
@@ -626,8 +633,6 @@ int stmtSetTbTags(TAOS_STMT* stmt, TAOS_MULTI_BIND* tags) {
                                   pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf,
                                   pStmt->exec.pRequest->msgBufLen));
 
-  pStmt->exec.autoCreateTbl = true;
-
   return TSDB_CODE_SUCCESS;
 }
 
@@ -637,8 +642,8 @@ int stmtFetchTagFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields
     STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
   }
 
-  STableDataBlocks** pDataBlock =
-      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
+  STableDataCxt** pDataBlock =
+      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
   if (NULL == pDataBlock) {
     tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
     STMT_ERR_RET(TSDB_CODE_APP_ERROR);
@@ -655,8 +660,8 @@ int stmtFetchColFields(STscStmt* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields
     STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
   }
 
-  STableDataBlocks** pDataBlock =
-      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
+  STableDataCxt** pDataBlock =
+      (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
   if (NULL == pDataBlock) {
     tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
     STMT_ERR_RET(TSDB_CODE_APP_ERROR);
@@ -729,11 +734,18 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
     return TSDB_CODE_SUCCESS;
   }
 
-  STableDataBlocks** pDataBlock =
-      (STableDataBlocks**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
+  STableDataCxt** pDataBlock = NULL;
+
+  if (pStmt->exec.pCurrBlock) {
+    pDataBlock = &pStmt->exec.pCurrBlock;
+  } else {
+    pDataBlock =
+        (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
     if (NULL == pDataBlock) {
       tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
-    STMT_ERR_RET(TSDB_CODE_APP_ERROR);
+      STMT_ERR_RET(TSDB_CODE_TSC_STMT_CACHE_ERROR);
+    }
+    pStmt->exec.pCurrBlock = *pDataBlock;
   }
 
   if (colIdx < 0) {
@@ -779,9 +791,9 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
   int32_t code = 0;
   int32_t finalCode = 0;
   size_t  keyLen = 0;
-  STableDataBlocks** pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
+  void*   pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
   while (pIter) {
-    STableDataBlocks* pBlock = *pIter;
+    STableDataCxt* pBlock = *(STableDataCxt**)pIter;
     char*          key = taosHashGetKey(pIter, &keyLen);
 
     STableMeta* pMeta = qGetTableMetaInDataBlock(pBlock);
@@ -861,7 +873,6 @@ int stmtExec(TAOS_STMT* stmt) {
   STscStmt*   pStmt = (STscStmt*)stmt;
   int32_t     code = 0;
   SSubmitRsp* pRsp = NULL;
-  bool        autoCreateTbl = pStmt->exec.autoCreateTbl;
 
   STMT_DLOG_E("start to exec");
 
@@ -870,8 +881,13 @@ int stmtExec(TAOS_STMT* stmt) {
   if (STMT_TYPE_QUERY == pStmt->sql.type) {
     launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
   } else {
+    tDestroySSubmitTbData(pStmt->exec.pCurrTbData, TSDB_MSG_FLG_ENCODE);
+    taosMemoryFreeClear(pStmt->exec.pCurrTbData);
+
+    STMT_ERR_RET(qCloneCurrentTbData(pStmt->exec.pCurrBlock, &pStmt->exec.pCurrTbData));
+
     STMT_ERR_RET(qBuildStmtOutput(pStmt->sql.pQuery, pStmt->sql.pVgHash, pStmt->exec.pBlockHash));
-    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, (autoCreateTbl ? (void**)&pRsp : NULL));
+    launchQueryImpl(pStmt->exec.pRequest, pStmt->sql.pQuery, true, NULL);
   }
 
   if (pStmt->exec.pRequest->code && NEED_CLIENT_HANDLE_ERROR(pStmt->exec.pRequest->code)) {
@@ -894,15 +910,6 @@ _return:
 
   stmtCleanExecInfo(pStmt, (code ? false : true), false);
 
-  if (TSDB_CODE_SUCCESS == code && autoCreateTbl) {
-    if (NULL == pRsp) {
-      tscError("no submit resp got for auto create table");
-      code = TSDB_CODE_APP_ERROR;
-    } else {
-      code = stmtUpdateTableUid(pStmt, pRsp);
-    }
-  }
-
   tFreeSSubmitRsp(pRsp);
 
   ++pStmt->sql.runTimes;

source/common/src/tdatablock.c

@@ -19,7 +19,7 @@
 #include "tlog.h"
 #include "tname.h"
 
-#define MALLOC_ALIGN_BYTES  256
+#define MALLOC_ALIGN_BYTES 32
 
 int32_t colDataGetLength(const SColumnInfoData* pColumnInfoData, int32_t numOfRows) {
   ASSERT(pColumnInfoData != NULL);
@@ -38,7 +38,8 @@ int32_t colDataGetFullLength(const SColumnInfoData* pColumnInfoData, int32_t num
   if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
     return pColumnInfoData->varmeta.length + sizeof(int32_t) * numOfRows;
   } else {
-    return ((pColumnInfoData->info.type == TSDB_DATA_TYPE_NULL) ? 0 : pColumnInfoData->info.bytes * numOfRows) + BitmapLen(numOfRows);
+    return ((pColumnInfoData->info.type == TSDB_DATA_TYPE_NULL) ? 0 : pColumnInfoData->info.bytes * numOfRows) +
+           BitmapLen(numOfRows);
   }
 }
 
@@ -359,7 +360,7 @@ size_t blockDataGetNumOfRows(const SSDataBlock* pBlock) { return pBlock->info.ro
 
 int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex) {
   if (pDataBlock->info.rows > 0) {
-//    ASSERT(pDataBlock->info.dataLoad == 1);
+    //    ASSERT(pDataBlock->info.dataLoad == 1);
   }
 
   if (pDataBlock == NULL || pDataBlock->info.rows <= 0 || pDataBlock->info.dataLoad == 0) {
@@ -1167,8 +1168,11 @@ void blockDataEmpty(SSDataBlock* pDataBlock) {
 }
 
 // todo temporarily disable it
-static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows, bool clearPayload) {
+
+static int32_t doEnsureCapacity(SColumnInfoData* pColumn, const SDataBlockInfo* pBlockInfo, uint32_t numOfRows,
+                                bool clearPayload) {
   ASSERT(numOfRows > 0);
+
   if (numOfRows <= pBlockInfo->capacity) {
     return TSDB_CODE_SUCCESS;
   }
@@ -1953,7 +1957,8 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
                   "===stream===%s|block type %d|child id %d|group id:%" PRIu64 "|uid:%" PRId64
                   "|rows:%d|version:%" PRIu64 "|cal start:%" PRIu64 "|cal end:%" PRIu64 "\n",
                   flag, (int32_t)pDataBlock->info.type, pDataBlock->info.childId, pDataBlock->info.id.groupId,
-                  pDataBlock->info.id.uid, pDataBlock->info.rows, pDataBlock->info.version, pDataBlock->info.calWin.skey, pDataBlock->info.calWin.ekey);
+                  pDataBlock->info.id.uid, pDataBlock->info.rows, pDataBlock->info.version,
+                  pDataBlock->info.calWin.skey, pDataBlock->info.calWin.ekey);
   if (len >= size - 1) return dumpBuf;
 
   for (int32_t j = 0; j < rows; j++) {
@@ -2055,6 +2060,7 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** pDataBuf)
  * @param suid
  *
  */
+#if 0
 int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataBlock, STSchema* pTSchema, int32_t vgId,
                                     tb_uid_t suid) {
   int32_t bufSize = sizeof(SSubmitReq);
@@ -2220,32 +2226,186 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
 
   return TSDB_CODE_SUCCESS;
 }
+#endif
+
+int32_t buildSubmitReqFromDataBlock(SSubmitReq2** ppReq, const SSDataBlock* pDataBlock, const STSchema* pTSchema,
+                                    int64_t uid, int32_t vgId, tb_uid_t suid) {
+  SSubmitReq2* pReq = *ppReq;
+  SArray*      pVals = NULL;
+  int32_t      numOfBlks = 0;
+  int32_t      sz = 1;
+
+  terrno = TSDB_CODE_SUCCESS;
+
+  if (NULL == pReq) {
+    if (!(pReq = taosMemoryMalloc(sizeof(SSubmitReq2)))) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      goto _end;
+    }
+
+    if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
+      goto _end;
+    }
+  }
+
+  for (int32_t i = 0; i < sz; ++i) {
+    int32_t colNum = taosArrayGetSize(pDataBlock->pDataBlock);
+    int32_t rows = pDataBlock->info.rows;
+
+    if (colNum <= 1) {  // invalid if only with TS col
+      continue;
+    }
+
+    // the rsma result should has the same column number with schema.
+    ASSERT(colNum == pTSchema->numOfCols);
+
+    SSubmitTbData tbData = {0};
+
+    if (!(tbData.aRowP = taosArrayInit(rows, sizeof(SRow*)))) {
+      goto _end;
+    }
+    tbData.suid = suid;
+    tbData.uid = uid;
+    tbData.sver = pTSchema->version;
+
+    if (!pVals && !(pVals = taosArrayInit(colNum, sizeof(SColVal)))) {
+      taosArrayDestroy(tbData.aRowP);
+      goto _end;
+    }
+
+    for (int32_t j = 0; j < rows; ++j) {  // iterate by row
+
+      taosArrayClear(pVals);
+
+      bool    isStartKey = false;
+      int32_t offset = 0;
+      for (int32_t k = 0; k < colNum; ++k) {  // iterate by column
+        SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
+        const STColumn*  pCol = &pTSchema->columns[k];
+        void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
+
+        switch (pColInfoData->info.type) {
+          case TSDB_DATA_TYPE_TIMESTAMP:
+            ASSERT(pColInfoData->info.type == pCol->type);
+            if (!isStartKey) {
+              isStartKey = true;
+              ASSERT(PRIMARYKEY_TIMESTAMP_COL_ID == pCol->colId);
+              SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, (SValue){.val = *(TSKEY*)var});
+              taosArrayPush(pVals, &cv);
+            } else if (colDataIsNull_s(pColInfoData, j)) {
+              SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
+              taosArrayPush(pVals, &cv);
+            } else {
+              SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, (SValue){.val = *(int64_t*)var});
+              taosArrayPush(pVals, &cv);
+            }
+            break;
+          case TSDB_DATA_TYPE_NCHAR:
+          case TSDB_DATA_TYPE_VARCHAR: {  // TSDB_DATA_TYPE_BINARY
+            ASSERT(pColInfoData->info.type == pCol->type);
+            if (colDataIsNull_s(pColInfoData, j)) {
+              SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
+              taosArrayPush(pVals, &cv);
+            } else {
+              void*   data = colDataGetVarData(pColInfoData, j);
+              SValue  sv = (SValue){.nData = varDataLen(data), .pData = varDataVal(data)};  // address copy, no value
+              SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, sv);
+              taosArrayPush(pVals, &cv);
+            }
+            break;
+          }
+          case TSDB_DATA_TYPE_VARBINARY:
+          case TSDB_DATA_TYPE_DECIMAL:
+          case TSDB_DATA_TYPE_BLOB:
+          case TSDB_DATA_TYPE_JSON:
+          case TSDB_DATA_TYPE_MEDIUMBLOB:
+            uError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
+            ASSERT(0);
+            break;
+          default:
+            if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
+              if (colDataIsNull_s(pColInfoData, j)) {
+                SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);  // should use pCol->type
+                taosArrayPush(pVals, &cv);
+              } else {
+                SValue sv;
+                if (pCol->type == pColInfoData->info.type) {
+                  memcpy(&sv.val, var, tDataTypes[pCol->type].bytes);
+                } else {
+                  /**
+                   *  1. sum/avg would convert to int64_t/uint64_t/double during aggregation
+                   *  2. below conversion may lead to overflow or loss, the app should select the right data type.
+                   */
+                  char tv[8] = {0};
+                  if (pColInfoData->info.type == TSDB_DATA_TYPE_FLOAT) {
+                    float v = 0;
+                    GET_TYPED_DATA(v, float, pColInfoData->info.type, var);
+                    SET_TYPED_DATA(&tv, pCol->type, v);
+                  } else if (pColInfoData->info.type == TSDB_DATA_TYPE_DOUBLE) {
+                    double v = 0;
+                    GET_TYPED_DATA(v, double, pColInfoData->info.type, var);
+                    SET_TYPED_DATA(&tv, pCol->type, v);
+                  } else if (IS_SIGNED_NUMERIC_TYPE(pColInfoData->info.type)) {
+                    int64_t v = 0;
+                    GET_TYPED_DATA(v, int64_t, pColInfoData->info.type, var);
+                    SET_TYPED_DATA(&tv, pCol->type, v);
+                  } else {
+                    uint64_t v = 0;
+                    GET_TYPED_DATA(v, uint64_t, pColInfoData->info.type, var);
+                    SET_TYPED_DATA(&tv, pCol->type, v);
+                  }
+                  memcpy(&sv.val, tv, tDataTypes[pCol->type].bytes);
+                }
+                SColVal cv = COL_VAL_VALUE(pCol->colId, pColInfoData->info.type, sv);
+                taosArrayPush(pVals, &cv);
+              }
+            } else {
+              uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
+              ASSERT(0);
+            }
+            break;
+        }
+      }
+      SRow* pRow = NULL;
+      if ((terrno = tRowBuild(pVals, pTSchema, &pRow)) < 0) {
+        tDestroySSubmitTbData(&tbData, TSDB_MSG_FLG_ENCODE);
+        goto _end;
+      }
+      ASSERT(pRow);
+      taosArrayPush(tbData.aRowP, &pRow);
+    }
+
+    taosArrayPush(pReq->aSubmitTbData, &tbData);
+  }
+_end:
+  taosArrayDestroy(pVals);
+  if (terrno != 0) {
+    *ppReq = NULL;
+    if (pReq) tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE);
+    return TSDB_CODE_FAILED;
+  }
+  *ppReq = pReq;
+  return TSDB_CODE_SUCCESS;
+}
 
 char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
   ASSERT(stbFullName[0] != 0);
-  SArray* tags = taosArrayInit(0, sizeof(void*));
+  SArray* tags = taosArrayInit(0, sizeof(SSmlKv));
   if (tags == NULL) {
     return NULL;
   }
 
-  SSmlKv* pTag = taosMemoryCalloc(1, sizeof(SSmlKv));
-  if (pTag == NULL) {
-    taosArrayDestroy(tags);
-    return NULL;
-  }
-
   void* cname = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
   if (cname == NULL) {
     taosArrayDestroy(tags);
-    taosMemoryFree(pTag);
     return NULL;
   }
 
-  pTag->key = "group_id";
-  pTag->keyLen = strlen(pTag->key);
-  pTag->type = TSDB_DATA_TYPE_UBIGINT;
-  pTag->u = groupId;
-  pTag->length = sizeof(uint64_t);
+  SSmlKv pTag = {.key = "group_id",
+                 .keyLen = sizeof("group_id") - 1,
+                 .type = TSDB_DATA_TYPE_UBIGINT,
+                 .u = groupId,
+                 .length = sizeof(uint64_t)};
   taosArrayPush(tags, &pTag);
 
   RandTableName rname = {
@@ -2257,7 +2417,6 @@ char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
 
   buildChildTableName(&rname);
 
-  taosMemoryFree(pTag);
   taosArrayDestroy(tags);
 
   ASSERT(rname.ctbShortName && rname.ctbShortName[0]);

source/common/src/tname.c

@@ -298,8 +298,8 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type) {
 }
 
 static int compareKv(const void* p1, const void* p2) {
-  SSmlKv* kv1 = *(SSmlKv**)p1;
-  SSmlKv* kv2 = *(SSmlKv**)p2;
+  SSmlKv* kv1 = (SSmlKv*)p1;
+  SSmlKv* kv2 = (SSmlKv*)p2;
   int32_t kvLen1 = kv1->keyLen;
   int32_t kvLen2 = kv2->keyLen;
   int32_t res = strncasecmp(kv1->key, kv2->key, TMIN(kvLen1, kvLen2));
@@ -320,7 +320,7 @@ void buildChildTableName(RandTableName* rName) {
   taosArraySort(rName->tags, compareKv);
   for (int j = 0; j < taosArrayGetSize(rName->tags); ++j) {
     taosStringBuilderAppendChar(&sb, ',');
-    SSmlKv* tagKv = taosArrayGetP(rName->tags, j);
+    SSmlKv* tagKv = taosArrayGet(rName->tags, j);
     taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
     taosStringBuilderAppendChar(&sb, '=');
     if (IS_VAR_DATA_TYPE(tagKv->type)) {

source/common/test/dataformatTest.cpp

@@ -117,7 +117,7 @@ STSchema *genSTSchema(int16_t nCols) {
   }
 
   STSchema *pResult = NULL;
-  pResult = tdGetSTSChemaFromSSChema(pSchema, nCols, 1);
+  pResult = tBuildTSchema(pSchema, nCols, 1);
 
   taosMemoryFree(pSchema);
   return pResult;

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

@@ -596,7 +596,7 @@ int32_t tDiskDataAddRow(SDiskDataBuilder *pBuilder, TSDBROW *pRow, STSchema *pTS
   if (pBuilder->bi.maxKey < kRow.ts) pBuilder->bi.maxKey = kRow.ts;
 
   STSDBRowIter iter = {0};
-  tsdbRowIterInit(&iter, pRow, pTSchema);
+  tsdbRowIterOpen(&iter, pRow, pTSchema);
 
   SColVal *pColVal = tsdbRowIterNext(&iter);
   for (int32_t iBuilder = 0; iBuilder < pBuilder->nBuilder; iBuilder++) {