Merge branch '3.0' into enh/TD-21211-3.0

include/common/taosdef.h

@@ -30,6 +30,11 @@ typedef int64_t tb_uid_t;
 #define IS_TSWINDOW_SPECIFIED(win) (((win).skey != INT64_MIN) || ((win).ekey != INT64_MAX))
 #define TSWINDOW_IS_EQUAL(t1, t2)  (((t1).skey == (t2).skey) && ((t1).ekey == (t2).ekey))
 
+//define show cluster alive and show db.alive
+#define SHOW_STATUS_NOT_AVAILABLE  0
+#define SHOW_STATUS_AVAILABLE      1
+#define SHOW_STATUS_HALF_AVAILABLE 2
+
 typedef enum {
   TSDB_SUPER_TABLE = 1,   // super table
   TSDB_CHILD_TABLE = 2,   // table created from super table

include/common/tcommon.h

@@ -292,6 +292,7 @@ typedef struct STableBlockDistInfo {
   uint16_t numOfFiles;
   uint32_t numOfTables;
   uint32_t numOfBlocks;
+  uint32_t numOfVgroups;
   uint64_t totalSize;
   uint64_t totalRows;
   int32_t  maxRows;

include/common/ttokendef.h

@@ -175,171 +175,172 @@
 #define TK_CONSUMERS                      157
 #define TK_SUBSCRIPTIONS                  158
 #define TK_VNODES                         159
-#define TK_LIKE                           160
-#define TK_TBNAME                         161
-#define TK_QTAGS                          162
-#define TK_AS                             163
-#define TK_INDEX                          164
-#define TK_FUNCTION                       165
-#define TK_INTERVAL                       166
-#define TK_COUNT                          167
-#define TK_LAST_ROW                       168
-#define TK_TOPIC                          169
-#define TK_WITH                           170
-#define TK_META                           171
-#define TK_CONSUMER                       172
-#define TK_GROUP                          173
-#define TK_DESC                           174
-#define TK_DESCRIBE                       175
-#define TK_RESET                          176
-#define TK_QUERY                          177
-#define TK_CACHE                          178
-#define TK_EXPLAIN                        179
-#define TK_ANALYZE                        180
-#define TK_VERBOSE                        181
-#define TK_NK_BOOL                        182
-#define TK_RATIO                          183
-#define TK_NK_FLOAT                       184
-#define TK_OUTPUTTYPE                     185
-#define TK_AGGREGATE                      186
-#define TK_BUFSIZE                        187
-#define TK_STREAM                         188
-#define TK_INTO                           189
-#define TK_TRIGGER                        190
-#define TK_AT_ONCE                        191
-#define TK_WINDOW_CLOSE                   192
-#define TK_IGNORE                         193
-#define TK_EXPIRED                        194
-#define TK_FILL_HISTORY                   195
-#define TK_SUBTABLE                       196
-#define TK_KILL                           197
-#define TK_CONNECTION                     198
-#define TK_TRANSACTION                    199
-#define TK_BALANCE                        200
-#define TK_VGROUP                         201
-#define TK_MERGE                          202
-#define TK_REDISTRIBUTE                   203
-#define TK_SPLIT                          204
-#define TK_DELETE                         205
-#define TK_INSERT                         206
-#define TK_NULL                           207
-#define TK_NK_QUESTION                    208
-#define TK_NK_ARROW                       209
-#define TK_ROWTS                          210
-#define TK_QSTART                         211
-#define TK_QEND                           212
-#define TK_QDURATION                      213
-#define TK_WSTART                         214
-#define TK_WEND                           215
-#define TK_WDURATION                      216
-#define TK_IROWTS                         217
-#define TK_ISFILLED                       218
-#define TK_CAST                           219
-#define TK_NOW                            220
-#define TK_TODAY                          221
-#define TK_TIMEZONE                       222
-#define TK_CLIENT_VERSION                 223
-#define TK_SERVER_VERSION                 224
-#define TK_SERVER_STATUS                  225
-#define TK_CURRENT_USER                   226
-#define TK_CASE                           227
-#define TK_END                            228
-#define TK_WHEN                           229
-#define TK_THEN                           230
-#define TK_ELSE                           231
-#define TK_BETWEEN                        232
-#define TK_IS                             233
-#define TK_NK_LT                          234
-#define TK_NK_GT                          235
-#define TK_NK_LE                          236
-#define TK_NK_GE                          237
-#define TK_NK_NE                          238
-#define TK_MATCH                          239
-#define TK_NMATCH                         240
-#define TK_CONTAINS                       241
-#define TK_IN                             242
-#define TK_JOIN                           243
-#define TK_INNER                          244
-#define TK_SELECT                         245
-#define TK_DISTINCT                       246
-#define TK_WHERE                          247
-#define TK_PARTITION                      248
-#define TK_BY                             249
-#define TK_SESSION                        250
-#define TK_STATE_WINDOW                   251
-#define TK_EVENT_WINDOW                   252
-#define TK_START                          253
-#define TK_SLIDING                        254
-#define TK_FILL                           255
-#define TK_VALUE                          256
-#define TK_NONE                           257
-#define TK_PREV                           258
-#define TK_LINEAR                         259
-#define TK_NEXT                           260
-#define TK_HAVING                         261
-#define TK_RANGE                          262
-#define TK_EVERY                          263
-#define TK_ORDER                          264
-#define TK_SLIMIT                         265
-#define TK_SOFFSET                        266
-#define TK_LIMIT                          267
-#define TK_OFFSET                         268
-#define TK_ASC                            269
-#define TK_NULLS                          270
-#define TK_ABORT                          271
-#define TK_AFTER                          272
-#define TK_ATTACH                         273
-#define TK_BEFORE                         274
-#define TK_BEGIN                          275
-#define TK_BITAND                         276
-#define TK_BITNOT                         277
-#define TK_BITOR                          278
-#define TK_BLOCKS                         279
-#define TK_CHANGE                         280
-#define TK_COMMA                          281
-#define TK_COMPACT                        282
-#define TK_CONCAT                         283
-#define TK_CONFLICT                       284
-#define TK_COPY                           285
-#define TK_DEFERRED                       286
-#define TK_DELIMITERS                     287
-#define TK_DETACH                         288
-#define TK_DIVIDE                         289
-#define TK_DOT                            290
-#define TK_EACH                           291
-#define TK_FAIL                           292
-#define TK_FILE                           293
-#define TK_FOR                            294
-#define TK_GLOB                           295
-#define TK_ID                             296
-#define TK_IMMEDIATE                      297
-#define TK_IMPORT                         298
-#define TK_INITIALLY                      299
-#define TK_INSTEAD                        300
-#define TK_ISNULL                         301
-#define TK_KEY                            302
-#define TK_MODULES                        303
-#define TK_NK_BITNOT                      304
-#define TK_NK_SEMI                        305
-#define TK_NOTNULL                        306
-#define TK_OF                             307
-#define TK_PLUS                           308
-#define TK_PRIVILEGE                      309
-#define TK_RAISE                          310
-#define TK_REPLACE                        311
-#define TK_RESTRICT                       312
-#define TK_ROW                            313
-#define TK_SEMI                           314
-#define TK_STAR                           315
-#define TK_STATEMENT                      316
-#define TK_STRICT                         317
-#define TK_STRING                         318
-#define TK_TIMES                          319
-#define TK_UPDATE                         320
-#define TK_VALUES                         321
-#define TK_VARIABLE                       322
-#define TK_VIEW                           323
-#define TK_WAL                            324
+#define TK_ALIVE                          160
+#define TK_LIKE                           161
+#define TK_TBNAME                         162
+#define TK_QTAGS                          163
+#define TK_AS                             164
+#define TK_INDEX                          165
+#define TK_FUNCTION                       166
+#define TK_INTERVAL                       167
+#define TK_COUNT                          168
+#define TK_LAST_ROW                       169
+#define TK_TOPIC                          170
+#define TK_WITH                           171
+#define TK_META                           172
+#define TK_CONSUMER                       173
+#define TK_GROUP                          174
+#define TK_DESC                           175
+#define TK_DESCRIBE                       176
+#define TK_RESET                          177
+#define TK_QUERY                          178
+#define TK_CACHE                          179
+#define TK_EXPLAIN                        180
+#define TK_ANALYZE                        181
+#define TK_VERBOSE                        182
+#define TK_NK_BOOL                        183
+#define TK_RATIO                          184
+#define TK_NK_FLOAT                       185
+#define TK_OUTPUTTYPE                     186
+#define TK_AGGREGATE                      187
+#define TK_BUFSIZE                        188
+#define TK_STREAM                         189
+#define TK_INTO                           190
+#define TK_TRIGGER                        191
+#define TK_AT_ONCE                        192
+#define TK_WINDOW_CLOSE                   193
+#define TK_IGNORE                         194
+#define TK_EXPIRED                        195
+#define TK_FILL_HISTORY                   196
+#define TK_SUBTABLE                       197
+#define TK_KILL                           198
+#define TK_CONNECTION                     199
+#define TK_TRANSACTION                    200
+#define TK_BALANCE                        201
+#define TK_VGROUP                         202
+#define TK_MERGE                          203
+#define TK_REDISTRIBUTE                   204
+#define TK_SPLIT                          205
+#define TK_DELETE                         206
+#define TK_INSERT                         207
+#define TK_NULL                           208
+#define TK_NK_QUESTION                    209
+#define TK_NK_ARROW                       210
+#define TK_ROWTS                          211
+#define TK_QSTART                         212
+#define TK_QEND                           213
+#define TK_QDURATION                      214
+#define TK_WSTART                         215
+#define TK_WEND                           216
+#define TK_WDURATION                      217
+#define TK_IROWTS                         218
+#define TK_ISFILLED                       219
+#define TK_CAST                           220
+#define TK_NOW                            221
+#define TK_TODAY                          222
+#define TK_TIMEZONE                       223
+#define TK_CLIENT_VERSION                 224
+#define TK_SERVER_VERSION                 225
+#define TK_SERVER_STATUS                  226
+#define TK_CURRENT_USER                   227
+#define TK_CASE                           228
+#define TK_END                            229
+#define TK_WHEN                           230
+#define TK_THEN                           231
+#define TK_ELSE                           232
+#define TK_BETWEEN                        233
+#define TK_IS                             234
+#define TK_NK_LT                          235
+#define TK_NK_GT                          236
+#define TK_NK_LE                          237
+#define TK_NK_GE                          238
+#define TK_NK_NE                          239
+#define TK_MATCH                          240
+#define TK_NMATCH                         241
+#define TK_CONTAINS                       242
+#define TK_IN                             243
+#define TK_JOIN                           244
+#define TK_INNER                          245
+#define TK_SELECT                         246
+#define TK_DISTINCT                       247
+#define TK_WHERE                          248
+#define TK_PARTITION                      249
+#define TK_BY                             250
+#define TK_SESSION                        251
+#define TK_STATE_WINDOW                   252
+#define TK_EVENT_WINDOW                   253
+#define TK_START                          254
+#define TK_SLIDING                        255
+#define TK_FILL                           256
+#define TK_VALUE                          257
+#define TK_NONE                           258
+#define TK_PREV                           259
+#define TK_LINEAR                         260
+#define TK_NEXT                           261
+#define TK_HAVING                         262
+#define TK_RANGE                          263
+#define TK_EVERY                          264
+#define TK_ORDER                          265
+#define TK_SLIMIT                         266
+#define TK_SOFFSET                        267
+#define TK_LIMIT                          268
+#define TK_OFFSET                         269
+#define TK_ASC                            270
+#define TK_NULLS                          271
+#define TK_ABORT                          272
+#define TK_AFTER                          273
+#define TK_ATTACH                         274
+#define TK_BEFORE                         275
+#define TK_BEGIN                          276
+#define TK_BITAND                         277
+#define TK_BITNOT                         278
+#define TK_BITOR                          279
+#define TK_BLOCKS                         280
+#define TK_CHANGE                         281
+#define TK_COMMA                          282
+#define TK_COMPACT                        283
+#define TK_CONCAT                         284
+#define TK_CONFLICT                       285
+#define TK_COPY                           286
+#define TK_DEFERRED                       287
+#define TK_DELIMITERS                     288
+#define TK_DETACH                         289
+#define TK_DIVIDE                         290
+#define TK_DOT                            291
+#define TK_EACH                           292
+#define TK_FAIL                           293
+#define TK_FILE                           294
+#define TK_FOR                            295
+#define TK_GLOB                           296
+#define TK_ID                             297
+#define TK_IMMEDIATE                      298
+#define TK_IMPORT                         299
+#define TK_INITIALLY                      300
+#define TK_INSTEAD                        301
+#define TK_ISNULL                         302
+#define TK_KEY                            303
+#define TK_MODULES                        304
+#define TK_NK_BITNOT                      305
+#define TK_NK_SEMI                        306
+#define TK_NOTNULL                        307
+#define TK_OF                             308
+#define TK_PLUS                           309
+#define TK_PRIVILEGE                      310
+#define TK_RAISE                          311
+#define TK_REPLACE                        312
+#define TK_RESTRICT                       313
+#define TK_ROW                            314
+#define TK_SEMI                           315
+#define TK_STAR                           316
+#define TK_STATEMENT                      317
+#define TK_STRICT                         318
+#define TK_STRING                         319
+#define TK_TIMES                          320
+#define TK_UPDATE                         321
+#define TK_VALUES                         322
+#define TK_VARIABLE                       323
+#define TK_VIEW                           324
+#define TK_WAL                            325
 
 #define TK_NK_SPACE   600
 #define TK_NK_COMMENT 601

include/libs/command/command.h

@@ -22,7 +22,7 @@
 
 typedef struct SExplainCtx SExplainCtx;
 
-int32_t qExecCommand(bool sysInfoUser, SNode *pStmt, SRetrieveTableRsp **pRsp);
+int32_t qExecCommand(int64_t* pConnId, bool sysInfoUser, SNode *pStmt, SRetrieveTableRsp **pRsp);
 
 int32_t qExecStaticExplain(SQueryPlan *pDag, SRetrieveTableRsp **pRsp);
 int32_t qExecExplainBegin(SQueryPlan *pDag, SExplainCtx **pCtx, int64_t startTs);

include/libs/nodes/cmdnodes.h

@@ -40,6 +40,7 @@ extern "C" {
 #define SHOW_LOCAL_VARIABLES_RESULT_FIELD1_LEN (TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE)
 #define SHOW_LOCAL_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
 
+#define SHOW_ALIVE_RESULT_COLS 1
 #define PRIVILEGE_TYPE_MASK(n) (1 << n)
 
 #define PRIVILEGE_TYPE_ALL       PRIVILEGE_TYPE_MASK(0)
@@ -262,6 +263,11 @@ typedef struct SShowCreateDatabaseStmt {
   void*     pCfg;  // SDbCfgInfo
 } SShowCreateDatabaseStmt;
 
+typedef struct SShowAliveStmt {
+  ENodeType type;
+  char      dbName[TSDB_DB_NAME_LEN];
+} SShowAliveStmt;
+
 typedef struct SShowCreateTableStmt {
   ENodeType type;
   char      dbName[TSDB_DB_NAME_LEN];

include/libs/nodes/nodes.h

@@ -208,6 +208,8 @@ typedef enum ENodeType {
   QUERY_NODE_DELETE_STMT,
   QUERY_NODE_INSERT_STMT,
   QUERY_NODE_QUERY,
+  QUERY_NODE_SHOW_DB_ALIVE_STMT,
+  QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT,  
 
   // logic plan node
   QUERY_NODE_LOGIC_PLAN_SCAN = 1000,

source/client/inc/clientSml.h

@@ -69,6 +69,7 @@ extern "C" {
 #define VALUE     "_value"
 #define VALUE_LEN 6
 
+#define OTD_JSON_FIELDS_NUM     4
 #define MAX_RETRY_TIMES 5
 typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;
 
@@ -177,12 +178,14 @@ typedef struct {
   int32_t      lineNum;
   SSmlMsgBuf   msgBuf;
 
-//  cJSON       *root;  // for parse json
-  int8_t             offset[4];
+  cJSON       *root;  // for parse json
+  int8_t             offset[OTD_JSON_FIELDS_NUM];
   SSmlLineInfo      *lines; // element is SSmlLineInfo
+  bool               parseJsonByLib;
 
   //
   SArray      *preLineTagKV;
+  SArray      *maxTagKVs;
   SArray      *preLineColKV;
 
   SSmlLineInfo preLine;
@@ -197,7 +200,7 @@ typedef struct {
 #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)
+#define IS_SAME_KEY (maxKV->keyLen == kv.keyLen && memcmp(maxKV->key, kv.key, kv.keyLen) == 0)
 
 extern int64_t smlFactorNS[3];
 extern int64_t smlFactorS[3];
@@ -206,9 +209,9 @@ 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);
+int           smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset);
+int           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);
@@ -226,6 +229,7 @@ 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);
+uint8_t           smlGetTimestampLen(int64_t num);
 
 int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
 int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);

source/client/src/clientImpl.c

@@ -272,7 +272,7 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
 
 int32_t execLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
   SRetrieveTableRsp* pRsp = NULL;
-  int32_t            code = qExecCommand(pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp);
+  int32_t            code = qExecCommand(&pRequest->pTscObj->id, pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp);
   if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
     code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false, true);
   }
@@ -310,7 +310,7 @@ void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
     return;
   }
 
-  int32_t code = qExecCommand(pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp);
+  int32_t code = qExecCommand(&pRequest->pTscObj->id ,pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp);
   if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
     code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false, true);
   }

source/client/src/clientSml.c

@@ -1008,12 +1008,16 @@ static int32_t smlUpdateMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols
   return TSDB_CODE_SUCCESS;
 }
 
-static void smlDestroyTableInfo(SSmlTableInfo *tag) {
+static void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag) {
   for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
     SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
     taosHashCleanup(kvHash);
   }
 
+  if(info->parseJsonByLib){
+    SSmlLineInfo *key = (SSmlLineInfo *)(tag->key);
+    if(key != NULL) taosMemoryFree(key->tags);
+  }
   taosMemoryFree(tag->key);
   taosArrayDestroy(tag->cols);
   taosArrayDestroy(tag->tags);
@@ -1028,7 +1032,7 @@ void smlDestroyInfo(SSmlHandle *info) {
   NodeList *tmp = info->childTables;
   while (tmp) {
     if (tmp->data.used) {
-      smlDestroyTableInfo((SSmlTableInfo *)tmp->data.value);
+      smlDestroyTableInfo(info, (SSmlTableInfo *)tmp->data.value);
     }
     NodeList *t = tmp->next;
     taosMemoryFree(tmp);
@@ -1050,11 +1054,15 @@ void smlDestroyInfo(SSmlHandle *info) {
   taosHashCleanup(info->pVgHash);
 
   taosArrayDestroy(info->preLineTagKV);
+  taosArrayDestroy(info->maxTagKVs);
   taosArrayDestroy(info->preLineColKV);
 
   if (!info->dataFormat) {
     for (int i = 0; i < info->lineNum; i++) {
       taosArrayDestroy(info->lines[i].colArray);
+      if(info->parseJsonByLib){
+        taosMemoryFree(info->lines[i].tags);
+      }
     }
     taosMemoryFree(info->lines);
   }
@@ -1083,6 +1091,7 @@ SSmlHandle *smlBuildSmlInfo(TAOS *taos) {
   info->dataFormat = true;
 
   info->preLineTagKV = taosArrayInit(8, sizeof(SSmlKv));
+  info->maxTagKVs = taosArrayInit(8, sizeof(SSmlKv));
   info->preLineColKV = taosArrayInit(8, sizeof(SSmlKv));
 
   if (NULL == info->pVgHash) {
@@ -1251,7 +1260,7 @@ int32_t smlClearForRerun(SSmlHandle *info) {
   NodeList *pList = info->childTables;
   while (pList) {
     if (pList->data.used) {
-      smlDestroyTableInfo((SSmlTableInfo *)pList->data.value);
+      smlDestroyTableInfo(info, (SSmlTableInfo *)pList->data.value);
       pList->data.used = false;
     }
     pList = pList->next;
@@ -1267,11 +1276,13 @@ int32_t smlClearForRerun(SSmlHandle *info) {
     pList = pList->next;
   }
 
-  if (unlikely(info->lines != NULL)) {
-    uError("SML:0x%" PRIx64 " info->lines != NULL", info->id);
-    return TSDB_CODE_SML_INVALID_DATA;
+  if (!info->dataFormat){
+    if (unlikely(info->lines != NULL)) {
+      uError("SML:0x%" PRIx64 " info->lines != NULL", info->id);
+      return TSDB_CODE_SML_INVALID_DATA;
+    }
+    info->lines = (SSmlLineInfo *)taosMemoryCalloc(info->lineNum, sizeof(SSmlLineInfo));
   }
-  info->lines = (SSmlLineInfo *)taosMemoryCalloc(info->lineNum, sizeof(SSmlLineInfo));
 
   memset(&info->preLine, 0, sizeof(SSmlLineInfo));
   info->currSTableMeta = NULL;

source/client/src/clientSmlLine.c

@@ -57,52 +57,6 @@ uint8_t smlPrecisionConvert[7] = {TSDB_TIME_PRECISION_NANO, TSDB_TIME_PRECISION_
                                   TSDB_TIME_PRECISION_SECONDS, TSDB_TIME_PRECISION_MILLI, TSDB_TIME_PRECISION_MICRO,
                                   TSDB_TIME_PRECISION_NANO};
 
-static bool smlParseBool(SSmlKv *kvVal) {
-  const char *pVal = kvVal->value;
-  int32_t     len = kvVal->length;
-  if ((len == 1) && (pVal[0] == 't' || pVal[0] == 'T')) {
-    kvVal->i = TSDB_TRUE;
-    return true;
-  }
-
-  if ((len == 1) && (pVal[0] == 'f' || pVal[0] == 'F')) {
-    kvVal->i = TSDB_FALSE;
-    return true;
-  }
-
-  if ((len == 4) && !strncasecmp(pVal, "true", len)) {
-    kvVal->i = TSDB_TRUE;
-    return true;
-  }
-  if ((len == 5) && !strncasecmp(pVal, "false", len)) {
-    kvVal->i = TSDB_FALSE;
-    return true;
-  }
-  return false;
-}
-
-static bool smlIsBinary(const char *pVal, uint16_t len) {
-  // binary: "abc"
-  if (len < 2) {
-    return false;
-  }
-  if (pVal[0] == '"' && pVal[len - 1] == '"') {
-    return true;
-  }
-  return false;
-}
-
-static bool smlIsNchar(const char *pVal, uint16_t len) {
-  // nchar: L"abc"
-  if (len < 3) {
-    return false;
-  }
-  if (pVal[1] == '"' && pVal[len - 1] == '"' && (pVal[0] == 'l' || pVal[0] == 'L')) {
-    return true;
-  }
-  return false;
-}
-
 static int64_t smlParseInfluxTime(SSmlHandle *info, const char *data, int32_t len) {
   uint8_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;
 
@@ -189,6 +143,7 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd,
 
   int     cnt = 0;
   SArray *preLineKV = info->preLineTagKV;
+  SArray *maxKVs = info->maxTagKVs;
   bool    isSuperKVInit = true;
   SArray *superKV = NULL;
   if(info->dataFormat){
@@ -212,12 +167,12 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd,
       if(unlikely(taosArrayGetSize(superKV) == 0)){
         isSuperKVInit = false;
       }
-      taosArraySetSize(preLineKV, 0);
+      taosArraySetSize(maxKVs, 0);
     }
   }else{
-    taosArraySetSize(preLineKV, 0);
+    taosArraySetSize(maxKVs, 0);
   }
-
+  taosArraySetSize(preLineKV, 0);
 
   while (*sql < sqlEnd) {
     if (unlikely(IS_SPACE(*sql))) {
@@ -295,14 +250,14 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd,
       }
 
       if(isSameMeasure){
-        if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
+        if(unlikely(cnt >= taosArrayGetSize(maxKVs))) {
           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;
+        SSmlKv *maxKV = (SSmlKv *)taosArrayGet(maxKVs, cnt);
+        if(unlikely(kv.length > maxKV->length)){
+          maxKV->length = kv.length;
           SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, currElement->measure, currElement->measureLen, NULL);
           ASSERT(tableMeta != NULL);
 
@@ -322,11 +277,11 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd,
             info->reRun      = true;
             return TSDB_CODE_SUCCESS;
           }
-          SSmlKv *preKV = (SSmlKv *)taosArrayGet(superKV, cnt);
-          if(unlikely(kv.length > preKV->length)) {
-            preKV->length = kv.length;
+          SSmlKv *maxKV = (SSmlKv *)taosArrayGet(superKV, cnt);
+          if(unlikely(kv.length > maxKV->length)) {
+            maxKV->length = kv.length;
           }else{
-            kv.length = preKV->length;
+            kv.length = maxKV->length;
           }
           info->needModifySchema = true;
 
@@ -338,11 +293,12 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd,
         }else{
           taosArrayPush(superKV, &kv);
         }
-        taosArrayPush(preLineKV, &kv);
+        taosArrayPush(maxKVs, &kv);
       }
     }else{
-      taosArrayPush(preLineKV, &kv);
+      taosArrayPush(maxKVs, &kv);
     }
+    taosArrayPush(preLineKV, &kv);
 
     cnt++;
     if(IS_SPACE(*sql)){
@@ -518,15 +474,15 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd,
           info->reRun      = true;
           return TSDB_CODE_SUCCESS;
         }
-        SSmlKv *preKV = (SSmlKv *)taosArrayGet(preLineKV, cnt);
-        if(kv.type != preKV->type){
+        SSmlKv *maxKV = (SSmlKv *)taosArrayGet(preLineKV, cnt);
+        if(kv.type != maxKV->type){
           info->dataFormat = false;
           info->reRun      = true;
           return TSDB_CODE_SUCCESS;
         }
 
-        if(unlikely(IS_VAR_DATA_TYPE(kv.type) && kv.length > preKV->length)){
-          preKV->length = kv.length;
+        if(unlikely(IS_VAR_DATA_TYPE(kv.type) && kv.length > maxKV->length)){
+          maxKV->length = kv.length;
           SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, currElement->measure, currElement->measureLen, NULL);
           ASSERT(tableMeta != NULL);
 
@@ -546,18 +502,18 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd,
             info->reRun      = true;
             return TSDB_CODE_SUCCESS;
           }
-          SSmlKv *preKV = (SSmlKv *)taosArrayGet(superKV, cnt);
-          if(unlikely(kv.type != preKV->type)){
+          SSmlKv *maxKV = (SSmlKv *)taosArrayGet(superKV, cnt);
+          if(unlikely(kv.type != maxKV->type)){
             info->dataFormat = false;
             info->reRun      = true;
             return TSDB_CODE_SUCCESS;
           }
 
           if(IS_VAR_DATA_TYPE(kv.type)){
-            if(kv.length > preKV->length) {
-              preKV->length = kv.length;
+            if(kv.length > maxKV->length) {
+              maxKV->length = kv.length;
             }else{
-              kv.length = preKV->length;
+              kv.length = maxKV->length;
             }
             info->needModifySchema = true;
           }

source/client/src/clientSmlTelnet.c

@@ -23,6 +23,11 @@
 int32_t is_same_child_table_telnet(const void *a, const void *b){
   SSmlLineInfo *t1 = (SSmlLineInfo *)a;
   SSmlLineInfo *t2 = (SSmlLineInfo *)b;
+//  uError("is_same_child_table_telnet len:%d,%d %s,%s @@@ len:%d,%d %s,%s", t1->measureLen, t2->measureLen,
+//         t1->measure, t2->measure, t1->tagsLen, t2->tagsLen, t1->tags, t2->tags);
+  if(t1 == NULL || t2 == NULL || t1->measure == NULL || t2->measure == NULL
+      || t1->tags == NULL || t2->tags == NULL)
+    return 1;
   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;
 }
@@ -73,6 +78,7 @@ static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SS
 
   int     cnt = 0;
   SArray *preLineKV = info->preLineTagKV;
+  SArray *maxKVs = info->maxTagKVs;
   bool    isSuperKVInit = true;
   SArray *superKV = NULL;
   if(info->dataFormat){
@@ -96,12 +102,13 @@ static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SS
       if(unlikely(taosArrayGetSize(superKV) == 0)){
         isSuperKVInit = false;
       }
-      taosArraySetSize(preLineKV, 0);
+      taosArraySetSize(maxKVs, 0);
     }
   }else{
-    taosArraySetSize(preLineKV, 0);
+    taosArraySetSize(maxKVs, 0);
   }
 
+  taosArraySetSize(preLineKV, 0);
   const char *sql = data;
   while (sql < sqlEnd) {
     JUMP_SPACE(sql, sqlEnd)
@@ -168,14 +175,14 @@ static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SS
       }
 
       if(isSameMeasure){
-        if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
+        if(unlikely(cnt >= taosArrayGetSize(maxKVs))) {
           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;
+        SSmlKv *maxKV = (SSmlKv *)taosArrayGet(maxKVs, cnt);
+        if(unlikely(kv.length > maxKV->length)){
+          maxKV->length = kv.length;
           SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
           ASSERT(tableMeta != NULL);
 
@@ -195,11 +202,11 @@ static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SS
             info->reRun      = true;
             return TSDB_CODE_SUCCESS;
           }
-          SSmlKv *preKV = (SSmlKv *)taosArrayGet(superKV, cnt);
-          if(unlikely(kv.length > preKV->length)) {
-            preKV->length = kv.length;
+          SSmlKv *maxKV = (SSmlKv *)taosArrayGet(superKV, cnt);
+          if(unlikely(kv.length > maxKV->length)) {
+            maxKV->length = kv.length;
           }else{
-            kv.length = preKV->length;
+            kv.length = maxKV->length;
           }
           info->needModifySchema = true;
 
@@ -211,11 +218,12 @@ static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SS
         }else{
           taosArrayPush(superKV, &kv);
         }
-        taosArrayPush(preLineKV, &kv);
+        taosArrayPush(maxKVs, &kv);
       }
     }else{
-      taosArrayPush(preLineKV, &kv);
+      taosArrayPush(maxKVs, &kv);
     }
+    taosArrayPush(preLineKV, &kv);
     cnt++;
   }
   SSmlTableInfo *tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements, POINTER_BYTES, is_same_child_table_telnet);

source/client/test/smlTest.cpp

@@ -411,28 +411,28 @@ TEST(testCase, smlParseCols_Test) {
   smlDestroyInfo(info);
 }
 
-//TEST(testCase, smlGetTimestampLen_Test) {
-//  uint8_t len = smlGetTimestampLen(0);
-//  ASSERT_EQ(len, 1);
-//
-//  len = smlGetTimestampLen(1);
-//  ASSERT_EQ(len, 1);
-//
-//  len = smlGetTimestampLen(10);
-//  ASSERT_EQ(len, 2);
-//
-//  len = smlGetTimestampLen(390);
-//  ASSERT_EQ(len, 3);
-//
-//  len = smlGetTimestampLen(-1);
-//  ASSERT_EQ(len, 1);
-//
-//  len = smlGetTimestampLen(-10);
-//  ASSERT_EQ(len, 2);
-//
-//  len = smlGetTimestampLen(-390);
-//  ASSERT_EQ(len, 3);
-//}
+TEST(testCase, smlGetTimestampLen_Test) {
+  uint8_t len = smlGetTimestampLen(0);
+  ASSERT_EQ(len, 1);
+
+  len = smlGetTimestampLen(1);
+  ASSERT_EQ(len, 1);
+
+  len = smlGetTimestampLen(10);
+  ASSERT_EQ(len, 2);
+
+  len = smlGetTimestampLen(390);
+  ASSERT_EQ(len, 3);
+
+  len = smlGetTimestampLen(-1);
+  ASSERT_EQ(len, 1);
+
+  len = smlGetTimestampLen(-10);
+  ASSERT_EQ(len, 2);
+
+  len = smlGetTimestampLen(-390);
+  ASSERT_EQ(len, 3);
+}
 
 TEST(testCase, smlParseNumber_Test) {
   SSmlKv     kv = {0};
@@ -503,35 +503,35 @@ TEST(testCase, smlParseTelnetLine_Test) {
   smlDestroyInfo(info);
 }
 
-TEST(testCase, smlParseTelnetLine_diff_json_type2_Test) {
-  SSmlHandle *info = smlBuildSmlInfo(NULL);
-  info->protocol = TSDB_SML_JSON_PROTOCOL;
-  ASSERT_NE(info, nullptr);
-
-  const char *sql[] = {
-      "[{\"metric\":\"sys.cpu.nice\",\"timestamp\": 1346846400,\"value\": 18,\"tags\": {\"host\": \"lga\"}},{\"metric\": \"sys.sdfa\",\"timestamp\": 1346846400,\"value\": \"18\",\"tags\": {\"host\": 8932}},]",
-  };
-  for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
-    char *dataPointStart = (char *)sql[i];
-    int8_t offset[4] = {0};
-    while (1) {
-      SSmlLineInfo elements = {0};
-      if(offset[0] == 0){
-        smlJsonParseObjFirst(&dataPointStart, &elements, offset);
-      }else{
-        smlJsonParseObj(&dataPointStart, &elements, offset);
-      }
-      if(*dataPointStart == '\0') break;
-
-      SArray *tags = smlJsonParseTags(elements.tags, elements.tags + elements.tagsLen);
-      size_t num = taosArrayGetSize(tags);
-      ASSERT_EQ(num, 1);
-
-      taosArrayDestroy(tags);
-    }
-  }
-  smlDestroyInfo(info);
-}
+//TEST(testCase, smlParseTelnetLine_diff_json_type2_Test) {
+//  SSmlHandle *info = smlBuildSmlInfo(NULL);
+//  info->protocol = TSDB_SML_JSON_PROTOCOL;
+//  ASSERT_NE(info, nullptr);
+//
+//  const char *sql[] = {
+//      "[{\"metric\":\"sys.cpu.nice\",\"timestamp\": 1346846400,\"value\": 18,\"tags\": {\"host\": \"lga\"}},{\"metric\": \"sys.sdfa\",\"timestamp\": 1346846400,\"value\": \"18\",\"tags\": {\"host\": 8932}},]",
+//  };
+//  for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) {
+//    char *dataPointStart = (char *)sql[i];
+//    int8_t offset[4] = {0};
+//    while (1) {
+//      SSmlLineInfo elements = {0};
+//      if(offset[0] == 0){
+//        smlJsonParseObjFirst(&dataPointStart, &elements, offset);
+//      }else{
+//        smlJsonParseObj(&dataPointStart, &elements, offset);
+//      }
+//      if(*dataPointStart == '\0') break;
+//
+//      SArray *tags = smlJsonParseTags(elements.tags, elements.tags + elements.tagsLen);
+//      size_t num = taosArrayGetSize(tags);
+//      ASSERT_EQ(num, 1);
+//
+//      taosArrayDestroy(tags);
+//    }
+//  }
+//  smlDestroyInfo(info);
+//}
 
 TEST(testCase, smlParseNumber_performance_Test) {
   char       msg[256] = {0};

source/common/src/tdatablock.c

@@ -1879,7 +1879,7 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag) {
   char    pBuf[128] = {0};
   int32_t sz = taosArrayGetSize(dataBlocks);
   for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock* pDataBlock = taosArrayGetP(dataBlocks, i);
+    SSDataBlock* pDataBlock = taosArrayGet(dataBlocks, i);
     size_t       numOfCols = taosArrayGetSize(pDataBlock->pDataBlock);
 
     int32_t rows = pDataBlock->info.rows;
@@ -1891,21 +1891,37 @@ void blockDebugShowDataBlocks(const SArray* dataBlocks, const char* flag) {
       for (int32_t k = 0; k < numOfCols; k++) {
         SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
         void*            var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
+        if (k == 0) {
+          printf("cols:%d |", (int32_t)numOfCols);
+        }
         if (colDataIsNull(pColInfoData, rows, j, NULL)) {
           printf(" %15s |", "NULL");
           continue;
         }
+
         switch (pColInfoData->info.type) {
           case TSDB_DATA_TYPE_TIMESTAMP:
             formatTimestamp(pBuf, *(uint64_t*)var, TSDB_TIME_PRECISION_MILLI);
             printf(" %25s |", pBuf);
             break;
           case TSDB_DATA_TYPE_BOOL:
-            printf(" %15d |", *(int32_t*)var);
+            printf(" %15" PRIi8 " |", *(int8_t*)var);
+            break;
+          case TSDB_DATA_TYPE_TINYINT:
+            printf(" %15" PRIi8 " |", *(int8_t*)var);
+            break;
+          case TSDB_DATA_TYPE_SMALLINT:
+            printf(" %15" PRIi16 " |", *(int16_t*)var);
             break;
           case TSDB_DATA_TYPE_INT:
             printf(" %15d |", *(int32_t*)var);
             break;
+          case TSDB_DATA_TYPE_UTINYINT:
+            printf(" %15" PRIu8 " |", *(uint8_t*)var);
+            break;
+          case TSDB_DATA_TYPE_USMALLINT:
+            printf(" %15" PRIu16 " |", *(uint16_t*)var);
+            break;
           case TSDB_DATA_TYPE_UINT:
             printf(" %15u |", *(uint32_t*)var);
             break;

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

@@ -182,9 +182,9 @@ int32_t         tqOffsetDelete(STqOffsetStore* pStore, const char* subscribeKey)
 int32_t         tqOffsetCommitFile(STqOffsetStore* pStore);
 
 // tqSink
-// void tqSinkToTableMerge(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
-// void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
-void tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
+int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBlock* pDataBlock,
+                         SBatchDeleteReq* deleteReq);
+void    tqSinkToTablePipeline2(SStreamTask* pTask, void* vnode, int64_t ver, void* data);
 
 // tqOffset
 char*   tqOffsetBuildFName(const char* path, int32_t fVer);

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

@@ -203,10 +203,6 @@ int32_t tqProcessTaskRecoverFinishReq(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqProcessTaskRecoverFinishRsp(STQ* pTq, SRpcMsg* pMsg);
 int32_t tqCheckLogInWal(STQ* pTq, int64_t version);
 
-int32_t tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pSchema,
-                        SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,
-                        SBatchDeleteReq* pDeleteReq, void** ppData, int32_t* pLen);
-
 // sma
 int32_t smaInit();
 void    smaCleanUp();

source/dnode/vnode/src/sma/smaTimeRange.c

@@ -14,6 +14,7 @@
  */
 
 #include "sma.h"
+#include "tq.h"
 #include "tsdb.h"
 
 #define SMA_STORAGE_MINUTES_MAX  86400
@@ -155,6 +156,200 @@ _exit:
   return code;
 }
 
+int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *pTSchema,
+                         SSchemaWrapper *pTagSchemaWrapper, bool createTb, int64_t suid, const char *stbFullName,
+                         SBatchDeleteReq *pDeleteReq, void **ppData, int32_t *pLen) {
+  void        *pBuf = NULL;
+  int32_t      len = 0;
+  SSubmitReq2 *pReq = NULL;
+  SArray      *tagArray = NULL;
+  SArray      *createTbArray = NULL;
+  SArray      *pVals = NULL;
+
+  int32_t sz = taosArrayGetSize(pBlocks);
+
+  if (!(tagArray = taosArrayInit(1, sizeof(STagVal)))) {
+    goto _end;
+  }
+
+  if (!(createTbArray = taosArrayInit(sz, POINTER_BYTES))) {
+    goto _end;
+  }
+
+  if (!(pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2)))) {
+    terrno = TSDB_CODE_OUT_OF_MEMORY;
+    goto _end;
+  }
+
+  if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
+    goto _end;
+  }
+
+  // create table req
+  if (createTb) {
+    for (int32_t i = 0; i < sz; ++i) {
+      SSDataBlock   *pDataBlock = taosArrayGet(pBlocks, i);
+      SVCreateTbReq *pCreateTbReq = NULL;
+      if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
+        taosArrayPush(createTbArray, &pCreateTbReq);
+        continue;
+      }
+
+      if (!(pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateStbReq)))) {
+        goto _end;
+      };
+
+      // don't move to the end of loop as to destroy in the end of func when error occur
+      taosArrayPush(createTbArray, &pCreateTbReq);
+
+      // set const
+      pCreateTbReq->flags = 0;
+      pCreateTbReq->type = TSDB_CHILD_TABLE;
+      pCreateTbReq->ctb.suid = suid;
+
+      // set super table name
+      SName name = {0};
+      tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
+      pCreateTbReq->ctb.stbName = strdup((char *)tNameGetTableName(&name));  // strdup(stbFullName);
+
+      // set tag content
+      taosArrayClear(tagArray);
+      STagVal tagVal = {
+          .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
+          .type = TSDB_DATA_TYPE_UBIGINT,
+          .i64 = (int64_t)pDataBlock->info.id.groupId,
+      };
+      taosArrayPush(tagArray, &tagVal);
+      pCreateTbReq->ctb.tagNum = taosArrayGetSize(tagArray);
+
+      STag *pTag = NULL;
+      tTagNew(tagArray, 1, false, &pTag);
+      if (pTag == NULL) {
+        terrno = TSDB_CODE_OUT_OF_MEMORY;
+        goto _end;
+      }
+      pCreateTbReq->ctb.pTag = (uint8_t *)pTag;
+
+      // set tag name
+      SArray *tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
+      char    tagNameStr[TSDB_COL_NAME_LEN] = {0};
+      strcpy(tagNameStr, "group_id");
+      taosArrayPush(tagName, tagNameStr);
+      pCreateTbReq->ctb.tagName = tagName;
+
+      // set table name
+      if (pDataBlock->info.parTbName[0]) {
+        pCreateTbReq->name = strdup(pDataBlock->info.parTbName);
+      } else {
+        pCreateTbReq->name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
+      }
+    }
+  }
+
+  // SSubmitTbData req
+  for (int32_t i = 0; i < sz; ++i) {
+    SSDataBlock *pDataBlock = taosArrayGet(pBlocks, i);
+    if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
+      pDeleteReq->suid = suid;
+      pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
+      tqBuildDeleteReq(pVnode, stbFullName, pDataBlock, pDeleteReq);
+      continue;
+    }
+
+    int32_t rows = pDataBlock->info.rows;
+
+    SSubmitTbData *pTbData = (SSubmitTbData *)taosMemoryCalloc(1, sizeof(SSubmitTbData));
+    if (!pTbData) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      goto _end;
+    }
+
+    if (!(pTbData->aRowP = taosArrayInit(rows, sizeof(SRow *)))) {
+      taosMemoryFree(pTbData);
+      goto _end;
+    }
+    pTbData->suid = suid;
+    pTbData->uid = 0;  // uid is assigned by vnode
+    pTbData->sver = pTSchema->version;
+
+    if (createTb) {
+      pTbData->pCreateTbReq = taosArrayGetP(createTbArray, i);
+      if (pTbData->pCreateTbReq) pTbData->flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
+    }
+
+    if (!pVals && !(pVals = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal)))) {
+      taosArrayDestroy(pTbData->aRowP);
+      taosMemoryFree(pTbData);
+      goto _end;
+    }
+
+    for (int32_t j = 0; j < rows; ++j) {
+      taosArrayClear(pVals);
+      for (int32_t k = 0; k < pTSchema->numOfCols; k++) {
+        const STColumn  *pCol = &pTSchema->columns[k];
+        SColumnInfoData *pColData = taosArrayGet(pDataBlock->pDataBlock, k);
+        if (colDataIsNull_s(pColData, j)) {
+          SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
+          taosArrayPush(pVals, &cv);
+        } else {
+          void *data = colDataGetData(pColData, j);
+          if (IS_STR_DATA_TYPE(pCol->type)) {
+            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);
+          } else {
+            SValue sv;
+            memcpy(&sv.val, data, tDataTypes[pCol->type].bytes);
+            SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, sv);
+            taosArrayPush(pVals, &cv);
+          }
+        }
+      }
+      SRow *pRow = NULL;
+      if ((terrno = tRowBuild(pVals, (STSchema *)pTSchema, &pRow)) < 0) {
+        tDestroySSubmitTbData(pTbData, TSDB_MSG_FLG_ENCODE);
+        goto _end;
+      }
+      ASSERT(pRow);
+      taosArrayPush(pTbData->aRowP, &pRow);
+    }
+
+    taosArrayPush(pReq->aSubmitTbData, pTbData);
+  }
+
+  // encode
+  tEncodeSize(tEncodeSSubmitReq2, pReq, len, terrno);
+  if (TSDB_CODE_SUCCESS == terrno) {
+    SEncoder encoder;
+    len += sizeof(SMsgHead);
+    pBuf = rpcMallocCont(len);
+    if (NULL == pBuf) {
+      goto _end;
+    }
+    ((SMsgHead *)pBuf)->vgId = TD_VID(pVnode);
+    ((SMsgHead *)pBuf)->contLen = htonl(len);
+    tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SMsgHead)), len - sizeof(SMsgHead));
+    if (tEncodeSSubmitReq2(&encoder, pReq) < 0) {
+      terrno = TSDB_CODE_OUT_OF_MEMORY;
+      /*vError("failed to encode submit req since %s", terrstr());*/
+    }
+    tEncoderClear(&encoder);
+  }
+_end:
+  taosArrayDestroy(tagArray);
+  taosArrayDestroy(pVals);
+  tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE);
+
+  if (terrno != 0) {
+    rpcFreeCont(pBuf);
+    taosArrayDestroy(pDeleteReq->deleteReqs);
+    return TSDB_CODE_FAILED;
+  }
+  if (ppData) *ppData = pBuf;
+  if (pLen) *pLen = len;
+  return TSDB_CODE_SUCCESS;
+}
+
 /**
  * @brief Insert/Update Time-range-wise SMA data.
  *
@@ -220,8 +415,9 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
   void           *pSubmitReq = NULL;
   int32_t         contLen = 0;
 
-  if (tqBlockToSubmit(pSma->pVnode, (const SArray *)msg, pTsmaStat->pTSchema, &pTsmaStat->pTSma->schemaTag, true,
-                      pTsmaStat->pTSma->dstTbUid, pTsmaStat->pTSma->dstTbName, &deleteReq, &pSubmitReq, &contLen) < 0) {
+  if (smaBlockToSubmit(pSma->pVnode, (const SArray *)msg, pTsmaStat->pTSchema, &pTsmaStat->pTSma->schemaTag, true,
+                       pTsmaStat->pTSma->dstTbUid, pTsmaStat->pTSma->dstTbName, &deleteReq, &pSubmitReq,
+                       &contLen) < 0) {
     smaError("vgId:%d, failed to gen submit msg while tsma insert for smaIndex %" PRIi64 " since %s", SMA_VID(pSma),
              indexUid, tstrerror(terrno));
     goto _err;
@@ -241,7 +437,7 @@ static int32_t tdProcessTSmaInsertImpl(SSma *pSma, int64_t indexUid, const char
   SRpcMsg submitReqMsg = {
       .msgType = TDMT_VND_SUBMIT,
       .pCont = pSubmitReq,
-      .contLen = ntohl(contLen),
+      .contLen = contLen,
   };
 
   if (tmsgPutToQueue(&pSma->pVnode->msgCb, WRITE_QUEUE, &submitReqMsg) < 0) {

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

@@ -71,427 +71,6 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
   return 0;
 }
 
-#if 0
-SSubmitReq* tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pTSchema,
-                            SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,
-                            SBatchDeleteReq* pDeleteReq) {
-  SSubmitReq* ret = NULL;
-  SArray*     schemaReqs = NULL;
-  SArray*     schemaReqSz = NULL;
-  SArray*     tagArray = taosArrayInit(1, sizeof(STagVal));
-  if (!tagArray) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    return NULL;
-  }
-
-  int32_t sz = taosArrayGetSize(pBlocks);
-
-  if (createTb) {
-    schemaReqs = taosArrayInit(sz, sizeof(void*));
-    schemaReqSz = taosArrayInit(sz, sizeof(int32_t));
-    for (int32_t i = 0; i < sz; i++) {
-      SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
-      if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
-        int32_t padding1 = 0;
-        void*   padding2 = NULL;
-        taosArrayPush(schemaReqSz, &padding1);
-        taosArrayPush(schemaReqs, &padding2);
-        continue;
-      }
-
-      //      STag* pTag = NULL;
-      //      taosArrayClear(tagArray);
-      //      SArray *tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
-      //      for(int j = 0; j < pTagSchemaWrapper->nCols; j++){
-      //        STagVal tagVal = {
-      //            .cid = pTagSchemaWrapper->pSchema[j].colId,
-      //            .type = pTagSchemaWrapper->pSchema[j].type,
-      //            .i64 = (int64_t)pDataBlock->info.id.groupId,
-      //        };
-      //        taosArrayPush(tagArray, &tagVal);
-      //        taosArrayPush(tagName, pTagSchemaWrapper->pSchema[j].name);
-      //      }
-      //
-      //      tTagNew(tagArray, 1, false, &pTag);
-      //      if (pTag == NULL) {
-      //        terrno = TSDB_CODE_OUT_OF_MEMORY;
-      //        taosArrayDestroy(tagArray);
-      //        taosArrayDestroy(tagName);
-      //        return NULL;
-      //      }
-
-      SVCreateTbReq createTbReq = {0};
-
-      // set const
-      createTbReq.flags = 0;
-      createTbReq.type = TSDB_CHILD_TABLE;
-      createTbReq.ctb.suid = suid;
-
-      // set super table name
-      SName name = {0};
-      tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
-      createTbReq.ctb.stbName = strdup((char*)tNameGetTableName(&name));  // strdup(stbFullName);
-
-      // set tag content
-      taosArrayClear(tagArray);
-      STagVal tagVal = {
-          .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
-          .type = TSDB_DATA_TYPE_UBIGINT,
-          .i64 = (int64_t)pDataBlock->info.id.groupId,
-      };
-      taosArrayPush(tagArray, &tagVal);
-      createTbReq.ctb.tagNum = taosArrayGetSize(tagArray);
-
-      STag* pTag = NULL;
-      tTagNew(tagArray, 1, false, &pTag);
-      if (pTag == NULL) {
-        terrno = TSDB_CODE_OUT_OF_MEMORY;
-        taosArrayDestroy(tagArray);
-        taosArrayDestroyP(schemaReqs, taosMemoryFree);
-        taosArrayDestroy(schemaReqSz);
-        return NULL;
-      }
-      createTbReq.ctb.pTag = (uint8_t*)pTag;
-
-      // set tag name
-      SArray* tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
-      char    tagNameStr[TSDB_COL_NAME_LEN] = {0};
-      strcpy(tagNameStr, "group_id");
-      taosArrayPush(tagName, tagNameStr);
-      createTbReq.ctb.tagName = tagName;
-
-      // set table name
-      if (pDataBlock->info.parTbName[0]) {
-        createTbReq.name = strdup(pDataBlock->info.parTbName);
-      } else {
-        createTbReq.name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
-      }
-
-      // save schema len
-      int32_t code;
-      int32_t schemaLen;
-      tEncodeSize(tEncodeSVCreateTbReq, &createTbReq, schemaLen, code);
-      if (code < 0) {
-        tdDestroySVCreateTbReq(&createTbReq);
-        taosArrayDestroy(tagArray);
-        taosArrayDestroyP(schemaReqs, taosMemoryFree);
-        taosArrayDestroy(schemaReqSz);
-        return NULL;
-      }
-      taosArrayPush(schemaReqSz, &schemaLen);
-
-      // save schema str
-      void* schemaStr = taosMemoryMalloc(schemaLen);
-      if (schemaStr == NULL) {
-        terrno = TSDB_CODE_OUT_OF_MEMORY;
-        tdDestroySVCreateTbReq(&createTbReq);
-        taosArrayDestroy(tagArray);
-        taosArrayDestroyP(schemaReqs, taosMemoryFree);
-        taosArrayDestroy(schemaReqSz);
-        return NULL;
-      }
-      taosArrayPush(schemaReqs, &schemaStr);
-
-      SEncoder encoder = {0};
-      tEncoderInit(&encoder, schemaStr, schemaLen);
-      code = tEncodeSVCreateTbReq(&encoder, &createTbReq);
-      if (code < 0) {
-        terrno = TSDB_CODE_OUT_OF_MEMORY;
-        tdDestroySVCreateTbReq(&createTbReq);
-        taosArrayDestroy(tagArray);
-        taosArrayDestroyP(schemaReqs, taosMemoryFree);
-        taosArrayDestroy(schemaReqSz);
-        tEncoderClear(&encoder);
-        return NULL;
-      }
-      tEncoderClear(&encoder);
-      tdDestroySVCreateTbReq(&createTbReq);
-    }
-  }
-  taosArrayDestroy(tagArray);
-
-  // cal size
-  int32_t cap = sizeof(SSubmitReq);
-  for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
-    if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
-      continue;
-    }
-    int32_t rows = pDataBlock->info.rows;
-    /*int32_t rowSize = pDataBlock->info.rowSize;*/
-    int32_t maxLen = TD_ROW_MAX_BYTES_FROM_SCHEMA(pTSchema);
-
-    int32_t schemaLen = 0;
-    if (createTb) {
-      schemaLen = *(int32_t*)taosArrayGet(schemaReqSz, i);
-    }
-    cap += sizeof(SSubmitBlk) + schemaLen + rows * maxLen;
-  }
-
-  // assign data
-  ret = rpcMallocCont(cap);
-  ret->header.vgId = pVnode->config.vgId;
-  ret->length = sizeof(SSubmitReq);
-  ret->numOfBlocks = htonl(sz);
-
-  SSubmitBlk* blkHead = POINTER_SHIFT(ret, sizeof(SSubmitReq));
-  for (int32_t i = 0; i < sz; i++) {
-    SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
-    if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
-      pDeleteReq->suid = suid;
-      pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
-      tqBuildDeleteReq(pVnode, stbFullName, pDataBlock, pDeleteReq);
-      continue;
-    }
-
-    blkHead->numOfRows = htonl(pDataBlock->info.rows);
-    blkHead->sversion = htonl(pTSchema->version);
-    blkHead->suid = htobe64(suid);
-    // uid is assigned by vnode
-    blkHead->uid = 0;
-
-    int32_t rows = pDataBlock->info.rows;
-
-    int32_t dataLen = 0;
-    int32_t schemaLen = 0;
-    void*   blkSchema = POINTER_SHIFT(blkHead, sizeof(SSubmitBlk));
-    if (createTb) {
-      schemaLen = *(int32_t*)taosArrayGet(schemaReqSz, i);
-      void* schemaStr = taosArrayGetP(schemaReqs, i);
-      memcpy(blkSchema, schemaStr, schemaLen);
-    }
-    blkHead->schemaLen = htonl(schemaLen);
-
-    STSRow* rowData = POINTER_SHIFT(blkSchema, schemaLen);
-    for (int32_t j = 0; j < rows; j++) {
-      SRowBuilder rb = {0};
-      tdSRowInit(&rb, pTSchema->version);
-      tdSRowSetTpInfo(&rb, pTSchema->numOfCols, pTSchema->flen);
-      tdSRowResetBuf(&rb, rowData);
-
-      for (int32_t k = 0; k < pTSchema->numOfCols; k++) {
-        const STColumn*  pColumn = &pTSchema->columns[k];
-        SColumnInfoData* pColData = taosArrayGet(pDataBlock->pDataBlock, k);
-        if (colDataIsNull_s(pColData, j)) {
-          tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NULL, NULL, false, pColumn->offset, k);
-        } else {
-          void* data = colDataGetData(pColData, j);
-          tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, data, true, pColumn->offset, k);
-        }
-      }
-      tdSRowEnd(&rb);
-      int32_t rowLen = TD_ROW_LEN(rowData);
-      rowData = POINTER_SHIFT(rowData, rowLen);
-      dataLen += rowLen;
-    }
-    blkHead->dataLen = htonl(dataLen);
-
-    ret->length += sizeof(SSubmitBlk) + schemaLen + dataLen;
-    blkHead = POINTER_SHIFT(blkHead, sizeof(SSubmitBlk) + schemaLen + dataLen);
-  }
-
-  ret->length = htonl(ret->length);
-
-  taosArrayDestroyP(schemaReqs, taosMemoryFree);
-  taosArrayDestroy(schemaReqSz);
-
-  return ret;
-}
-#endif
-
-int32_t tqBlockToSubmit(SVnode* pVnode, const SArray* pBlocks, const STSchema* pTSchema,
-                        SSchemaWrapper* pTagSchemaWrapper, bool createTb, int64_t suid, const char* stbFullName,
-                        SBatchDeleteReq* pDeleteReq, void** ppData, int32_t* pLen) {
-  void*        pBuf = NULL;
-  int32_t      len = 0;
-  SSubmitReq2* pReq = NULL;
-  SArray*      tagArray = NULL;
-  SArray*      createTbArray = NULL;
-  SArray*      pVals = NULL;
-
-  int32_t sz = taosArrayGetSize(pBlocks);
-
-  if (!(tagArray = taosArrayInit(1, sizeof(STagVal)))) {
-    goto _end;
-  }
-
-  if (!(createTbArray = taosArrayInit(sz, POINTER_BYTES))) {
-    goto _end;
-  }
-
-  if (!(pReq = taosMemoryCalloc(1, sizeof(SSubmitReq2)))) {
-    terrno = TSDB_CODE_OUT_OF_MEMORY;
-    goto _end;
-  }
-
-  if (!(pReq->aSubmitTbData = taosArrayInit(1, sizeof(SSubmitTbData)))) {
-    goto _end;
-  }
-
-  // create table req
-  if (createTb) {
-    for (int32_t i = 0; i < sz; ++i) {
-      SSDataBlock*   pDataBlock = taosArrayGet(pBlocks, i);
-      SVCreateTbReq* pCreateTbReq = NULL;
-      if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
-        taosArrayPush(createTbArray, &pCreateTbReq);
-        continue;
-      }
-
-      if (!(pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateStbReq)))) {
-        goto _end;
-      };
-
-      // don't move to the end of loop as to destroy in the end of func when error occur
-      taosArrayPush(createTbArray, &pCreateTbReq);
-
-      // set const
-      pCreateTbReq->flags = 0;
-      pCreateTbReq->type = TSDB_CHILD_TABLE;
-      pCreateTbReq->ctb.suid = suid;
-
-      // set super table name
-      SName name = {0};
-      tNameFromString(&name, stbFullName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
-      pCreateTbReq->ctb.stbName = strdup((char*)tNameGetTableName(&name));  // strdup(stbFullName);
-
-      // set tag content
-      taosArrayClear(tagArray);
-      STagVal tagVal = {
-          .cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
-          .type = TSDB_DATA_TYPE_UBIGINT,
-          .i64 = (int64_t)pDataBlock->info.id.groupId,
-      };
-      taosArrayPush(tagArray, &tagVal);
-      pCreateTbReq->ctb.tagNum = taosArrayGetSize(tagArray);
-
-      STag* pTag = NULL;
-      tTagNew(tagArray, 1, false, &pTag);
-      if (pTag == NULL) {
-        terrno = TSDB_CODE_OUT_OF_MEMORY;
-        goto _end;
-      }
-      pCreateTbReq->ctb.pTag = (uint8_t*)pTag;
-
-      // set tag name
-      SArray* tagName = taosArrayInit(1, TSDB_COL_NAME_LEN);
-      char    tagNameStr[TSDB_COL_NAME_LEN] = {0};
-      strcpy(tagNameStr, "group_id");
-      taosArrayPush(tagName, tagNameStr);
-      pCreateTbReq->ctb.tagName = tagName;
-
-      // set table name
-      if (pDataBlock->info.parTbName[0]) {
-        pCreateTbReq->name = strdup(pDataBlock->info.parTbName);
-      } else {
-        pCreateTbReq->name = buildCtbNameByGroupId(stbFullName, pDataBlock->info.id.groupId);
-      }
-    }
-  }
-
-  // SSubmitTbData req
-  for (int32_t i = 0; i < sz; ++i) {
-    SSDataBlock* pDataBlock = taosArrayGet(pBlocks, i);
-    if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
-      pDeleteReq->suid = suid;
-      pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
-      tqBuildDeleteReq(pVnode, stbFullName, pDataBlock, pDeleteReq);
-      continue;
-    }
-
-    int32_t rows = pDataBlock->info.rows;
-
-    SSubmitTbData* pTbData = (SSubmitTbData*)taosMemoryCalloc(1, sizeof(SSubmitTbData));
-    if (!pTbData) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      goto _end;
-    }
-
-    if (!(pTbData->aRowP = taosArrayInit(rows, sizeof(SRow*)))) {
-      taosMemoryFree(pTbData);
-      goto _end;
-    }
-    pTbData->suid = suid;
-    pTbData->uid = 0;  // uid is assigned by vnode
-    pTbData->sver = pTSchema->version;
-
-    if (createTb) {
-      pTbData->pCreateTbReq = taosArrayGetP(createTbArray, i);
-      if (pTbData->pCreateTbReq) pTbData->flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
-    }
-
-    if (!pVals && !(pVals = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal)))) {
-      taosArrayDestroy(pTbData->aRowP);
-      taosMemoryFree(pTbData);
-      goto _end;
-    }
-
-    for (int32_t j = 0; j < rows; j++) {
-      taosArrayClear(pVals);
-      for (int32_t k = 0; k < pTSchema->numOfCols; k++) {
-        const STColumn*  pCol = &pTSchema->columns[k];
-        SColumnInfoData* pColData = taosArrayGet(pDataBlock->pDataBlock, k);
-        if (colDataIsNull_s(pColData, j)) {
-          SColVal cv = COL_VAL_NULL(pCol->colId, pCol->type);
-          taosArrayPush(pVals, &cv);
-        } else {
-          void* data = colDataGetData(pColData, j);
-          if (IS_STR_DATA_TYPE(pCol->type)) {
-            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);
-          } else {
-            SValue sv;
-            memcpy(&sv.val, data, tDataTypes[pCol->type].bytes);
-            SColVal cv = COL_VAL_VALUE(pCol->colId, pCol->type, sv);
-            taosArrayPush(pVals, &cv);
-          }
-        }
-      }
-      SRow* pRow = NULL;
-      if ((terrno = tRowBuild(pVals, (STSchema*)pTSchema, &pRow)) < 0) {
-        tDestroySSubmitTbData(pTbData, TSDB_MSG_FLG_ENCODE);
-        goto _end;
-      }
-      ASSERT(pRow);
-      taosArrayPush(pTbData->aRowP, &pRow);
-    }
-
-    taosArrayPush(pReq->aSubmitTbData, pTbData);
-  }
-
-  // encode
-  tEncodeSize(tEncodeSSubmitReq2, pReq, len, terrno);
-  if (TSDB_CODE_SUCCESS == terrno) {
-    SEncoder encoder;
-    len += sizeof(SMsgHead);
-    pBuf = rpcMallocCont(len);
-    if (NULL == pBuf) {
-      goto _end;
-    }
-    ((SMsgHead*)pBuf)->vgId = htonl(TD_VID(pVnode));
-    ((SMsgHead*)pBuf)->contLen = htonl(len);
-    tEncoderInit(&encoder, POINTER_SHIFT(pBuf, sizeof(SMsgHead)), len - sizeof(SMsgHead));
-    if (tEncodeSSubmitReq2(&encoder, pReq) < 0) {
-      terrno = TSDB_CODE_OUT_OF_MEMORY;
-      tqError("failed to encode submit req since %s", terrstr());
-    }
-    tEncoderClear(&encoder);
-  }
-_end:
-  taosArrayDestroy(tagArray);
-  taosArrayDestroy(pVals);
-  tDestroySSubmitReq2(pReq, TSDB_MSG_FLG_ENCODE);
-
-  if (terrno != 0) {
-    rpcFreeCont(pBuf);
-    taosArrayDestroy(pDeleteReq->deleteReqs);
-    return TSDB_CODE_FAILED;
-  }
-  if (ppData) *ppData = pBuf;
-  if (pLen) *pLen = len;
-  return TSDB_CODE_SUCCESS;
-}
 
 void tqSinkToTablePipeline(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
   const SArray*   pBlocks = (const SArray*)data;
@@ -984,56 +563,3 @@ _end:
   taosArrayDestroy(pVals);
   // TODO: change
 }
-
-#if 0
-void tqSinkToTableMerge(SStreamTask* pTask, void* vnode, int64_t ver, void* data) {
-  const SArray*   pRes = (const SArray*)data;
-  SVnode*         pVnode = (SVnode*)vnode;
-  SBatchDeleteReq deleteReq = {0};
-
-  tqDebug("vgId:%d, task %d write into table, block num: %d", TD_VID(pVnode), pTask->taskId, (int32_t)pRes->size);
-
-  deleteReq.deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq));
-  SSubmitReq* submitReq = tqBlockToSubmit(pVnode, pRes, pTask->tbSink.pTSchema, pTask->tbSink.pSchemaWrapper, true,
-                                          pTask->tbSink.stbUid, pTask->tbSink.stbFullName, &deleteReq);
-
-  tqDebug("vgId:%d, task %d convert blocks over, put into write-queue", TD_VID(pVnode), pTask->taskId);
-
-  if (taosArrayGetSize(deleteReq.deleteReqs) != 0) {
-    int32_t code;
-    int32_t len;
-    tEncodeSize(tEncodeSBatchDeleteReq, &deleteReq, len, code);
-    SEncoder encoder;
-    void*    serializedDeleteReq = rpcMallocCont(len + sizeof(SMsgHead));
-    void*    abuf = POINTER_SHIFT(serializedDeleteReq, sizeof(SMsgHead));
-    tEncoderInit(&encoder, abuf, len);
-    tEncodeSBatchDeleteReq(&encoder, &deleteReq);
-    tEncoderClear(&encoder);
-
-    ((SMsgHead*)serializedDeleteReq)->vgId = pVnode->config.vgId;
-
-    SRpcMsg msg = {
-        .msgType = TDMT_VND_BATCH_DEL,
-        .pCont = serializedDeleteReq,
-        .contLen = len + sizeof(SMsgHead),
-    };
-    if (tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) != 0) {
-      rpcFreeCont(serializedDeleteReq);
-      tqDebug("failed to put into write-queue since %s", terrstr());
-    }
-  }
-  taosArrayDestroy(deleteReq.deleteReqs);
-
-  /*tPrintFixedSchemaSubmitReq(pReq, pTask->tbSink.pTSchema);*/
-  // build write msg
-  SRpcMsg msg = {
-      .msgType = TDMT_VND_SUBMIT,
-      .pCont = submitReq,
-      .contLen = ntohl(submitReq->length),
-  };
-
-  if (tmsgPutToQueue(&pVnode->msgCb, WRITE_QUEUE, &msg) != 0) {
-    tqDebug("failed to put into write-queue since %s", terrstr());
-  }
-}
-#endif

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

@@ -4271,6 +4271,7 @@ int32_t tsdbGetFileBlocksDistInfo(STsdbReader* pReader, STableBlockDistInfo* pTa
   int32_t code = TSDB_CODE_SUCCESS;
   pTableBlockInfo->totalSize = 0;
   pTableBlockInfo->totalRows = 0;
+  pTableBlockInfo->numOfVgroups = 1;
 
   // find the start data block in file
   SReaderStatus* pStatus = &pReader->status;

source/libs/command/src/command.c

@@ -21,6 +21,7 @@
 #include "tdatablock.h"
 #include "tglobal.h"
 #include "tgrant.h"
+#include "taosdef.h"
 
 extern SConfig* tsCfg;
 
@@ -153,6 +154,23 @@ static int32_t buildCreateDBResultDataBlock(SSDataBlock** pOutput) {
   return code;
 }
 
+static int32_t buildAliveResultDataBlock(SSDataBlock** pOutput) {
+  SSDataBlock* pBlock = createDataBlock();
+  if (NULL == pBlock) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_INT, sizeof(int32_t), 1);
+  int32_t         code = blockDataAppendColInfo(pBlock, &infoData);
+
+  if (TSDB_CODE_SUCCESS == code) {
+    *pOutput = pBlock;
+  } else {
+    blockDataDestroy(pBlock);
+  }
+  return code;
+}
+
 int64_t getValOfDiffPrecision(int8_t unit, int64_t val) {
   int64_t v = 0;
   switch (unit) {
@@ -280,6 +298,108 @@ static void setCreateDBResultIntoDataBlock(SSDataBlock* pBlock, char* dbFName, S
   colDataAppend(pCol2, 0, buf2, false);
 }
 
+#define CHECK_LEADER(n) (row[n] && (fields[n].type == TSDB_DATA_TYPE_VARCHAR && strncasecmp(row[n], "leader", varDataLen((char *)row[n] - VARSTR_HEADER_SIZE)) == 0))
+// on this row, if have leader return true else return false
+bool existLeaderRole(TAOS_ROW row, TAOS_FIELD* fields, int nFields) {
+  // vgroup_id | db_name | tables | v1_dnode | v1_status | v2_dnode | v2_status | v3_dnode | v3_status | v4_dnode |
+  // v4_status |  cacheload  | tsma |
+  if (nFields != 13) {
+    return false;
+  }
+
+  // check have leader on cloumn v*_status on 4 6 8 10
+  if (CHECK_LEADER(4) || CHECK_LEADER(6) || CHECK_LEADER(8) || CHECK_LEADER(10)) {
+    return true;
+  }
+
+  return false;
+}
+
+// get db alive status, return 1 is alive else return 0
+int32_t getAliveStatusFromApi(int64_t* pConnId, char* dbName, int32_t* pStatus) {
+  char    sql[128 + TSDB_DB_NAME_LEN] = "select * from information_schema.ins_vgroups";
+  int32_t code;
+
+  // filter with db name
+  if (dbName && dbName[0] != 0) {
+    char str[64 + TSDB_DB_NAME_LEN] = "";
+    // test db name exist
+    sprintf(str, "show create database %s ;", dbName);
+    TAOS_RES* dbRes = taos_query(pConnId, str);
+    code = taos_errno(dbRes);
+    if (code != TSDB_CODE_SUCCESS) {
+      taos_free_result(dbRes);
+      return code;
+    }
+    taos_free_result(dbRes);
+
+    sprintf(str, " where db_name='%s' ;", dbName);
+    strcat(sql, str);
+  }
+
+  TAOS_RES* res = taos_query(pConnId, sql);
+  code = taos_errno(res);
+  if (code != TSDB_CODE_SUCCESS) {
+    taos_free_result(res);
+    return code;
+  }
+
+  TAOS_ROW    row = NULL;
+  TAOS_FIELD* fields = taos_fetch_fields(res);
+  int32_t     nFields = taos_num_fields(res);
+  int32_t     nAvailble = 0;
+  int32_t     nUnAvailble = 0;
+
+  while ((row = taos_fetch_row(res)) != NULL) {
+    if (existLeaderRole(row, fields, nFields)) {
+      nAvailble++;
+    } else {
+      nUnAvailble++;
+    }
+  }
+  taos_free_result(res);
+
+  int32_t status = 0;
+  if (nAvailble + nUnAvailble == 0 || nUnAvailble == 0) {
+    status = SHOW_STATUS_AVAILABLE;
+  } else if (nAvailble > 0 && nUnAvailble > 0) {
+    status = SHOW_STATUS_HALF_AVAILABLE;
+  } else {
+    status = SHOW_STATUS_NOT_AVAILABLE;
+  }
+
+  if (pStatus) {
+    *pStatus = status;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t setAliveResultIntoDataBlock(int64_t* pConnId, SSDataBlock* pBlock, char* dbName) {
+  blockDataEnsureCapacity(pBlock, 1);
+  pBlock->info.rows = 1;
+
+  SColumnInfoData* pCol1 = taosArrayGet(pBlock->pDataBlock, 0);
+  int32_t          status = 0;
+  int32_t          code = getAliveStatusFromApi(pConnId, dbName, &status);
+  if (code == TSDB_CODE_SUCCESS) {
+    colDataAppend(pCol1, 0, (const char*)&status, false);
+  }
+  return code;
+}
+
+static int32_t execShowAliveStatus(int64_t* pConnId, SShowAliveStmt* pStmt, SRetrieveTableRsp** pRsp) {
+  SSDataBlock* pBlock = NULL;
+  int32_t      code = buildAliveResultDataBlock(&pBlock);
+  if (TSDB_CODE_SUCCESS == code) {
+    code = setAliveResultIntoDataBlock(pConnId, pBlock, pStmt->dbName);
+  }
+  if (TSDB_CODE_SUCCESS == code) {
+    code = buildRetrieveTableRsp(pBlock, SHOW_ALIVE_RESULT_COLS, pRsp);
+  }
+  blockDataDestroy(pBlock);
+  return code;
+}
+
 static int32_t execShowCreateDatabase(SShowCreateDatabaseStmt* pStmt, SRetrieveTableRsp** pRsp) {
   SSDataBlock* pBlock = NULL;
   int32_t      code = buildCreateDBResultDataBlock(&pBlock);
@@ -735,7 +855,7 @@ static int32_t execSelectWithoutFrom(SSelectStmt* pSelect, SRetrieveTableRsp** p
   return code;
 }
 
-int32_t qExecCommand(bool sysInfoUser, SNode* pStmt, SRetrieveTableRsp** pRsp) {
+int32_t qExecCommand(int64_t* pConnId, bool sysInfoUser, SNode* pStmt, SRetrieveTableRsp** pRsp) {
   switch (nodeType(pStmt)) {
     case QUERY_NODE_DESCRIBE_STMT:
       return execDescribe(sysInfoUser, pStmt, pRsp);
@@ -753,6 +873,9 @@ int32_t qExecCommand(bool sysInfoUser, SNode* pStmt, SRetrieveTableRsp** pRsp) {
       return execShowLocalVariables(pRsp);
     case QUERY_NODE_SELECT_STMT:
       return execSelectWithoutFrom((SSelectStmt*)pStmt, pRsp);
+    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
+    case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:
+      return execShowAliveStatus(pConnId, (SShowAliveStmt*)pStmt, pRsp);       
     default:
       break;
   }

source/libs/function/src/builtinsimpl.c

@@ -5192,6 +5192,7 @@ int32_t blockDistFunction(SqlFunctionCtx* pCtx) {
   pDistInfo->numOfBlocks += p1.numOfBlocks;
   pDistInfo->numOfTables += p1.numOfTables;
   pDistInfo->numOfInmemRows += p1.numOfInmemRows;
+  pDistInfo->numOfVgroups += p1.numOfVgroups;
   pDistInfo->totalSize += p1.totalSize;
   pDistInfo->totalRows += p1.totalRows;
   pDistInfo->numOfFiles += p1.numOfFiles;
@@ -5226,6 +5227,7 @@ int32_t tSerializeBlockDistInfo(void* buf, int32_t bufLen, const STableBlockDist
   if (tEncodeU16(&encoder, pInfo->numOfFiles) < 0) return -1;
   if (tEncodeU32(&encoder, pInfo->numOfBlocks) < 0) return -1;
   if (tEncodeU32(&encoder, pInfo->numOfTables) < 0) return -1;
+  if (tEncodeU32(&encoder, pInfo->numOfVgroups) < 0) return -1;
 
   if (tEncodeU64(&encoder, pInfo->totalSize) < 0) return -1;
   if (tEncodeU64(&encoder, pInfo->totalRows) < 0) return -1;
@@ -5257,6 +5259,7 @@ int32_t tDeserializeBlockDistInfo(void* buf, int32_t bufLen, STableBlockDistInfo
   if (tDecodeU16(&decoder, &pInfo->numOfFiles) < 0) return -1;
   if (tDecodeU32(&decoder, &pInfo->numOfBlocks) < 0) return -1;
   if (tDecodeU32(&decoder, &pInfo->numOfTables) < 0) return -1;
+  if (tDecodeU32(&decoder, &pInfo->numOfVgroups) < 0) return -1;
 
   if (tDecodeU64(&decoder, &pInfo->totalSize) < 0) return -1;
   if (tDecodeU64(&decoder, &pInfo->totalRows) < 0) return -1;
@@ -5317,7 +5320,7 @@ int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
   colDataAppend(pColInfo, row++, st, false);
 
   len = sprintf(st + VARSTR_HEADER_SIZE, "Total_Tables=[%d] Total_Files=[%d] Total_Vgroups=[%d]", pData->numOfTables,
-                pData->numOfFiles, 0);
+                pData->numOfFiles, pData->numOfVgroups);
 
   varDataSetLen(st, len);
   colDataAppend(pColInfo, row++, st, false);

source/libs/nodes/src/nodesCodeFuncs.c

@@ -173,6 +173,10 @@ const char* nodesNodeName(ENodeType type) {
       return "BalanceVgroupStmt";
     case QUERY_NODE_MERGE_VGROUP_STMT:
       return "MergeVgroupStmt";
+    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
+      return "ShowDbAliveStmt";
+    case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:
+      return "ShowClusterAliveStmt";      
     case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
       return "RedistributeVgroupStmt";
     case QUERY_NODE_SPLIT_VGROUP_STMT:

source/libs/nodes/src/nodesUtilFuncs.c

@@ -434,6 +434,9 @@ SNode* nodesMakeNode(ENodeType type) {
       return makeNode(type, sizeof(SShowDnodeVariablesStmt));
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
       return makeNode(type, sizeof(SShowCreateDatabaseStmt));
+    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
+    case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:
+      return makeNode(type, sizeof(SShowAliveStmt));      
     case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
     case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
       return makeNode(type, sizeof(SShowCreateTableStmt));
@@ -960,6 +963,8 @@ void nodesDestroyNode(SNode* pNode) {
     case QUERY_NODE_SHOW_USERS_STMT:
     case QUERY_NODE_SHOW_LICENCES_STMT:
     case QUERY_NODE_SHOW_VGROUPS_STMT:
+    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
+    case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:    
     case QUERY_NODE_SHOW_TOPICS_STMT:
     case QUERY_NODE_SHOW_CONSUMERS_STMT:
     case QUERY_NODE_SHOW_CONNECTIONS_STMT:

source/libs/parser/inc/parAst.h

@@ -176,6 +176,7 @@ SNode* createShowStmt(SAstCreateContext* pCxt, ENodeType type);
 SNode* createShowStmtWithCond(SAstCreateContext* pCxt, ENodeType type, SNode* pDbName, SNode* pTbName,
                               EOperatorType tableCondType);
 SNode* createShowCreateDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName);
+SNode* createShowAliveStmt(SAstCreateContext* pCxt, SNode* pDbName, ENodeType type);
 SNode* createShowCreateTableStmt(SAstCreateContext* pCxt, ENodeType type, SNode* pRealTable);
 SNode* createShowTableDistributedStmt(SAstCreateContext* pCxt, SNode* pRealTable);
 SNode* createShowDnodeVariablesStmt(SAstCreateContext* pCxt, SNode* pDnodeId, SNode* pLikePattern);

source/libs/parser/inc/sql.y

@@ -433,6 +433,9 @@ cmd ::= SHOW TAGS FROM table_name_cond(A) from_db_opt(B).
 cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM table_name_cond(A) from_db_opt(B).   { pCxt->pRootNode = createShowTableTagsStmt(pCxt, A, B, C); }
 cmd ::= SHOW VNODES NK_INTEGER(A).                                                { pCxt->pRootNode = createShowVnodesStmt(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &A), NULL); }
 cmd ::= SHOW VNODES NK_STRING(A).                                                 { pCxt->pRootNode = createShowVnodesStmt(pCxt, NULL, createValueNode(pCxt, TSDB_DATA_TYPE_VARCHAR, &A)); }
+// show alive
+cmd ::= SHOW db_name_cond_opt(A) ALIVE.                                           { pCxt->pRootNode = createShowAliveStmt(pCxt, A,    QUERY_NODE_SHOW_DB_ALIVE_STMT); }
+cmd ::= SHOW CLUSTER ALIVE.                                                       { pCxt->pRootNode = createShowAliveStmt(pCxt, NULL, QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT); }
 
 db_name_cond_opt(A) ::= .                                                         { A = createDefaultDatabaseCondValue(pCxt); }
 db_name_cond_opt(A) ::= db_name(B) NK_DOT.                                        { A = createIdentifierValueNode(pCxt, &B); }

source/libs/parser/src/parAstCreater.c

@@ -1370,6 +1370,38 @@ SNode* createShowCreateDatabaseStmt(SAstCreateContext* pCxt, SToken* pDbName) {
   return (SNode*)pStmt;
 }
 
+SNode* createShowAliveStmt(SAstCreateContext* pCxt, SNode* pNode, ENodeType type) {
+  CHECK_PARSER_STATUS(pCxt);
+  SToken  dbToken = {0};
+  SToken* pDbToken = NULL;
+
+  if (pNode) {
+    SValueNode* pDbName = (SValueNode*)pNode;
+    if (pDbName->literal) {
+      dbToken.z = pDbName->literal;
+      dbToken.n = strlen(pDbName->literal);
+      pDbToken = &dbToken;
+    }
+  }
+
+  if (pDbToken && !checkDbName(pCxt, pDbToken, true)) {
+    nodesDestroyNode(pNode);
+    return NULL;
+  }
+
+  SShowAliveStmt* pStmt = (SShowAliveStmt*)nodesMakeNode(type);
+  CHECK_OUT_OF_MEM(pStmt);
+
+  if (pDbToken) {
+    COPY_STRING_FORM_ID_TOKEN(pStmt->dbName, pDbToken);
+  }
+  if (pNode) {
+    nodesDestroyNode(pNode);
+  }
+
+  return (SNode*)pStmt;
+}
+
 SNode* createShowCreateTableStmt(SAstCreateContext* pCxt, ENodeType type, SNode* pRealTable) {
   CHECK_PARSER_STATUS(pCxt);
   SShowCreateTableStmt* pStmt = (SShowCreateTableStmt*)nodesMakeNode(type);

source/libs/parser/src/parAuthenticator.c

@@ -145,6 +145,8 @@ static int32_t authQuery(SAuthCxt* pCxt, SNode* pStmt) {
     case QUERY_NODE_SHOW_CLUSTER_STMT:
     case QUERY_NODE_SHOW_LICENCES_STMT:
     case QUERY_NODE_SHOW_VGROUPS_STMT:
+    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
+    case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:    
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
     case QUERY_NODE_SHOW_TABLE_DISTRIBUTED_STMT:
     case QUERY_NODE_SHOW_VNODES_STMT:

source/libs/parser/src/parInsertSml.c

@@ -189,14 +189,23 @@ int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32
   SSchema* pColSchema = schema + index;
   SColVal* pVal = taosArrayGet(pTableCxt->pValues, index);
   SSmlKv*  kv = (SSmlKv*)data;
+  if(kv->keyLen != strlen(pColSchema->name) || memcmp(kv->key, pColSchema->name, kv->keyLen) != 0){
+    ret = TSDB_CODE_SML_INVALID_DATA;
+    goto end;
+  }
   if (kv->type == TSDB_DATA_TYPE_NCHAR) {
     int32_t len = 0;
-    char*   pUcs4 = taosMemoryCalloc(1, pColSchema->bytes - VARSTR_HEADER_SIZE);
+    int64_t size = pColSchema->bytes - VARSTR_HEADER_SIZE;
+    if(size <= 0){
+      ret = TSDB_CODE_SML_INVALID_DATA;
+      goto end;
+    }
+    char*   pUcs4 = taosMemoryCalloc(1, size);
     if (NULL == pUcs4) {
       ret = TSDB_CODE_OUT_OF_MEMORY;
       goto end;
     }
-    if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)pUcs4, pColSchema->bytes - VARSTR_HEADER_SIZE, &len)) {
+    if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)pUcs4, size, &len)) {
       if (errno == E2BIG) {
         ret = TSDB_CODE_PAR_VALUE_TOO_LONG;
         goto end;

source/libs/parser/src/parTokenizer.c

@@ -269,6 +269,7 @@ static SKeyword keywordTable[] = {
     {"_WDURATION",           TK_WDURATION},
     {"_WEND",                TK_WEND},
     {"_WSTART",              TK_WSTART},
+    {"ALIVE",                TK_ALIVE},
 };
 // clang-format on
 

source/libs/parser/src/parTranslater.c

@@ -6364,6 +6364,20 @@ static int32_t extractShowCreateDatabaseResultSchema(int32_t* numOfCols, SSchema
   return TSDB_CODE_SUCCESS;
 }
 
+static int32_t extractShowAliveResultSchema(int32_t* numOfCols, SSchema** pSchema) {
+  *numOfCols = 1;
+  *pSchema = taosMemoryCalloc((*numOfCols), sizeof(SSchema));
+  if (NULL == (*pSchema)) {
+    return TSDB_CODE_OUT_OF_MEMORY;
+  }
+
+  (*pSchema)[0].type = TSDB_DATA_TYPE_INT;
+  (*pSchema)[0].bytes = sizeof(int32_t);
+  strcpy((*pSchema)[0].name, "status");
+
+  return TSDB_CODE_SUCCESS;
+}
+
 static int32_t extractShowCreateTableResultSchema(int32_t* numOfCols, SSchema** pSchema) {
   *numOfCols = 2;
   *pSchema = taosMemoryCalloc((*numOfCols), sizeof(SSchema));
@@ -6415,6 +6429,9 @@ int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pS
       return extractDescribeResultSchema(numOfCols, pSchema);
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
       return extractShowCreateDatabaseResultSchema(numOfCols, pSchema);
+    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
+    case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:
+      return extractShowAliveResultSchema(numOfCols, pSchema);      
     case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
     case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
       return extractShowCreateTableResultSchema(numOfCols, pSchema);
@@ -7934,6 +7951,8 @@ static int32_t setQuery(STranslateContext* pCxt, SQuery* pQuery) {
       break;
     case QUERY_NODE_DESCRIBE_STMT:
     case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
+    case QUERY_NODE_SHOW_DB_ALIVE_STMT:
+    case QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT:    
     case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
     case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
     case QUERY_NODE_SHOW_LOCAL_VARIABLES_STMT:

source/os/CMakeLists.txt

@@ -59,4 +59,8 @@ endif()
 
 IF (JEMALLOC_ENABLED)
     target_link_libraries(os PUBLIC -ljemalloc)
-ENDIF ()
\ No newline at end of file
+ENDIF ()
+
+if(${BUILD_TEST})
+    add_subdirectory(test)
+endif(${BUILD_TEST})
\ No newline at end of file

source/os/src/osFile.c

@@ -560,6 +560,21 @@ int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime) {
 
 int32_t taosLockFile(TdFilePtr pFile) {
 #ifdef WINDOWS
+  BOOL          fSuccess = FALSE;
+  LARGE_INTEGER fileSize;
+  OVERLAPPED    overlapped = {0};
+
+  HANDLE hFile = (HANDLE)_get_osfhandle(pFile->fd);
+
+  fSuccess = LockFileEx(hFile, LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY,
+                        0,           // reserved
+                        ~0,          // number of bytes to lock low
+                        ~0,          // number of bytes to lock high
+                        &overlapped  // overlapped structure
+  );
+  if (!fSuccess) {
+    return GetLastError();
+  }
   return 0;
 #else
   assert(pFile->fd >= 0);  // Please check if you have closed the file.
@@ -570,6 +585,14 @@ int32_t taosLockFile(TdFilePtr pFile) {
 
 int32_t taosUnLockFile(TdFilePtr pFile) {
 #ifdef WINDOWS
+  BOOL          fSuccess = FALSE;
+  OVERLAPPED    overlapped = {0};
+  HANDLE        hFile = (HANDLE)_get_osfhandle(pFile->fd);
+
+  fSuccess = UnlockFileEx(hFile, 0, ~0, ~0, &overlapped);
+  if (!fSuccess) {
+    return GetLastError();
+  }
   return 0;
 #else
   assert(pFile->fd >= 0);  // Please check if you have closed the file.

source/os/test/osTests.cpp

@@ -27,6 +27,7 @@
 #pragma GCC diagnostic ignored "-Wpointer-arith"
 
 #include "os.h"
+#include "tlog.h"
 
 TEST(osTest, osSystem) {
   const char *flags = "UTL FATAL ";
@@ -35,4 +36,97 @@ TEST(osTest, osSystem) {
   taosPrintTrace(flags, level, dflag);
 }
 
+void fileOperateOnFree(void *param) {
+  char *    fname = (char *)param;
+  TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
+  printf("On free thread open file\n");
+  ASSERT_NE(pFile, nullptr);
+
+  int ret = taosLockFile(pFile);
+  printf("On free thread lock file ret:%d\n", ret);
+  ASSERT_EQ(ret, 0);
+
+  ret = taosUnLockFile(pFile);
+  printf("On free thread unlock file ret:%d\n", ret);
+  ASSERT_EQ(ret, 0);
+
+  ret = taosCloseFile(&pFile);
+  ASSERT_EQ(ret, 0);
+  printf("On free thread close file ret:%d\n", ret);
+}
+void *fileOperateOnFreeThread(void *param) {
+  fileOperateOnFree(param);
+  return NULL;
+}
+void fileOperateOnBusy(void *param) {
+  char *    fname = (char *)param;
+  TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
+  printf("On busy thread open file\n");
+  ASSERT_NE(pFile, nullptr);
+
+  int ret = taosLockFile(pFile);
+  printf("On busy thread lock file ret:%d\n", ret);
+  ASSERT_NE(ret, 0);
+
+  ret = taosUnLockFile(pFile);
+  printf("On busy thread unlock file ret:%d\n", ret);
+#ifdef _TD_DARWIN_64
+  ASSERT_EQ(ret, 0);
+#else
+  ASSERT_NE(ret, 0);
+#endif
+
+  ret = taosCloseFile(&pFile);
+  printf("On busy thread close file ret:%d\n", ret);
+  ASSERT_EQ(ret, 0);
+}
+void *fileOperateOnBusyThread(void *param) {
+  fileOperateOnBusy(param);
+  return NULL;
+}
+
+TEST(osTest, osFile) {
+  char *fname = "./osfiletest1.txt";
+
+  TdFilePtr pOutFD = taosCreateFile(fname, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
+  ASSERT_NE(pOutFD, nullptr);
+  printf("create file success\n");
+
+  TdFilePtr pFile = taosOpenFile(fname, TD_FILE_CREATE | TD_FILE_WRITE);
+  printf("open file\n");
+  ASSERT_NE(pFile, nullptr);
+
+  int ret = taosLockFile(pFile);
+  printf("lock file ret:%d\n", ret);
+  ASSERT_EQ(ret, 0);
+
+  TdThreadAttr thattr;
+  taosThreadAttrInit(&thattr);
+
+  TdThread thread1, thread2;
+  taosThreadCreate(&(thread1), &thattr, fileOperateOnBusyThread, (void *)fname);
+  taosThreadAttrDestroy(&thattr);
+
+  taosThreadJoin(thread1, NULL);
+  taosThreadClear(&thread1);
+
+  ret = taosUnLockFile(pFile);
+  printf("unlock file ret:%d\n", ret);
+  ASSERT_EQ(ret, 0);
+
+  ret = taosCloseFile(&pFile);
+  printf("close file ret:%d\n", ret);
+  ASSERT_EQ(ret, 0);
+
+  taosThreadCreate(&(thread2), &thattr, fileOperateOnFreeThread, (void *)fname);
+  taosThreadAttrDestroy(&thattr);
+
+  taosThreadJoin(thread2, NULL);
+  taosThreadClear(&thread2);
+
+  //int ret = taosRemoveFile(fname);
+  //ASSERT_EQ(ret, 0);
+  //printf("remove file success");
+}
+
 #pragma GCC diagnostic pop

tests/parallel_test/cases.task

@@ -423,8 +423,8 @@
 ,,n,system-test,python3 ./test.py -f 0-others/compatibility.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_database.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
-,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_telnet_line_taosc_insert.py
-,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
+,,n,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_telnet_line_taosc_insert.py
+,,n,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/test_stmt_set_tbname_tag.py
 ,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_stable.py
@@ -1037,7 +1037,7 @@
 ,,y,system-test,./pytest.sh python3 ./test.py -f 99-TDcase/TD-20582.py
 
 #develop test
-#,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/auto_create_table_json.py
+,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/auto_create_table_json.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/custom_col_tag.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/default_json.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/demo.py
@@ -1046,7 +1046,7 @@
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/json_tag.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/query_json.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/sample_csv_json.py
-#,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/sml_json_alltypes.py
+,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/sml_json_alltypes.py
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/taosdemoTestQueryWithJson.py -R
 ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/telnet_tcp.py -R
 

tests/pytest/util/sql.py

@@ -261,6 +261,70 @@ class TDSql:
 
         tdLog.info(f"sql:{self.sql}, row:{row} col:{col} data:{self.queryResult[row][col]} == expect:{data}")
 
+    # return true or false replace exit, no print out
+    def checkRowColNoExit(self, row, col):
+        caller = inspect.getframeinfo(inspect.stack()[2][0])
+        if row < 0:
+            args = (caller.filename, caller.lineno, self.sql, row)
+            return False
+        if col < 0:
+            args = (caller.filename, caller.lineno, self.sql, row)
+            return False
+        if row > self.queryRows:
+            args = (caller.filename, caller.lineno, self.sql, row, self.queryRows)
+            return False
+        if col > self.queryCols:
+            args = (caller.filename, caller.lineno, self.sql, col, self.queryCols)
+            return False
+            
+        return True
+
+
+    # return true or false replace exit, no print out
+    def checkDataNoExit(self, row, col, data):
+        if self.checkRowColNoExit(row, col) == False:
+            return False
+        if self.queryResult[row][col] != data:
+            if self.cursor.istype(col, "TIMESTAMP"):
+                # suppose user want to check nanosecond timestamp if a longer data passed
+                if (len(data) >= 28):
+                    if pd.to_datetime(self.queryResult[row][col]) == pd.to_datetime(data):
+                        return True
+                else:
+                    if self.queryResult[row][col] == _parse_datetime(data):
+                        return True
+                return False
+
+            if str(self.queryResult[row][col]) == str(data):
+                return True
+            elif isinstance(data, float):
+                if abs(data) >= 1 and abs((self.queryResult[row][col] - data) / data) <= 0.000001:
+                    return True
+                elif abs(data) < 1 and abs(self.queryResult[row][col] - data) <= 0.000001:
+                    return True
+                else:
+                    return False
+            else:
+                return False
+                
+        return True
+
+
+    # loop execute sql then sleep(waitTime) , if checkData ok break loop
+    def checkDataLoop(self, row, col, data, sql, loopCount, waitTime):
+        # loop check util checkData return true
+        for i in range(loopCount):
+            self.query(sql)
+            if self.checkDataNoExit(row, col, data) :
+                self.checkData(row, col, data)
+                return
+            time.sleep(waitTime)
+
+        # last check
+        self.query(sql)
+        self.checkData(row, col, data)
+
+
     def getData(self, row, col):
         self.checkRowCol(row, col)
         return self.queryResult[row][col]

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

@@ -1432,9 +1432,9 @@ class TDTestCase:
         self._conn.schemaless_insert([json.dumps(input_json)], TDSmlProtocolType.JSON.value, None)
         query_sql = 'select * from `rFa$sta`'
         query_res = tdSql.query(query_sql, True)
-        tdSql.checkEqual(query_res, [(datetime.datetime(2021, 7, 11, 20, 33, 54), True, 'rFas$ta_1', 'ncharTagValue', 2147483647, 9223372036854775807, 22.123456789, 'binaryTagValue', 32767, 11.12345027923584, False, 127)])
+        tdSql.checkEqual(query_res, [(datetime.datetime(2021, 7, 11, 20, 33, 54), True, False, 127, 32767, 2147483647, 9223372036854775807, 11.12345027923584, 22.123456789, 'binaryTagValue', 'ncharTagValue', 'rFas$ta_1')])
         col_tag_res = tdSql.getColNameList(query_sql)
-        tdSql.checkEqual(col_tag_res, ['_ts', '_value', 'id', 't!@#$%^&*()_+[];:<>?,9', 't$3', 't%4', 't&6', 't*7', 't@2', 't^5', 'Tt!0', 'tT@1'])
+        tdSql.checkEqual(col_tag_res, ['_ts', '_value', 'Tt!0', 'tT@1', 't@2', 't$3', 't%4', 't^5', 't&6', 't*7', 't!@#$%^&*()_+[];:<>?,9', 'id'])
         tdSql.execute('drop table `rFa$sta`')
 
     def pointTransCheckCase(self, value_type="obj"):
@@ -1719,7 +1719,6 @@ class TDTestCase:
             print(err.errno)
 
     def runAll(self):
-        """
         for value_type in ["obj", "default"]:
             self.initCheckCase(value_type)
             self.symbolsCheckCase(value_type)
@@ -1772,7 +1771,7 @@ class TDTestCase:
         # self.sStbStbDdataDtsMtInsertMultiThreadCheckCase()
         # self.sStbDtbDdataDtsMtInsertMultiThreadCheckCase()
         # self.lengthIcreaseCrashCheckCase()
-        """
+
     def run(self):
         print("running {}".format(__file__))
         self.createDb()

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

@@ -243,7 +243,7 @@ class TDTestCase:
             if t_add_tag is not None:
                 sql_seq = f'{stb_name} {ts} {value} t0={t0} t1={t1} t2={t2} t3={t3} t4={t4} t5={t5} t6={t6} t7={t7} t8={t8} t9={t8}'
         if id_change_tag is not None:
-            sql_seq = f'{stb_name} {ts} {value} t0={t0} {id}={tb_name} t1={t1} t2={t2} t3={t3} t4={t4} t5={t5} t6={t6} t7={t7} t8={t8}'
+            sql_seq = f'{stb_name} {ts} {value} {id}={tb_name} t0={t0} t1={t1} t2={t2} t3={t3} t4={t4} t5={t5} t6={t6} t7={t7} t8={t8}'
         if id_double_tag is not None:
             sql_seq = f'{stb_name} {ts} {value} {id}=\"{tb_name}_1\" t0={t0} t1={t1} {id}=\"{tb_name}_2\" t2={t2} t3={t3} t4={t4} t5={t5} t6={t6} t7={t7} t8={t8}'
         if t_add_tag is not None:
@@ -1126,9 +1126,9 @@ class TDTestCase:
             self._conn.schemaless_insert([input_sql], TDSmlProtocolType.TELNET.value, None)
             query_sql = 'select * from `rFa$sta`'
             query_res = tdSql.query(query_sql, True)
-            tdSql.checkEqual(query_res, [(datetime.datetime(2021, 7, 11, 20, 33, 54), 9.223372036854776e+18, '2147483647i32', 'L"ncharTagValue"', '32767i16', '9223372036854775807i64', '22.123456789f64', '"ddzhiksj"', '11.12345f32', 'true', '127Ii8')])
+            tdSql.checkEqual(query_res, [(datetime.datetime(2021, 7, 11, 20, 33, 54), 9.223372036854776e+18, 'true', '127Ii8', '32767i16', '2147483647i32', '9223372036854775807i64', '11.12345f32', '22.123456789f64', '"ddzhiksj"', 'L"ncharTagValue"')])
             col_tag_res = tdSql.getColNameList(query_sql)
-            tdSql.checkEqual(col_tag_res, ['_ts', '_value', '"t$3"', 't!@#$%^&*()_+[];:<>?,9', 't#2', 't%4', 't&6', 't*7', 't^5', 'Tt!0', 'tT@1'])
+            tdSql.checkEqual(col_tag_res, ['_ts', '_value', 'Tt!0', 'tT@1', 't#2', '"t$3"', 't%4', 't^5', 't&6', 't*7', 't!@#$%^&*()_+[];:<>?,9'])
             tdSql.execute('drop table `rFa$sta`')
 
     def tcpKeywordsCheckCase(self, protocol="telnet-tcp"):
@@ -1207,7 +1207,6 @@ class TDTestCase:
         tdLog.info(f'{sys._getframe().f_code.co_name}() function is running')
         tdCom.cleanTb(dbname="test")
         input_sql = self.genSqlList()[0]
-        print(input_sql)
         self.multiThreadRun(self.genMultiThreadSeq(input_sql))
         tdSql.query(f"show tables;")
         tdSql.checkRows(5)
@@ -1416,8 +1415,8 @@ class TDTestCase:
         self.symbolsCheckCase()
         self.tsCheckCase()
         self.openTstbTelnetTsCheckCase()
-        #self.idSeqCheckCase()
-        #self.idLetterCheckCase()
+        # self.idSeqCheckCase()
+        self.idLetterCheckCase()
         self.noIdCheckCase()
         self.maxColTagCheckCase()
         self.stbTbNameCheckCase()
@@ -1450,7 +1449,7 @@ class TDTestCase:
         self.spellCheckCase()
         self.pointTransCheckCase()
         self.defaultTypeCheckCase()
-        #self.tbnameTagsColsNameCheckCase()
+        self.tbnameTagsColsNameCheckCase()
         # # # MultiThreads
         # self.stbInsertMultiThreadCheckCase()
         # self.sStbStbDdataInsertMultiThreadCheckCase()

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

@@ -71,23 +71,26 @@ class TDTestCase:
         tdSql.checkData(0, 2, "web01")
 
         tdSql.query(f"select distinct tbname from {dbname}.`sys.cpu.nice`")
-        tdSql.checkRows(2)
+        tdSql.checkRows(3)
 
         tdSql.query(f"select * from {dbname}.`sys.cpu.nice` order by _ts")
-        tdSql.checkRows(2)
-        tdSql.checkData(0, 1, 9.000000000)
-        tdSql.checkData(0, 2, "web02")
+        tdSql.checkRows(4)
+        tdSql.checkData(0, 1, 13.000000000)
+        tdSql.checkData(0, 2, "web01")
         tdSql.checkData(0, 3, None)
         tdSql.checkData(0, 4, "lga")
 
-        tdSql.checkData(1, 1, 18.000000000)
-        tdSql.checkData(1, 2, "web01")
-        tdSql.checkData(1, 3, "t1")
+        tdSql.checkData(1, 1, 9.000000000)
+        tdSql.checkData(1, 2, "web02")
+        tdSql.checkData(3, 3, "t1")
         tdSql.checkData(0, 4, "lga")
 
         tdSql.query(f"select * from {dbname}.macylr")
         tdSql.checkRows(2)
 
+        tdSql.query(f"select * from {dbname}.qelhxo")
+        tdSql.checkRows(5)
+
         tdSql.query(f"desc {dbname}.macylr")
         tdSql.checkRows(25)
         return

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

@@ -137,11 +137,34 @@ class TDTestCase:
         config_dir = dnode.cfgDir
         return taos.connect(host=host, port=int(port), config=config_dir)
 
+    def check_alive(self):
+        # check cluster alive
+        tdLog.printNoPrefix("======== test cluster alive: ")
+        tdSql.checkDataLoop(0, 0, 1, "show cluster alive;", 20, 0.5)
+
+        tdSql.query("show db.alive;")
+        tdSql.checkData(0, 0, 1)
+
+        # stop 5 dnode 
+        self.TDDnodes.stoptaosd(5)
+        tdSql.checkDataLoop(0, 0, 2, "show cluster alive;", 20, 0.5)
+        
+        tdSql.query("show db.alive;")
+        tdSql.checkData(0, 0, 2)
+
+        # stop 2 dnode
+        self.TDDnodes.stoptaosd(2)
+        tdSql.checkDataLoop(0, 0, 0, "show cluster alive;", 20, 0.5)
+
+        tdSql.query("show db.alive;")
+        tdSql.checkData(0, 0, 0)
+        
 
     def run(self):
         # print(self.master_dnode.cfgDict)
         self.five_dnode_one_mnode()
-
+        # check cluster and db alive
+        self.check_alive()
 
     def stop(self):
         tdSql.close()

tools/shell/src/shellAuto.c

@@ -138,6 +138,7 @@ SWords shellCommands[] = {
     {"show create table <tb_name> \\G;", 0, 0, NULL},
     {"show connections;", 0, 0, NULL},
     {"show cluster;", 0, 0, NULL},
+    {"show cluster alive;", 0, 0, NULL},
     {"show databases;", 0, 0, NULL},
     {"show dnodes;", 0, 0, NULL},
     {"show dnode <dnode_id> variables;", 0, 0, NULL},
@@ -425,6 +426,7 @@ void showHelp() {
     show create table <tb_name>;\n\
     show connections;\n\
     show cluster;\n\
+    show cluster alive;\n\
     show databases;\n\
     show dnodes;\n\
     show dnode <dnode_id> variables;\n\

utils/test/c/sml_test.c

@@ -78,21 +78,23 @@ int smlProcess_telnet_Test() {
   pRes = taos_query(taos, "use sml_db");
   taos_free_result(pRes);
 
-  char *sql[4] = {0};
-  sql[0] = taosMemoryCalloc(1, 128);
-  sql[1] = taosMemoryCalloc(1, 128);
-  sql[2] = taosMemoryCalloc(1, 128);
-  sql[3] = taosMemoryCalloc(1, 128);
+//  char *sql[4] = {0};
+//  sql[0] = taosMemoryCalloc(1, 128);
+//  sql[1] = taosMemoryCalloc(1, 128);
+//  sql[2] = taosMemoryCalloc(1, 128);
+//  sql[3] = taosMemoryCalloc(1, 128);
   const char *sql1[] = {"sys.if.bytes.out  1479496100 1.3E0 host=web01 interface=eth0",
                        "sys.if.bytes.out  1479496101 1.3E1 interface=eth0    host=web01   ",
                        "sys.if.bytes.out  1479496102 1.3E3 network=tcp",
                        " sys.procs.running   1479496100 42 host=web01   "};
 
-  for(int i = 0; i < 4; i++){
-    strncpy(sql[i], sql1[i], 128);
-  }
+//  for(int i = 0; i < 4; i++){
+//    strncpy(sql[i], sql1[i], 128);
+//  }
 
-  pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_TELNET_PROTOCOL,
+//  pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_TELNET_PROTOCOL,
+//                                TSDB_SML_TIMESTAMP_NANO_SECONDS);
+  pRes = taos_schemaless_insert(taos, (char **)sql1, sizeof(sql1) / sizeof(sql1[0]), TSDB_SML_TELNET_PROTOCOL,
                                 TSDB_SML_TIMESTAMP_NANO_SECONDS);
   printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
   int code = taos_errno(pRes);
@@ -112,13 +114,53 @@ int smlProcess_json1_Test() {
   taos_free_result(pRes);
 
   const char *sql[] = {
-      "[{\"metric\":\"sys.cpu.nice\",\"timestamp\":0,\"value\":18,\"tags\":{\"host\":\"web01\",\"id\":\"t1\",\"dc\":\"lga\"}},{\"metric\":\"sys.cpu.nice\",\"timestamp\":1662344042,\"value\":9,\"tags\":{\"host\":\"web02\",\"dc\":\"lga\"}}]"
+      "[{\"metric\":\"sys.cpu.nice\",\"timestamp\":0,\"value\":18,\"tags\":{\"host\":\"web01\",\"id\":\"t1\",\"dc\":\"lga\"}},{\"metric\":\"sys.cpu.nice\",\"timestamp\":1662344045,\"value\":9,\"tags\":{\"host\":\"web02\",\"dc\":\"lga\"}}]"
   };
-  pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL,
+
+  char *sql1[1] = {0};
+  for(int i = 0; i < 1; i++){
+    sql1[i] = taosMemoryCalloc(1, 1024);
+    strncpy(sql1[i], sql[i], 1023);
+  }
+
+  pRes = taos_schemaless_insert(taos, (char **)sql1, sizeof(sql1) / sizeof(sql1[0]), TSDB_SML_JSON_PROTOCOL,
                                 TSDB_SML_TIMESTAMP_NANO_SECONDS);
-  printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
   int code = taos_errno(pRes);
+  if(code != 0){
+    printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
+  }else{
+    printf("%s result:success\n", __FUNCTION__);
+  }
   taos_free_result(pRes);
+
+  for(int i = 0; i < 1; i++){
+    taosMemoryFree(sql1[i]);
+  }
+
+  const char *sql2[] = {
+      "[{\"metric\":\"sys.cpu.nice\",\"timestamp\":1662344041,\"value\":13,\"tags\":{\"host\":\"web01\",\"dc\":\"lga\"}},{\"metric\":\"sys.cpu.nice\",\"timestamp\":1662344042,\"value\":9,\"tags\":{\"host\":\"web02\",\"dc\":\"lga\"}}]",
+  };
+
+  char *sql3[1] = {0};
+  for(int i = 0; i < 1; i++){
+    sql3[i] = taosMemoryCalloc(1, 1024);
+    strncpy(sql3[i], sql2[i], 1023);
+  }
+
+  pRes = taos_schemaless_insert(taos, (char **)sql3, sizeof(sql3) / sizeof(sql3[0]), TSDB_SML_JSON_PROTOCOL,
+                                TSDB_SML_TIMESTAMP_NANO_SECONDS);
+  code = taos_errno(pRes);
+  if(code != 0){
+    printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
+  }else{
+    printf("%s result:success\n", __FUNCTION__);
+  }
+  taos_free_result(pRes);
+
+  for(int i = 0; i < 1; i++){
+    taosMemoryFree(sql3[i]);
+  }
+
   taos_close(taos);
 
   return code;
@@ -136,13 +178,26 @@ int smlProcess_json2_Test() {
   const char *sql[] = {
       "{\"metric\":\"meter_current0\",\"timestamp\":{\"value\":1662344042,\"type\":\"s\"},\"value\":{\"value\":10.3,\"type\":\"i64\"},\"tags\":{\"groupid\":{\"value\":2,\"type\":\"bigint\"},\"location\":{\"value\":\"北京\",\"type\":\"binary\"},\"id\":\"d1001\"}}"
   };
-  pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL,
+  char *sql1[1] = {0};
+  for(int i = 0; i < 1; i++){
+    sql1[i] = taosMemoryCalloc(1, 1024);
+    strncpy(sql1[i], sql[i], 1023);
+  }
+
+  pRes = taos_schemaless_insert(taos, (char **)sql1, sizeof(sql1) / sizeof(sql1[0]), TSDB_SML_JSON_PROTOCOL,
                                 TSDB_SML_TIMESTAMP_NANO_SECONDS);
-  printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
   int code = taos_errno(pRes);
+  if(code != 0){
+    printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
+  }else{
+    printf("%s result:success\n", __FUNCTION__);
+  }
   taos_free_result(pRes);
   taos_close(taos);
 
+  for(int i = 0; i < 1; i++){
+    taosMemoryFree(sql1[i]);
+  }
   return code;
 }
 
@@ -156,15 +211,29 @@ int smlProcess_json3_Test() {
   taos_free_result(pRes);
 
   const char *sql[] = {
-      "[{\"metric\":\"sys.cpu.nice\",\"timestamp\":0,\"value\":\"18\",\"tags\":{\"host\":\"web01\",\"id\":\"t1\",\"dc\":\"lga\"}}]"
+//      "[{\"metric\":\"sys.cpu.nice3\",\"timestamp\":0,\"value\":\"18\",\"tags\":{\"host\":\"web01\",\"id\":\"t1\",\"dc\":\"lga\"}}]"
+        "{\"metric\": \"dcxnmr\", \"timestamp\": {\"value\": 1626006833639000000, \"type\": \"ns\"}, \"value\": {\"value\": false, \"type\": \"bool\"}, \"tags\": {\"t0\": {\"value\": false, \"type\": \"bool\"}, \"t1\": {\"value\": 127, \"type\": \"tinyint\"}, \"t2\": {\"value\": 32767, \"type\": \"smallint\"}, \"t3\": {\"value\": 2147483647, \"type\": \"int\"}, \"t4\": {\"value\": 9223372036854775807, \"type\": \"bigint\"}, \"t5\": {\"value\": 11.12345027923584, \"type\": \"float\"}, \"t6\": {\"value\": 22.123456789, \"type\": \"double\"}, \"t7\": {\"value\": \"binaryTagValue\", \"type\": \"binary\"}, \"t8\": {\"value\": \"abc{aaa\", \"type\": \"nchar\"}}}"
   };
-  pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_JSON_PROTOCOL,
+  char *sql1[1] = {0};
+  for(int i = 0; i < 1; i++){
+    sql1[i] = taosMemoryCalloc(1, 1024);
+    strncpy(sql1[i], sql[i], 1023);
+  }
+
+  pRes = taos_schemaless_insert(taos, (char **)sql1, sizeof(sql1) / sizeof(sql1[0]), TSDB_SML_JSON_PROTOCOL,
                                 TSDB_SML_TIMESTAMP_NANO_SECONDS);
-  printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
   int code = taos_errno(pRes);
+  if(code != 0){
+    printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
+  }else{
+    printf("%s result:success\n", __FUNCTION__);
+  }
   taos_free_result(pRes);
   taos_close(taos);
 
+  for(int i = 0; i < 1; i++){
+    taosMemoryFree(sql1[i]);
+  }
   return code;
 }
 
@@ -233,7 +302,7 @@ int sml_16384_Test() {
   if(code) return code;
 
   const char *sql1[] = {
-      "qelhxo,id=pnnqhsa,t0=t,t1=127i8 c0=f,c1=127i8,c11=L\"ncharColValue\",c10=t 1626006833639000000",
+      "qelhxo,id=pnnqhsa,t0=t,t1=127i8 c0=f,c1=127i8,c11=L\"ncharColValue\",c10=t 1626006833631000000",
   };
   pRes = taos_schemaless_insert(taos, (char **)sql1, 1, TSDB_SML_LINE_PROTOCOL, 0);
   printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
@@ -797,7 +866,7 @@ int sml_19221_Test() {
   taos_free_result(pRes);
 
   const char *sql[] = {
-      "qelhxo,id=pnnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000\nqelhxo,id=pnnhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000\n#comment\nqelhxo,id=pnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000",
+      "qelhxo,id=pnnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833632000000\nqelhxo,id=pnnhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833633000000\n#comment\nqelhxo,id=pnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833634000000",
   };
 
   pRes = taos_query(taos, "use sml_db");
@@ -936,9 +1005,9 @@ int main(int argc, char *argv[]) {
   ret = smlProcess_json1_Test();
   ASSERT(!ret);
   ret = smlProcess_json2_Test();
-  ASSERT(ret);
+  ASSERT(!ret);
   ret = smlProcess_json3_Test();
-  ASSERT(ret);
+  ASSERT(!ret);
   ret = sml_TD15662_Test();
   ASSERT(!ret);
   ret = sml_TD15742_Test();