#5058

Committed 17 May 2026 01:15AM UTC coverage: 73.387% (-0.02%) from 73.406%

Build # #5058

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat (TDgpt): Dynamic Model Synchronization Enhancements (#35344)

* refactor: do some internal refactor.

* fix: fix multiprocess sync issue.

* feat: add dynamic anomaly detection and forecasting services

* fix: log error message for undeploying model in exception handling

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle undeploy when model exists only on disk

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/286aafa0-c3ce-4c27-b803-2707571e9dc1

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: guard dynamic registry concurrent access

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: tighten service list locking scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: restore prophet support and update tests per review feedback

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: improve test name and move copy inside lock scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* Potential fix for pull request finding

Co-au... (continued)

Coverage Stats

281656 of 383795 relevant lines covered (73.39%)

135114337.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.17

/source/client/src/clientSmlJson.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "clientSml.h"

#define OTD_JSON_SUB_FIELDS_NUM 2

static inline int32_t smlParseMetricFromJSON(SSmlHandle *info, cJSON *metric, SSmlLineInfo *elements) {
  elements->measureLen = strlen(metric->valuestring);
  if (IS_INVALID_TABLE_LEN(elements->measureLen)) {
    uError("SML:0x%" PRIx64 " Metric length is 0 or large than 192", info->id);
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
  }

  elements->measure = metric->valuestring;
  return TSDB_CODE_SUCCESS;
}

const char    *jsonName[OTD_JSON_FIELDS_NUM] = {"metric", "timestamp", "value", "tags"};
static int32_t smlGetJsonElements(cJSON *root, cJSON ***marks) {
  for (int i = 0; i < OTD_JSON_FIELDS_NUM; ++i) {
    cJSON *child = root->child;
    while (child != NULL) {
      if (strcasecmp(child->string, jsonName[i]) == 0) {
        *marks[i] = child;
        break;
      }
      child = child->next;
    }
    if (*marks[i] == NULL) {
      uError("SML %s error, not find mark:%d:%s", __FUNCTION__, i, jsonName[i]);
      return TSDB_CODE_TSC_INVALID_JSON;
    }
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t smlConvertJSONBool(SSmlKv *pVal, char *typeStr, cJSON *value) {
  if (strcasecmp(typeStr, "bool") != 0) {
    uError("SML:invalid type(%s) for JSON Bool", typeStr);
    return TSDB_CODE_TSC_INVALID_JSON_TYPE;
  }
  pVal->type = TSDB_DATA_TYPE_BOOL;
  pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
  pVal->i = value->valueint;

  return TSDB_CODE_SUCCESS;
}

static int32_t smlConvertJSONNumber(SSmlKv *pVal, char *typeStr, cJSON *value) {
  // tinyint
  if (strcasecmp(typeStr, "i8") == 0 || strcasecmp(typeStr, "tinyint") == 0) {
    if (!IS_VALID_TINYINT(value->valuedouble)) {
      uError("SML:JSON value(%f) cannot fit in type(tinyint)", value->valuedouble);
      return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
    }
    pVal->type = TSDB_DATA_TYPE_TINYINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    pVal->i = value->valuedouble;
    return TSDB_CODE_SUCCESS;
  }
  // smallint
  if (strcasecmp(typeStr, "i16") == 0 || strcasecmp(typeStr, "smallint") == 0) {
    if (!IS_VALID_SMALLINT(value->valuedouble)) {
      uError("SML:JSON value(%f) cannot fit in type(smallint)", value->valuedouble);
      return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
    }
    pVal->type = TSDB_DATA_TYPE_SMALLINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    pVal->i = value->valuedouble;
    return TSDB_CODE_SUCCESS;
  }
  // int
  if (strcasecmp(typeStr, "i32") == 0 || strcasecmp(typeStr, "int") == 0) {
    if (!IS_VALID_INT(value->valuedouble)) {
      uError("SML:JSON value(%f) cannot fit in type(int)", value->valuedouble);
      return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
    }
    pVal->type = TSDB_DATA_TYPE_INT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    pVal->i = value->valuedouble;
    return TSDB_CODE_SUCCESS;
  }
  // bigint
  if (strcasecmp(typeStr, "i64") == 0 || strcasecmp(typeStr, "bigint") == 0) {
    pVal->type = TSDB_DATA_TYPE_BIGINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    if (value->valuedouble >= (double)INT64_MAX) {
      pVal->i = INT64_MAX;
    } else if (value->valuedouble <= (double)INT64_MIN) {
      pVal->i = INT64_MIN;
    } else {
      pVal->i = value->valuedouble;
    }
    return TSDB_CODE_SUCCESS;
  }
  // float
  if (strcasecmp(typeStr, "f32") == 0 || strcasecmp(typeStr, "float") == 0) {
    if (!IS_VALID_FLOAT(value->valuedouble)) {
      uError("SML:JSON value(%f) cannot fit in type(float)", value->valuedouble);
      return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
    }
    pVal->type = TSDB_DATA_TYPE_FLOAT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    pVal->f = value->valuedouble;
    return TSDB_CODE_SUCCESS;
  }
  // double
  if (strcasecmp(typeStr, "f64") == 0 || strcasecmp(typeStr, "double") == 0) {
    pVal->type = TSDB_DATA_TYPE_DOUBLE;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    pVal->d = value->valuedouble;
    return TSDB_CODE_SUCCESS;
  }

  // if reach here means type is unsupported
  uError("SML:invalid type(%s) for JSON Number", typeStr);
  return TSDB_CODE_TSC_INVALID_JSON_TYPE;
}

static int32_t smlConvertJSONString(SSmlKv *pVal, char *typeStr, cJSON *value) {
  if (strcasecmp(typeStr, "binary") == 0) {
    pVal->type = TSDB_DATA_TYPE_BINARY;
  } else if (strcasecmp(typeStr, "varbinary") == 0) {
    pVal->type = TSDB_DATA_TYPE_VARBINARY;
  } else if (strcasecmp(typeStr, "nchar") == 0) {
    pVal->type = TSDB_DATA_TYPE_NCHAR;
  } else {
    uError("SML:invalid type(%s) for JSON String", typeStr);
    return TSDB_CODE_TSC_INVALID_JSON_TYPE;
  }
  pVal->length = strlen(value->valuestring);

  if ((pVal->type == TSDB_DATA_TYPE_BINARY || pVal->type == TSDB_DATA_TYPE_VARBINARY) && pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
    return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
  }
  if (pVal->type == TSDB_DATA_TYPE_NCHAR &&
      pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
    return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
  }

  pVal->value = value->valuestring;
  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseValueFromJSONObj(cJSON *root, SSmlKv *kv) {
  int32_t ret = TSDB_CODE_SUCCESS;
  int32_t size = cJSON_GetArraySize(root);

  if (size != OTD_JSON_SUB_FIELDS_NUM) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  cJSON *value = cJSON_GetObjectItem(root, "value");
  if (value == NULL) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  cJSON *type = cJSON_GetObjectItem(root, "type");
  if (!cJSON_IsString(type)) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  switch (value->type) {
    case cJSON_True:
    case cJSON_False: {
      ret = smlConvertJSONBool(kv, type->valuestring, value);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
      }
      break;
    }
    case cJSON_Number: {
      ret = smlConvertJSONNumber(kv, type->valuestring, value);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
      }
      break;
    }
    case cJSON_String: {
      ret = smlConvertJSONString(kv, type->valuestring, value);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
      }
      break;
    }
    default:
      return TSDB_CODE_TSC_INVALID_JSON_TYPE;
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseValueFromJSON(cJSON *root, SSmlKv *kv) {
  switch (root->type) {
    case cJSON_True:
    case cJSON_False: {
      kv->type = TSDB_DATA_TYPE_BOOL;
      kv->length = (int16_t)tDataTypes[kv->type].bytes;
      kv->i = root->valueint;
      break;
    }
    case cJSON_Number: {
      kv->type = TSDB_DATA_TYPE_DOUBLE;
      kv->length = (int16_t)tDataTypes[kv->type].bytes;
      kv->d = root->valuedouble;
      break;
    }
    case cJSON_String: {
      int32_t ret = smlConvertJSONString(kv, "binary", root);
      if (ret != TSDB_CODE_SUCCESS) {
        uError("SML:Failed to parse binary value from JSON Obj");
        return ret;
      }
      break;
    }
    case cJSON_Object: {
      int32_t ret = smlParseValueFromJSONObj(root, kv);
      if (ret != TSDB_CODE_SUCCESS) {
        uError("SML:Failed to parse value from JSON Obj");
        return ret;
      }
      break;
    }
    default:
      return TSDB_CODE_TSC_INVALID_JSON;
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t smlProcessTagJson(SSmlHandle *info, cJSON *tags){
  SArray *preLineKV = info->preLineTagKV;
  taosArrayClearEx(preLineKV, freeSSmlKv);
  int     cnt = 0;

  int32_t tagNum = cJSON_GetArraySize(tags);
  if (unlikely(tagNum == 0)) {
    uError("SML:Tag should not be empty");
    return TSDB_CODE_TSC_INVALID_JSON;
  }
  for (int32_t i = 0; i < tagNum; ++i) {
    cJSON *tag = cJSON_GetArrayItem(tags, i);
    if (unlikely(tag == NULL)) {
      return TSDB_CODE_TSC_INVALID_JSON;
    }
    size_t keyLen = strlen(tag->string);
    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
      uError("SML:Tag key length is 0 or too large than 64");
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
    }

    // add kv to SSmlKv
    SSmlKv kv = {0};
    kv.key = tag->string;
    kv.keyLen = keyLen;

    // value
    int32_t ret = smlParseValueFromJSON(tag, &kv);
    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
      return ret;
    }
    if (taosArrayPush(preLineKV, &kv) == NULL) {
      return terrno;
    }

    if (info->dataFormat && !isSmlTagAligned(info, cnt, &kv)) {
      return TSDB_CODE_TSC_INVALID_JSON;
    }

    cnt++;
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseTagsFromJSON(SSmlHandle *info, cJSON *tags, SSmlLineInfo *elements) {
  if (is_same_child_table_telnet(elements, &info->preLine) == 0) {
    elements->measureTag = info->preLine.measureTag;
    return TSDB_CODE_SUCCESS;
  }
  int32_t code = 0;
  int32_t lino = 0;
  if(info->dataFormat){
    SML_CHECK_CODE(smlProcessSuperTable(info, elements));
  }
  SML_CHECK_CODE(smlProcessTagJson(info, tags));
  SML_CHECK_CODE(smlJoinMeasureTag(elements));
  return smlProcessChildTable(info, elements);

END:
  if(info->reRun){
    return TSDB_CODE_SUCCESS;
  }
  RETURN
}

static int64_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int32_t toPrecision) {
  int32_t size = cJSON_GetArraySize(root);
  if (unlikely(size != OTD_JSON_SUB_FIELDS_NUM)) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  cJSON *value = cJSON_GetObjectItem(root, "value");
  if (unlikely(!cJSON_IsNumber(value))) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  cJSON *type = cJSON_GetObjectItem(root, "type");
  if (unlikely(!cJSON_IsString(type))) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  double timeDouble = value->valuedouble;
  if (unlikely(smlDoubleToInt64OverFlow(timeDouble))) {
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
    return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
  }

  if (timeDouble == 0) {
    return taosGetTimestampNs() / smlFactorNS[toPrecision];
  }

  if (timeDouble < 0) {
    return (int64_t)timeDouble;
  }

  int64_t tsInt64 = (int64_t)timeDouble;
  size_t  typeLen = strlen(type->valuestring);
  if (typeLen == 1 && (type->valuestring[0] == 's' || type->valuestring[0] == 'S')) {
    // seconds
    if (smlFactorS[toPrecision] < INT64_MAX / tsInt64) {
      return tsInt64 * smlFactorS[toPrecision];
    }
    return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
  } else if (typeLen == 2 && (type->valuestring[1] == 's' || type->valuestring[1] == 'S')) {
    switch (type->valuestring[0]) {
      case 'm':
      case 'M':
        // milliseconds
        return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_MILLI, toPrecision);
      case 'u':
      case 'U':
        // microseconds
        return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_MICRO, toPrecision);
      case 'n':
      case 'N':
        return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_NANO, toPrecision);
      default:
        return TSDB_CODE_TSC_INVALID_JSON_TYPE;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON_TYPE;
  }
}

uint8_t smlGetTimestampLen(int64_t num) {
  uint8_t len = 0;
  while ((num /= 10) != 0) {
    len++;
  }
  len++;
  return len;
}

static int64_t smlParseTSFromJSON(SSmlHandle *info, cJSON *timestamp) {
  // Timestamp must be the first KV to parse
  int32_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;
  if (cJSON_IsNumber(timestamp)) {
    // timestamp value 0 indicates current system time
    double timeDouble = timestamp->valuedouble;
    if (unlikely(smlDoubleToInt64OverFlow(timeDouble))) {
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
      return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
    }

    if (unlikely(timeDouble < 0)) {
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is negative", NULL);
      return (int64_t)timeDouble;
    } else if (unlikely(timeDouble == 0)) {
      return taosGetTimestampNs() / smlFactorNS[toPrecision];
    }

    uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);

    int8_t fromPrecision = smlGetTsTypeByLen(tsLen);
    if (unlikely(fromPrecision == -1)) {
      smlBuildInvalidDataMsg(&info->msgBuf,
                             "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", NULL);
      return TSDB_CODE_SML_INVALID_DATA;
    }
    int64_t tsInt64 = (int64_t)timeDouble;
    if (fromPrecision == TSDB_TIME_PRECISION_SECONDS) {
      if (smlFactorS[toPrecision] < INT64_MAX / tsInt64) {
        return tsInt64 * smlFactorS[toPrecision];
      }
      return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;
    } else {
      return convertTimePrecision(timeDouble, fromPrecision, toPrecision);
    }
  } else if (cJSON_IsObject(timestamp)) {
    return smlParseTSFromJSONObj(info, timestamp, toPrecision);
  } else {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);
    return TSDB_CODE_TSC_INVALID_JSON;
  }
}

static int32_t smlParseJSONStringExt(SSmlHandle *info, cJSON *root, SSmlLineInfo *elements) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;

  cJSON *metricJson = NULL;
  cJSON *tsJson = NULL;
  cJSON *valueJson = NULL;
  cJSON *tagsJson = NULL;

  int32_t size = cJSON_GetArraySize(root);
  // outmost json fields has to be exactly 4
  if (size != OTD_JSON_FIELDS_NUM) {
    uError("SML:0x%" PRIx64 " Invalid number of JSON fields in data point %d", info->id, size);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  cJSON **marks[OTD_JSON_FIELDS_NUM] = {&metricJson, &tsJson, &valueJson, &tagsJson};
  SML_CHECK_CODE(smlGetJsonElements(root, marks));
  // Parse metric
  SML_CHECK_CODE(smlParseMetricFromJSON(info, metricJson, elements));
  // Parse metric value
  SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN};
  SML_CHECK_CODE(smlParseValueFromJSON(valueJson, &kv));

  // Parse tags
  elements->tags = cJSON_PrintUnformatted(tagsJson);
  SML_CHECK_NULL(elements->tags);

  elements->tagsLen = strlen(elements->tags);
  SML_CHECK_CODE(smlParseTagsFromJSON(info, tagsJson, elements));

  if (unlikely(info->reRun)) {
    goto END;
  }

  // Parse timestamp
  // notice!!! put ts back to tag to ensure get meta->precision
  int64_t ts = smlParseTSFromJSON(info, tsJson);
  if (unlikely(ts < 0)) {
    char* tmp = cJSON_PrintUnformatted(tsJson);
    if (tmp == NULL) {
      uError("SML:0x%" PRIx64 " Unable to parse timestamp from JSON payload %s %" PRId64, info->id, info->msgBuf.buf, ts);
    } else {
      uError("SML:0x%" PRIx64 " Unable to parse timestamp from JSON payload %s %s %" PRId64, info->id, info->msgBuf.buf,tmp, ts);
      taosMemoryFree(tmp);
    }
    SML_CHECK_CODE(TSDB_CODE_INVALID_TIMESTAMP);
  }
  SSmlKv kvTs = {0};
  smlBuildTsKv(&kvTs, ts);
  if (info->dataFormat){
    code = smlParseEndTelnetJsonFormat(info, elements, &kvTs, &kv);
  } else {
    code = smlParseEndTelnetJsonUnFormat(info, elements, &kvTs, &kv);
  }
  SML_CHECK_CODE(code);
  taosMemoryFreeClear(info->preLine.tags);
  info->preLine = *elements;
  elements->tags = NULL;

END:
  taosMemoryFree(elements->tags);
  RETURN
}

int32_t smlParseJSONExt(SSmlHandle *info, char *payload) {
  int32_t payloadNum = 0;
  int32_t ret = TSDB_CODE_SUCCESS;

  info->root = cJSON_Parse(payload);
  if (unlikely(info->root == NULL)) {
    uError("SML:0x%" PRIx64 " parse json failed:%s", info->id, payload);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  // multiple data points must be sent in JSON array
  if (cJSON_IsArray(info->root)) {
    payloadNum = cJSON_GetArraySize(info->root);
  } else if (cJSON_IsObject(info->root)) {
    payloadNum = 1;
  } else {
    uError("SML:0x%" PRIx64 " Invalid JSON type:%s", info->id, payload);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  info->lineNum = payloadNum;
  cJSON *head = (payloadNum == 1 && cJSON_IsObject(info->root)) ? info->root : info->root->child;

  int    cnt = 0;
  cJSON *dataPoint = head;
  while (dataPoint) {
    if (info->dataFormat) {
      SSmlLineInfo element = {0};
      ret = smlParseJSONStringExt(info, dataPoint, &element);
      if (element.measureTagsLen != 0) taosMemoryFree(element.measureTag);
    } else {
      ret = smlParseJSONStringExt(info, dataPoint, info->lines + cnt);
    }
    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
      uError("SML:0x%" PRIx64 " Invalid JSON Payload 2:%s", info->id, payload);
      return ret;
    }

    if (unlikely(info->reRun)) {
      cnt = 0;
      dataPoint = head;
      info->lineNum = payloadNum;
      ret = smlClearForRerun(info);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
      }
      continue;
    }
    cnt++;
    dataPoint = dataPoint->next;
  }

  return TSDB_CODE_SUCCESS;
}


1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include <ctype.h>
17	#include <stdio.h>
18	#include <stdlib.h>
19	#include <string.h>
20	#include "clientSml.h"
21
22	#define OTD_JSON_SUB_FIELDS_NUM 2
23
24	static inline int32_t smlParseMetricFromJSON(SSmlHandle info, cJSON metric, SSmlLineInfo *elements) {	467,577✔
25	elements->measureLen = strlen(metric->valuestring);	467,577✔
26	if (IS_INVALID_TABLE_LEN(elements->measureLen)) {	467,577✔
27	uError("SML:0x%" PRIx64 " Metric length is 0 or large than 192", info->id);	1,308✔
28	return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;	1,308✔
29	}
30
31	elements->measure = metric->valuestring;	466,281✔
32	return TSDB_CODE_SUCCESS;	466,281✔
33	}
34
35	const char *jsonName[OTD_JSON_FIELDS_NUM] = {"metric", "timestamp", "value", "tags"};
36	static int32_t smlGetJsonElements(cJSON root, cJSON **marks) {	466,909✔
37	for (int i = 0; i < OTD_JSON_FIELDS_NUM; ++i) {	2,336,470✔
38	cJSON *child = root->child;	1,869,561✔
39	while (child != NULL) {	4,674,145✔
40	if (strcasecmp(child->string, jsonName[i]) == 0) {	4,672,209✔
41	*marks[i] = child;	1,870,241✔
42	break;	1,870,241✔
43	}
44	child = child->next;	2,804,584✔
45	}
46	if (*marks[i] == NULL) {	1,872,177✔
47	uError("SML %s error, not find mark:%d:%s", __FUNCTION__, i, jsonName[i]);	×
48	return TSDB_CODE_TSC_INVALID_JSON;	×
49	}
50	}
51	return TSDB_CODE_SUCCESS;	466,909✔
52	}
53
54	static int32_t smlConvertJSONBool(SSmlKv pVal, char typeStr, cJSON *value) {	453,672✔
55	if (strcasecmp(typeStr, "bool") != 0) {	453,672✔
56	uError("SML:invalid type(%s) for JSON Bool", typeStr);	×
57	return TSDB_CODE_TSC_INVALID_JSON_TYPE;	×
58	}
59	pVal->type = TSDB_DATA_TYPE_BOOL;	453,672✔
60	pVal->length = (int16_t)tDataTypes[pVal->type].bytes;	453,672✔
61	pVal->i = value->valueint;	453,672✔
62
63	return TSDB_CODE_SUCCESS;	453,672✔
64	}
65
66	static int32_t smlConvertJSONNumber(SSmlKv pVal, char typeStr, cJSON *value) {	677,730✔
67	// tinyint
68	if (strcasecmp(typeStr, "i8") == 0 \|\| strcasecmp(typeStr, "tinyint") == 0) {	677,730✔
69	if (!IS_VALID_TINYINT(value->valuedouble)) {	118,146✔
70	uError("SML:JSON value(%f) cannot fit in type(tinyint)", value->valuedouble);	2,616✔
71	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	2,616✔
72	}
73	pVal->type = TSDB_DATA_TYPE_TINYINT;	115,542✔
74	pVal->length = (int16_t)tDataTypes[pVal->type].bytes;	115,542✔
75	pVal->i = value->valuedouble;	115,542✔
76	return TSDB_CODE_SUCCESS;	115,542✔
77	}
78	// smallint
79	if (strcasecmp(typeStr, "i16") == 0 \|\| strcasecmp(typeStr, "smallint") == 0) {	559,584✔
80	if (!IS_VALID_SMALLINT(value->valuedouble)) {	114,258✔
81	uError("SML:JSON value(%f) cannot fit in type(smallint)", value->valuedouble);	2,628✔
82	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	2,616✔
83	}
84	pVal->type = TSDB_DATA_TYPE_SMALLINT;	111,630✔
85	pVal->length = (int16_t)tDataTypes[pVal->type].bytes;	111,630✔
86	pVal->i = value->valuedouble;	111,630✔
87	return TSDB_CODE_SUCCESS;	111,630✔
88	}
89	// int
90	if (strcasecmp(typeStr, "i32") == 0 \|\| strcasecmp(typeStr, "int") == 0) {	445,326✔
91	if (!IS_VALID_INT(value->valuedouble)) {	112,923✔
92	uError("SML:JSON value(%f) cannot fit in type(int)", value->valuedouble);	2,616✔
93	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	2,616✔
94	}
95	pVal->type = TSDB_DATA_TYPE_INT;	110,319✔
96	pVal->length = (int16_t)tDataTypes[pVal->type].bytes;	110,319✔
97	pVal->i = value->valuedouble;	110,319✔
98	return TSDB_CODE_SUCCESS;	110,319✔
99	}
100	// bigint
101	if (strcasecmp(typeStr, "i64") == 0 \|\| strcasecmp(typeStr, "bigint") == 0) {	332,403✔
102	pVal->type = TSDB_DATA_TYPE_BIGINT;	112,953✔
103	pVal->length = (int16_t)tDataTypes[pVal->type].bytes;	112,953✔
104	if (value->valuedouble >= (double)INT64_MAX) {	112,953✔
105	pVal->i = INT64_MAX;	103,986✔
106	} else if (value->valuedouble <= (double)INT64_MIN) {	8,967✔
107	pVal->i = INT64_MIN;	1,308✔
108	} else {
109	pVal->i = value->valuedouble;	7,659✔
110	}
111	return TSDB_CODE_SUCCESS;	112,953✔
112	}
113	// float
114	if (strcasecmp(typeStr, "f32") == 0 \|\| strcasecmp(typeStr, "float") == 0) {	219,450✔
115	if (!IS_VALID_FLOAT(value->valuedouble)) {	111,744✔
116	uError("SML:JSON value(%f) cannot fit in type(float)", value->valuedouble);	2,616✔
117	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	2,616✔
118	}
119	pVal->type = TSDB_DATA_TYPE_FLOAT;	109,128✔
120	pVal->length = (int16_t)tDataTypes[pVal->type].bytes;	109,128✔
121	pVal->f = value->valuedouble;	109,128✔
122	return TSDB_CODE_SUCCESS;	109,128✔
123	}
124	// double
125	if (strcasecmp(typeStr, "f64") == 0 \|\| strcasecmp(typeStr, "double") == 0) {	107,706✔
126	pVal->type = TSDB_DATA_TYPE_DOUBLE;	107,706✔
127	pVal->length = (int16_t)tDataTypes[pVal->type].bytes;	107,706✔
128	pVal->d = value->valuedouble;	107,706✔
129	return TSDB_CODE_SUCCESS;	107,706✔
130	}
131
132	// if reach here means type is unsupported
133	uError("SML:invalid type(%s) for JSON Number", typeStr);	×
134	return TSDB_CODE_TSC_INVALID_JSON_TYPE;	×
135	}
136
137	static int32_t smlConvertJSONString(SSmlKv pVal, char typeStr, cJSON *value) {	477,202✔
138	if (strcasecmp(typeStr, "binary") == 0) {	477,202✔
139	pVal->type = TSDB_DATA_TYPE_BINARY;	347,164✔
140	} else if (strcasecmp(typeStr, "varbinary") == 0) {	130,038✔
141	pVal->type = TSDB_DATA_TYPE_VARBINARY;	×
142	} else if (strcasecmp(typeStr, "nchar") == 0) {	130,038✔
143	pVal->type = TSDB_DATA_TYPE_NCHAR;	111,084✔
144	} else {
145	uError("SML:invalid type(%s) for JSON String", typeStr);	18,966✔
146	return TSDB_CODE_TSC_INVALID_JSON_TYPE;	18,966✔
147	}
148	pVal->length = strlen(value->valuestring);	458,248✔
149
150	if ((pVal->type == TSDB_DATA_TYPE_BINARY \|\| pVal->type == TSDB_DATA_TYPE_VARBINARY) && pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {	458,248✔
151	return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;	×
152	}
153	if (pVal->type == TSDB_DATA_TYPE_NCHAR &&	458,248✔
154	pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {	111,084✔
155	return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;	×
156	}
157
158	pVal->value = value->valuestring;	458,248✔
159	return TSDB_CODE_SUCCESS;	458,248✔
160	}
161
162	static int32_t smlParseValueFromJSONObj(cJSON root, SSmlKv kv) {	1,403,180✔
163	int32_t ret = TSDB_CODE_SUCCESS;	1,403,180✔
164	int32_t size = cJSON_GetArraySize(root);	1,403,180✔
165
166	if (size != OTD_JSON_SUB_FIELDS_NUM) {	1,403,180✔
167	return TSDB_CODE_TSC_INVALID_JSON;	×
168	}
169
170	cJSON *value = cJSON_GetObjectItem(root, "value");	1,403,180✔
171	if (value == NULL) {	1,403,180✔
172	return TSDB_CODE_TSC_INVALID_JSON;	×
173	}
174
175	cJSON *type = cJSON_GetObjectItem(root, "type");	1,403,180✔
176	if (!cJSON_IsString(type)) {	1,403,192✔
177	return TSDB_CODE_TSC_INVALID_JSON;	×
178	}
179
180	switch (value->type) {	1,403,180✔
181	case cJSON_True:	453,672✔
182	case cJSON_False: {
183	ret = smlConvertJSONBool(kv, type->valuestring, value);	453,672✔
184	if (ret != TSDB_CODE_SUCCESS) {	453,672✔
185	return ret;	×
186	}
187	break;	453,672✔
188	}
189	case cJSON_Number: {	677,742✔
190	ret = smlConvertJSONNumber(kv, type->valuestring, value);	677,742✔
191	if (ret != TSDB_CODE_SUCCESS) {	677,742✔
192	return ret;	10,464✔
193	}
194	break;	667,278✔
195	}
196	case cJSON_String: {	271,766✔
197	ret = smlConvertJSONString(kv, type->valuestring, value);	271,766✔
198	if (ret != TSDB_CODE_SUCCESS) {	271,778✔
199	return ret;	18,966✔
200	}
201	break;	252,812✔
202	}
203	default:	×
204	return TSDB_CODE_TSC_INVALID_JSON_TYPE;	×
205	}
206
207	return TSDB_CODE_SUCCESS;	1,373,762✔
208	}
209
210	static int32_t smlParseValueFromJSON(cJSON root, SSmlKv kv) {	2,374,805✔
211	switch (root->type) {	2,374,805✔
212	case cJSON_True:	307,091✔
213	case cJSON_False: {
214	kv->type = TSDB_DATA_TYPE_BOOL;	307,091✔
215	kv->length = (int16_t)tDataTypes[kv->type].bytes;	306,437✔
216	kv->i = root->valueint;	307,091✔
217	break;	307,091✔
218	}
219	case cJSON_Number: {	457,828✔
220	kv->type = TSDB_DATA_TYPE_DOUBLE;	457,828✔
221	kv->length = (int16_t)tDataTypes[kv->type].bytes;	457,828✔
222	kv->d = root->valuedouble;	457,828✔
223	break;	457,828✔
224	}
225	case cJSON_String: {	205,436✔
226	int32_t ret = smlConvertJSONString(kv, "binary", root);	205,436✔
227	if (ret != TSDB_CODE_SUCCESS) {	205,436✔
228	uError("SML:Failed to parse binary value from JSON Obj");	×
229	return ret;	×
230	}
231	break;	205,436✔
232	}
233	case cJSON_Object: {	1,403,180✔
234	int32_t ret = smlParseValueFromJSONObj(root, kv);	1,403,180✔
235	if (ret != TSDB_CODE_SUCCESS) {	1,403,192✔
236	uError("SML:Failed to parse value from JSON Obj");	29,430✔
237	return ret;	29,430✔
238	}
239	break;	1,373,762✔
240	}
241	default:	1,308✔
242	return TSDB_CODE_TSC_INVALID_JSON;	1,308✔
243	}
244
245	return TSDB_CODE_SUCCESS;	2,344,117✔
246	}
247
248	static int32_t smlProcessTagJson(SSmlHandle info, cJSON tags){	209,619✔
249	SArray *preLineKV = info->preLineTagKV;	209,619✔
250	taosArrayClearEx(preLineKV, freeSSmlKv);	209,619✔
251	int cnt = 0;	209,619✔
252
253	int32_t tagNum = cJSON_GetArraySize(tags);	209,619✔
254	if (unlikely(tagNum == 0)) {	209,619✔
255	uError("SML:Tag should not be empty");	1,308✔
256	return TSDB_CODE_TSC_INVALID_JSON;	1,308✔
257	}
258	for (int32_t i = 0; i < tagNum; ++i) {	2,099,217✔
259	cJSON *tag = cJSON_GetArrayItem(tags, i);	1,909,189✔
260	if (unlikely(tag == NULL)) {	1,909,228✔
261	return TSDB_CODE_TSC_INVALID_JSON;	×
262	}
263	size_t keyLen = strlen(tag->string);	1,909,228✔
264	if (unlikely(IS_INVALID_COL_LEN(keyLen))) {	1,909,228✔
265	uError("SML:Tag key length is 0 or too large than 64");	654✔
266	return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;	654✔
267	}
268
269	// add kv to SSmlKv
270	SSmlKv kv = {0};	1,908,574✔
271	kv.key = tag->string;	1,908,574✔
272	kv.keyLen = keyLen;	1,908,574✔
273
274	// value
275	int32_t ret = smlParseValueFromJSON(tag, &kv);	1,908,574✔
276	if (unlikely(ret != TSDB_CODE_SUCCESS)) {	1,908,574✔
277	return ret;	13,080✔
278	}
279	if (taosArrayPush(preLineKV, &kv) == NULL) {	1,895,481✔
280	return terrno;	×
281	}
282
283	if (info->dataFormat && !isSmlTagAligned(info, cnt, &kv)) {	1,895,481✔
284	return TSDB_CODE_TSC_INVALID_JSON;	4,536✔
285	}
286
287	cnt++;	1,890,906✔
288	}
289	return TSDB_CODE_SUCCESS;	190,028✔
290	}
291
292	static int32_t smlParseTagsFromJSON(SSmlHandle info, cJSON tags, SSmlLineInfo *elements) {	448,612✔
293	if (is_same_child_table_telnet(elements, &info->preLine) == 0) {	448,612✔
294	elements->measureTag = info->preLine.measureTag;	57,066✔
295	return TSDB_CODE_SUCCESS;	57,066✔
296	}
297	int32_t code = 0;	391,557✔
298	int32_t lino = 0;	391,557✔
299	if(info->dataFormat){	391,557✔
300	SML_CHECK_CODE(smlProcessSuperTable(info, elements));	201,709✔
301	}
302	SML_CHECK_CODE(smlProcessTagJson(info, tags));	209,632✔
303	SML_CHECK_CODE(smlJoinMeasureTag(elements));	190,041✔
304	return smlProcessChildTable(info, elements);	190,054✔
305
306	END:	201,516✔
307	if(info->reRun){	201,516✔
308	return TSDB_CODE_SUCCESS;	184,444✔
309	}
310	RETURN	17,072✔
311	}
312
313	static int64_t smlParseTSFromJSONObj(SSmlHandle info, cJSON root, int32_t toPrecision) {	146,382✔
314	int32_t size = cJSON_GetArraySize(root);	146,382✔
315	if (unlikely(size != OTD_JSON_SUB_FIELDS_NUM)) {	146,382✔
316	smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);	×
317	return TSDB_CODE_TSC_INVALID_JSON;	×
318	}
319
320	cJSON *value = cJSON_GetObjectItem(root, "value");	146,382✔
321	if (unlikely(!cJSON_IsNumber(value))) {	146,382✔
322	smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);	1,962✔
323	return TSDB_CODE_TSC_INVALID_JSON;	1,962✔
324	}
325
326	cJSON *type = cJSON_GetObjectItem(root, "type");	144,420✔
327	if (unlikely(!cJSON_IsString(type))) {	144,408✔
328	smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);	×
329	return TSDB_CODE_TSC_INVALID_JSON;	×
330	}
331
332	double timeDouble = value->valuedouble;	144,420✔
333	if (unlikely(smlDoubleToInt64OverFlow(timeDouble))) {	144,420✔
334	smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);	×
335	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	×
336	}
337
338	if (timeDouble == 0) {	144,420✔
339	return taosGetTimestampNs() / smlFactorNS[toPrecision];	×
340	}
341
342	if (timeDouble < 0) {	144,420✔
343	return (int64_t)timeDouble;	×
344	}
345
346	int64_t tsInt64 = (int64_t)timeDouble;	144,420✔
347	size_t typeLen = strlen(type->valuestring);	144,420✔
348	if (typeLen == 1 && (type->valuestring[0] == 's' \|\| type->valuestring[0] == 'S')) {	144,420✔
349	// seconds
350	if (smlFactorS[toPrecision] < INT64_MAX / tsInt64) {	1,967✔
351	return tsInt64 * smlFactorS[toPrecision];	1,967✔
352	}
353	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	×
354	} else if (typeLen == 2 && (type->valuestring[1] == 's' \|\| type->valuestring[1] == 'S')) {	142,453✔
355	switch (type->valuestring[0]) {	142,453✔
356	case 'm':	20,155✔
357	case 'M':
358	// milliseconds
359	return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_MILLI, toPrecision);	20,155✔
360	case 'u':	×
361	case 'U':
362	// microseconds
363	return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_MICRO, toPrecision);	×
364	case 'n':	122,298✔
365	case 'N':
366	return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_NANO, toPrecision);	122,298✔
367	default:	×
368	return TSDB_CODE_TSC_INVALID_JSON_TYPE;	×
369	}
370	} else {
371	return TSDB_CODE_TSC_INVALID_JSON_TYPE;	×
372	}
373	}
374
375	uint8_t smlGetTimestampLen(int64_t num) {	96,210✔
376	uint8_t len = 0;	96,210✔
377	while ((num /= 10) != 0) {	1,105,598✔
378	len++;	1,009,388✔
379	}
380	len++;	96,210✔
381	return len;	96,210✔
382	}
383
384	static int64_t smlParseTSFromJSON(SSmlHandle info, cJSON timestamp) {	247,107✔
385	// Timestamp must be the first KV to parse
386	int32_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;	247,107✔
387	if (cJSON_IsNumber(timestamp)) {	247,107✔
388	// timestamp value 0 indicates current system time
389	double timeDouble = timestamp->valuedouble;	98,776✔
390	if (unlikely(smlDoubleToInt64OverFlow(timeDouble))) {	98,776✔
391	smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);	×
392	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	×
393	}
394
395	if (unlikely(timeDouble < 0)) {	98,763✔
396	smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is negative", NULL);	×
397	return (int64_t)timeDouble;	×
398	} else if (unlikely(timeDouble == 0)) {	98,763✔
399	return taosGetTimestampNs() / smlFactorNS[toPrecision];	7,878✔
400	}
401
402	uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);	94,824✔
403
404	int8_t fromPrecision = smlGetTsTypeByLen(tsLen);	94,837✔
405	if (unlikely(fromPrecision == -1)) {	94,824✔
406	smlBuildInvalidDataMsg(&info->msgBuf,	×
407	"timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", NULL);
408	return TSDB_CODE_SML_INVALID_DATA;	×
409	}
410	int64_t tsInt64 = (int64_t)timeDouble;	94,824✔
411	if (fromPrecision == TSDB_TIME_PRECISION_SECONDS) {	94,824✔
412	if (smlFactorS[toPrecision] < INT64_MAX / tsInt64) {	43,238✔
413	return tsInt64 * smlFactorS[toPrecision];	43,238✔
414	}
415	return TSDB_CODE_TSC_VALUE_OUT_OF_RANGE;	×
416	} else {
417	return convertTimePrecision(timeDouble, fromPrecision, toPrecision);	51,586✔
418	}
419	} else if (cJSON_IsObject(timestamp)) {	148,344✔
420	return smlParseTSFromJSONObj(info, timestamp, toPrecision);	146,382✔
421	} else {
422	smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);	1,962✔
423	return TSDB_CODE_TSC_INVALID_JSON;	1,962✔
424	}
425	}
426
427	static int32_t smlParseJSONStringExt(SSmlHandle info, cJSON root, SSmlLineInfo *elements) {	469,526✔
428	int32_t code = TSDB_CODE_SUCCESS;	469,526✔
429	int32_t lino = 0;	469,526✔
430
431	cJSON *metricJson = NULL;	469,526✔
432	cJSON *tsJson = NULL;	469,526✔
433	cJSON *valueJson = NULL;	470,180✔
434	cJSON *tagsJson = NULL;	470,180✔
435
436	int32_t size = cJSON_GetArraySize(root);	469,526✔
437	// outmost json fields has to be exactly 4
438	if (size != OTD_JSON_FIELDS_NUM) {	470,180✔
439	uError("SML:0x%" PRIx64 " Invalid number of JSON fields in data point %d", info->id, size);	2,616✔
440	return TSDB_CODE_TSC_INVALID_JSON;	2,616✔
441	}
442
443	cJSON **marks[OTD_JSON_FIELDS_NUM] = {&metricJson, &tsJson, &valueJson, &tagsJson};	467,564✔
444	SML_CHECK_CODE(smlGetJsonElements(root, marks));	467,564✔
445	// Parse metric
446	SML_CHECK_CODE(smlParseMetricFromJSON(info, metricJson, elements));	467,577✔
447	// Parse metric value
448	SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN};	466,281✔
449	SML_CHECK_CODE(smlParseValueFromJSON(valueJson, &kv));	465,627✔
450
451	// Parse tags
452	elements->tags = cJSON_PrintUnformatted(tagsJson);	448,623✔
453	SML_CHECK_NULL(elements->tags);	448,624✔
454
455	elements->tagsLen = strlen(elements->tags);	448,624✔
456	SML_CHECK_CODE(smlParseTagsFromJSON(info, tagsJson, elements));	448,624✔
457
458	if (unlikely(info->reRun)) {	431,564✔
459	goto END;	184,444✔
460	}
461
462	// Parse timestamp
463	// notice!!! put ts back to tag to ensure get meta->precision
464	int64_t ts = smlParseTSFromJSON(info, tsJson);	247,120✔
465	if (unlikely(ts < 0)) {	247,107✔
466	char* tmp = cJSON_PrintUnformatted(tsJson);	3,924✔
467	if (tmp == NULL) {	3,924✔
468	uError("SML:0x%" PRIx64 " Unable to parse timestamp from JSON payload %s %" PRId64, info->id, info->msgBuf.buf, ts);	×
469	} else {
470	uError("SML:0x%" PRIx64 " Unable to parse timestamp from JSON payload %s %s %" PRId64, info->id, info->msgBuf.buf,tmp, ts);	3,924✔
471	taosMemoryFree(tmp);	3,924✔
472	}
473	SML_CHECK_CODE(TSDB_CODE_INVALID_TIMESTAMP);	3,924✔
474	}
475	SSmlKv kvTs = {0};	243,183✔
476	smlBuildTsKv(&kvTs, ts);	243,183✔
477	if (info->dataFormat){	243,170✔
478	code = smlParseEndTelnetJsonFormat(info, elements, &kvTs, &kv);	56,124✔
479	} else {
480	code = smlParseEndTelnetJsonUnFormat(info, elements, &kvTs, &kv);	187,046✔
481	}
482	SML_CHECK_CODE(code);	243,144✔
483	taosMemoryFreeClear(info->preLine.tags);	243,144✔
484	info->preLine = *elements;	243,170✔
485	elements->tags = NULL;	243,170✔
486
487	END:	467,576✔
488	taosMemoryFree(elements->tags);	467,576✔
489	RETURN	467,589✔
490	}
491
492	int32_t smlParseJSONExt(SSmlHandle info, char payload) {	223,257✔
493	int32_t payloadNum = 0;	223,257✔
494	int32_t ret = TSDB_CODE_SUCCESS;	223,257✔
495
496	info->root = cJSON_Parse(payload);	223,257✔
497	if (unlikely(info->root == NULL)) {	223,911✔
498	uError("SML:0x%" PRIx64 " parse json failed:%s", info->id, payload);	2,616✔
499	return TSDB_CODE_TSC_INVALID_JSON;	2,616✔
500	}
501
502	// multiple data points must be sent in JSON array
503	if (cJSON_IsArray(info->root)) {	221,295✔
504	payloadNum = cJSON_GetArraySize(info->root);	27,023✔
505	} else if (cJSON_IsObject(info->root)) {	194,272✔
506	payloadNum = 1;	194,260✔
507	} else {
508	uError("SML:0x%" PRIx64 " Invalid JSON type:%s", info->id, payload);	×
509	return TSDB_CODE_TSC_INVALID_JSON;	×
510	}
511
512	info->lineNum = payloadNum;	221,283✔
513	cJSON *head = (payloadNum == 1 && cJSON_IsObject(info->root)) ? info->root : info->root->child;	220,629✔
514
515	int cnt = 0;	221,283✔
516	cJSON *dataPoint = head;	221,283✔
517	while (dataPoint) {	648,884✔
518	if (info->dataFormat) {	469,538✔
519	SSmlLineInfo element = {0};	264,180✔
520	ret = smlParseJSONStringExt(info, dataPoint, &element);	264,834✔
521	if (element.measureTagsLen != 0) taosMemoryFree(element.measureTag);	264,847✔
522	} else {
523	ret = smlParseJSONStringExt(info, dataPoint, info->lines + cnt);	205,358✔
524	}
525	if (unlikely(ret != TSDB_CODE_SUCCESS)) {	470,179✔
526	uError("SML:0x%" PRIx64 " Invalid JSON Payload 2:%s", info->id, payload);	42,578✔
527	return ret;	42,578✔
528	}
529
530	if (unlikely(info->reRun)) {	427,601✔
531	cnt = 0;	184,444✔
532	dataPoint = head;	184,444✔
533	info->lineNum = payloadNum;	184,444✔
534	ret = smlClearForRerun(info);	184,444✔
535	if (ret != TSDB_CODE_SUCCESS) {	184,444✔
536	return ret;	×
537	}
538	continue;	184,444✔
539	}
540	cnt++;	243,157✔
541	dataPoint = dataPoint->next;	243,157✔
542	}
543
544	return TSDB_CODE_SUCCESS;	179,346✔
545	}
546

taosdata / TDengine / #5058

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous