enh:[TD-32166]refactor code in sml

source/client/inc/clientSml.h

@@ -92,6 +92,22 @@ extern "C" {
     }                                        \
   }
 
+#define SML_CHECK_CODE(CMD)                \
+  do {                                     \
+    code = (CMD);                          \
+    if (TSDB_CODE_SUCCESS != code) {       \
+      goto END;                            \
+    }                                      \
+  } while (0)
+
+#define SML_CHECK_NULL(CMD)                \
+  do {                                     \
+    if (NULL == (CMD)) {                   \
+      code = terrno;                       \
+      goto END;                            \
+    }                                      \
+  } while (0)
+
 typedef enum {
   SCHEMA_ACTION_NULL,
   SCHEMA_ACTION_CREATE_STABLE,
@@ -211,13 +227,10 @@ typedef struct {
 extern int64_t smlFactorNS[];
 extern int64_t smlFactorS[];
 
-typedef int32_t (*_equal_fn_sml)(const void *, const void *);
-
 int32_t       smlBuildSmlInfo(TAOS *taos, SSmlHandle **handle);
 void          smlDestroyInfo(SSmlHandle *info);
 int           smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset);
-int           smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset);
+int           smlJsonParseObj(char **start, char *end, SSmlLineInfo *element, int8_t *offset);
-bool          smlParseNumberOld(SSmlKv *kvVal, SSmlMsgBuf *msg);
 void          smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2);
 int32_t       smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg);
 int64_t       smlGetTimeValue(const char *value, int32_t len, uint8_t fromPrecision, uint8_t toPrecision);
@@ -246,7 +259,8 @@ int32_t         smlProcessChildTable(SSmlHandle *info, SSmlLineInfo *elements);
 int32_t         smlProcessSuperTable(SSmlHandle *info, SSmlLineInfo *elements);
 int32_t         smlJoinMeasureTag(SSmlLineInfo *elements);
 void            smlBuildTsKv(SSmlKv *kv, int64_t ts);
-int32_t         smlParseEndTelnetJson(SSmlHandle *info, SSmlLineInfo *elements, SSmlKv *kvTs, SSmlKv *kv);
+int32_t         smlParseEndTelnetJsonFormat(SSmlHandle *info, SSmlLineInfo *elements, SSmlKv *kvTs, SSmlKv *kv);
+int32_t         smlParseEndTelnetJsonUnFormat(SSmlHandle *info, SSmlLineInfo *elements, SSmlKv *kvTs, SSmlKv *kv);
 int32_t         smlParseEndLine(SSmlHandle *info, SSmlLineInfo *elements, SSmlKv *kvTs);
 
 static inline bool smlDoubleToInt64OverFlow(double num) {

source/client/src/clientSmlJson.c

@@ -184,7 +184,7 @@ int smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset) {
   return 0;
 }
 
-int smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset) {
+int smlJsonParseObj(char **start, char *end, SSmlLineInfo *element, int8_t *offset) {
   int index = 0;
   while (*(*start)) {
     if ((*start)[0] != '"') {
@@ -197,6 +197,10 @@ int smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset) {
       return TSDB_CODE_TSC_INVALID_JSON;
     }
 
+    if ((*start) + offset[index] >= end){
+      return TSDB_CODE_TSC_INVALID_JSON;
+    }
+
     if ((*start)[1] == 'm') {
       (*start) += offset[index++];
       element->measure = *start;
@@ -539,28 +543,19 @@ static int32_t smlProcessTagJson(SSmlHandle *info, cJSON *tags){
 }
 
 static int32_t smlParseTagsFromJSON(SSmlHandle *info, cJSON *tags, SSmlLineInfo *elements) {
-  int32_t ret = 0;
+  int32_t code = 0;
   if(info->dataFormat){
-    ret = smlProcessSuperTable(info, elements);
+    SML_CHECK_CODE(smlProcessSuperTable(info, elements));
-    if(ret != 0){
-      if(info->reRun){
-        return TSDB_CODE_SUCCESS;
-      }
-      return ret;
-    }
-  }
-  ret = smlProcessTagJson(info, tags);
-  if(ret != 0){
-    if(info->reRun){
-      return TSDB_CODE_SUCCESS;
-    }
-    return ret;
-  }
-  ret = smlJoinMeasureTag(elements);
-  if(ret != 0){
-    return ret;
   }
+  SML_CHECK_CODE(smlProcessTagJson(info, tags));
+  SML_CHECK_CODE(smlJoinMeasureTag(elements));
   return smlProcessChildTable(info, elements);
+
+END:
+  if(info->reRun){
+    return TSDB_CODE_SUCCESS;
+  }
+  return code;
 }
 
 static int64_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int32_t toPrecision) {
@@ -757,19 +752,18 @@ static int32_t smlParseJSONStringExt(SSmlHandle *info, cJSON *root, SSmlLineInfo
   }
   SSmlKv kvTs = {0};
   smlBuildTsKv(&kvTs, ts);
-
+  if (info->dataFormat){
-  return smlParseEndTelnetJson(info, elements, &kvTs, &kv);
+    ret = smlParseEndTelnetJsonFormat(info, elements, &kvTs, &kv);
+  } else {
+    ret = smlParseEndTelnetJsonUnFormat(info, elements, &kvTs, &kv);
+  }
+  return ret;
 }
 
 static int32_t smlParseJSONExt(SSmlHandle *info, char *payload) {
   int32_t payloadNum = 0;
   int32_t ret = TSDB_CODE_SUCCESS;
 
-  if (unlikely(payload == NULL)) {
-    uError("SML:0x%" PRIx64 " empty JSON Payload", info->id);
-    return TSDB_CODE_TSC_INVALID_JSON;
-  }
-
   info->root = cJSON_Parse(payload);
   if (unlikely(info->root == NULL)) {
     uError("SML:0x%" PRIx64 " parse json failed:%s", info->id, payload);
@@ -836,13 +830,13 @@ static int32_t smlParseJSONExt(SSmlHandle *info, char *payload) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *elements) {
+static int32_t smlParseJSONString(SSmlHandle *info, char **start, char *end, SSmlLineInfo *elements) {
   int32_t ret = TSDB_CODE_SUCCESS;
 
   if (info->offset[0] == 0) {
     ret = smlJsonParseObjFirst(start, elements, info->offset);
   } else {
-    ret = smlJsonParseObj(start, elements, info->offset);
+    ret = smlJsonParseObj(start, end, elements, info->offset);
   }
 
   if (ret != TSDB_CODE_SUCCESS) {
@@ -870,7 +864,7 @@ static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *
       elements->cols[elements->colsLen] = tmp;
       return TSDB_CODE_TSC_INVALID_JSON;
     }
-    if (taosArrayPush(info->tagJsonArray, &valueJson) == NULL){
+    if (taosArrayPush(info->valueJsonArray, &valueJson) == NULL){
       cJSON_Delete(valueJson);
       elements->cols[elements->colsLen] = tmp;
       return terrno;
@@ -945,20 +939,26 @@ static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *
   SSmlKv kvTs = {0};
   smlBuildTsKv(&kvTs, ts);
 
-  return smlParseEndTelnetJson(info, elements, &kvTs, &kv);
+  if (info->dataFormat){
+    ret = smlParseEndTelnetJsonFormat(info, elements, &kvTs, &kv);
+  } else {
+    ret = smlParseEndTelnetJsonUnFormat(info, elements, &kvTs, &kv);
+  }
+  return ret;
 }
 
 int32_t smlParseJSON(SSmlHandle *info, char *payload) {
-  int32_t payloadNum = 1 << 15;
+  int32_t payloadNum = 1 << 10;
   int32_t ret = TSDB_CODE_SUCCESS;
 
   uDebug("SML:0x%" PRIx64 "json:%s", info->id, payload);
   int   cnt = 0;
   char *dataPointStart = payload;
+  char *dataPointEnd   = payload + strlen(payload);
   while (1) {
     if (info->dataFormat) {
       SSmlLineInfo element = {0};
-      ret = smlParseJSONString(info, &dataPointStart, &element);
+      ret = smlParseJSONString(info, &dataPointStart, dataPointEnd, &element);
       if (element.measureTagsLen != 0) taosMemoryFree(element.measureTag);
     } else {
       if (cnt >= payloadNum) {
@@ -971,7 +971,7 @@ int32_t smlParseJSON(SSmlHandle *info, char *payload) {
         info->lines = (SSmlLineInfo *)tmp;
         (void)memset(info->lines + cnt, 0, (payloadNum - cnt) * sizeof(SSmlLineInfo));
       }
-      ret = smlParseJSONString(info, &dataPointStart, info->lines + cnt);
+      ret = smlParseJSONString(info, &dataPointStart, dataPointEnd, info->lines + cnt);
       if ((info->lines + cnt)->measure == NULL) break;
     }
     if (unlikely(ret != TSDB_CODE_SUCCESS)) {

source/client/src/clientSmlLine.c

@@ -298,7 +298,7 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
     }
 
     if (info->dataFormat && !isSmlTagAligned(info, cnt, &kv)) {
-      return TSDB_CODE_TSC_INVALID_JSON;
+      return TSDB_CODE_SML_INVALID_DATA;
     }
 
     cnt++;
@@ -311,31 +311,23 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
 }
 
 static int32_t smlParseTagLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLineInfo *elements) {
+  int32_t code = 0;
   bool isSameCTable = IS_SAME_CHILD_TABLE;
   if(isSameCTable){
     return TSDB_CODE_SUCCESS;
   }
 
-  int32_t ret = 0;
   if(info->dataFormat){
-    ret = smlProcessSuperTable(info, elements);
+    SML_CHECK_CODE(smlProcessSuperTable(info, elements));
-    if(ret != 0){
-      if(info->reRun){
-        return TSDB_CODE_SUCCESS;
-      }
-      return ret;
-    }
   }
-
+  SML_CHECK_CODE(smlProcessTagLine(info, sql, sqlEnd));
-  ret = smlProcessTagLine(info, sql, sqlEnd);
-  if(ret != 0){
-    if (info->reRun){
-      return TSDB_CODE_SUCCESS;
-    }
-    return ret;
-  }
-
   return smlProcessChildTable(info, elements);
+
+END:
+  if(info->reRun){
+    return TSDB_CODE_SUCCESS;
+  }
+  return code;
 }
 
 static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLineInfo *currElement) {

source/client/src/clientSmlTelnet.c

@@ -148,31 +148,20 @@ static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SS
     return TSDB_CODE_SUCCESS;
   }
 
-  int32_t ret = 0;
+  int32_t code = 0;
   if(info->dataFormat){
-    ret = smlProcessSuperTable(info, elements);
+    SML_CHECK_CODE(smlProcessSuperTable(info, elements));
-    if(ret != 0){
-      if(info->reRun){
-        return TSDB_CODE_SUCCESS;
-      }
-      return ret;
-    }
-  }
-
-  ret = smlProcessTagTelnet(info, data, sqlEnd);
-  if(ret != 0){
-    if (info->reRun){
-      return TSDB_CODE_SUCCESS;
-    }
-    return ret;
-  }
-
-  ret = smlJoinMeasureTag(elements);
-  if(ret != 0){
-    return ret;
   }
+  SML_CHECK_CODE(smlProcessTagTelnet(info, data, sqlEnd));
+  SML_CHECK_CODE(smlJoinMeasureTag(elements));
 
   return smlProcessChildTable(info, elements);
+
+END:
+  if(info->reRun){
+    return TSDB_CODE_SUCCESS;
+  }
+  return code;
 }
 
 // format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
@@ -239,5 +228,10 @@ int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
     kvTs.i = convertTimePrecision(kvTs.i, TSDB_TIME_PRECISION_NANO, info->currSTableMeta->tableInfo.precision);
   }
 
-  return smlParseEndTelnetJson(info, elements, &kvTs, &kv);
+  if (info->dataFormat){
+    ret = smlParseEndTelnetJsonFormat(info, elements, &kvTs, &kv);
+  } else {
+    ret = smlParseEndTelnetJsonUnFormat(info, elements, &kvTs, &kv);
+  }
+  return ret;
 }

source/client/test/smlTest.cpp

@@ -591,6 +591,104 @@ TEST(testCase, smlParseTelnetLine_Test) {
 //  smlDestroyInfo(info);
 //}
 
+bool smlParseNumberOld(SSmlKv *kvVal, SSmlMsgBuf *msg) {
+  const char *pVal = kvVal->value;
+  int32_t     len = kvVal->length;
+  char       *endptr = NULL;
+  double      result = taosStr2Double(pVal, &endptr);
+  if (pVal == endptr) {
+    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
+    return false;
+  }
+
+  int32_t left = len - (endptr - pVal);
+  if (left == 0 || (left == 3 && strncasecmp(endptr, "f64", left) == 0)) {
+    kvVal->type = TSDB_DATA_TYPE_DOUBLE;
+    kvVal->d = result;
+  } else if ((left == 3 && strncasecmp(endptr, "f32", left) == 0)) {
+    if (!IS_VALID_FLOAT(result)) {
+      smlBuildInvalidDataMsg(msg, "float out of range[-3.402823466e+38,3.402823466e+38]", pVal);
+      return false;
+    }
+    kvVal->type = TSDB_DATA_TYPE_FLOAT;
+    kvVal->f = (float)result;
+  } else if ((left == 1 && *endptr == 'i') || (left == 3 && strncasecmp(endptr, "i64", left) == 0)) {
+    if (smlDoubleToInt64OverFlow(result)) {
+      errno = 0;
+      int64_t tmp = taosStr2Int64(pVal, &endptr, 10);
+      if (errno == ERANGE) {
+        smlBuildInvalidDataMsg(msg, "big int out of range[-9223372036854775808,9223372036854775807]", pVal);
+        return false;
+      }
+      kvVal->type = TSDB_DATA_TYPE_BIGINT;
+      kvVal->i = tmp;
+      return true;
+    }
+    kvVal->type = TSDB_DATA_TYPE_BIGINT;
+    kvVal->i = (int64_t)result;
+  } else if ((left == 1 && *endptr == 'u') || (left == 3 && strncasecmp(endptr, "u64", left) == 0)) {
+    if (result >= (double)UINT64_MAX || result < 0) {
+      errno = 0;
+      uint64_t tmp = taosStr2UInt64(pVal, &endptr, 10);
+      if (errno == ERANGE || result < 0) {
+        smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", pVal);
+        return false;
+      }
+      kvVal->type = TSDB_DATA_TYPE_UBIGINT;
+      kvVal->u = tmp;
+      return true;
+    }
+    kvVal->type = TSDB_DATA_TYPE_UBIGINT;
+    kvVal->u = result;
+  } else if (left == 3 && strncasecmp(endptr, "i32", left) == 0) {
+    if (!IS_VALID_INT(result)) {
+      smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", pVal);
+      return false;
+    }
+    kvVal->type = TSDB_DATA_TYPE_INT;
+    kvVal->i = result;
+  } else if (left == 3 && strncasecmp(endptr, "u32", left) == 0) {
+    if (!IS_VALID_UINT(result)) {
+      smlBuildInvalidDataMsg(msg, "unsigned int out of range[0,4294967295]", pVal);
+      return false;
+    }
+    kvVal->type = TSDB_DATA_TYPE_UINT;
+    kvVal->u = result;
+  } else if (left == 3 && strncasecmp(endptr, "i16", left) == 0) {
+    if (!IS_VALID_SMALLINT(result)) {
+      smlBuildInvalidDataMsg(msg, "small int our of range[-32768,32767]", pVal);
+      return false;
+    }
+    kvVal->type = TSDB_DATA_TYPE_SMALLINT;
+    kvVal->i = result;
+  } else if (left == 3 && strncasecmp(endptr, "u16", left) == 0) {
+    if (!IS_VALID_USMALLINT(result)) {
+      smlBuildInvalidDataMsg(msg, "unsigned small int out of rang[0,65535]", pVal);
+      return false;
+    }
+    kvVal->type = TSDB_DATA_TYPE_USMALLINT;
+    kvVal->u = result;
+  } else if (left == 2 && strncasecmp(endptr, "i8", left) == 0) {
+    if (!IS_VALID_TINYINT(result)) {
+      smlBuildInvalidDataMsg(msg, "tiny int out of range[-128,127]", pVal);
+      return false;
+    }
+    kvVal->type = TSDB_DATA_TYPE_TINYINT;
+    kvVal->i = result;
+  } else if (left == 2 && strncasecmp(endptr, "u8", left) == 0) {
+    if (!IS_VALID_UTINYINT(result)) {
+      smlBuildInvalidDataMsg(msg, "unsigned tiny int out of range[0,255]", pVal);
+      return false;
+    }
+    kvVal->type = TSDB_DATA_TYPE_UTINYINT;
+    kvVal->u = result;
+  } else {
+    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
+    return false;
+  }
+  return true;
+}
+
 TEST(testCase, smlParseNumber_performance_Test) {
   char       msg[256] = {0};
   SSmlMsgBuf msgBuf;

source/libs/parser/src/parInsertSml.c

@@ -468,35 +468,38 @@ end:
 int32_t smlInitHandle(SQuery** query) {
   *query = NULL;
   SQuery* pQuery = NULL;
+  SVnodeModifyOpStmt* stmt = NULL;
+
   int32_t code = nodesMakeNode(QUERY_NODE_QUERY, (SNode**)&pQuery);
-  if (NULL == pQuery) {
+  if (code != 0) {
     uError("create pQuery error");
-    return code;
+    goto END;
   }
   pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
   pQuery->haveResultSet = false;
   pQuery->msgType = TDMT_VND_SUBMIT;
-  SVnodeModifyOpStmt* stmt = NULL;
   code = nodesMakeNode(QUERY_NODE_VNODE_MODIFY_STMT, (SNode**)&stmt);
-  if (NULL == stmt) {
+  if (code != 0) {
     uError("create SVnodeModifyOpStmt error");
-    qDestroyQuery(pQuery);
+    goto END;
-    return code;
   }
   stmt->pTableBlockHashObj = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
   if (stmt->pTableBlockHashObj == NULL){
     uError("create pTableBlockHashObj error");
-    qDestroyQuery(pQuery);
+    code = terrno;
-    nodesDestroyNode((SNode*)stmt);
+    goto END;
-    return terrno;
   }
   stmt->freeHashFunc = insDestroyTableDataCxtHashMap;
   stmt->freeArrayFunc = insDestroyVgroupDataCxtList;
 
   pQuery->pRoot = (SNode*)stmt;
   *query = pQuery;
+  return code;
 
-  return TSDB_CODE_SUCCESS;
+END:
+  nodesDestroyNode((SNode*)stmt);
+  qDestroyQuery(pQuery);
+  return code;
 }
 
 int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash) {