clientSml.c

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

#include "query.h"
#include "taos.h"
#include "taoserror.h"
#include "tdef.h"
#include "tlog.h"
#include "tmsg.h"
#include "ttime.h"
#include "ttypes.h"
#include "tcommon.h"
#include "catalog.h"
#include "clientInt.h"
#include "tname.h"
#include "cJSON.h"
#include "tglobal.h"
#include "osSemaphore.h"
#include "osThread.h"

//=================================================================================================

#define SPACE ' '
#define COMMA ','
#define EQUAL '='
#define QUOTE '"'
#define SLASH '\\'

#define JUMP_SPACE(sql) while (*sql != '\0'){if(*sql == SPACE) sql++;else break;}
// comma ,
#define IS_SLASH_COMMA(sql) (*(sql) == COMMA && *((sql) - 1) == SLASH)
#define IS_COMMA(sql) (*(sql) == COMMA && *((sql) - 1) != SLASH)
// space
#define IS_SLASH_SPACE(sql) (*(sql) == SPACE && *((sql) - 1) == SLASH)
#define IS_SPACE(sql) (*(sql) == SPACE && *((sql) - 1) != SLASH)
// equal =
#define IS_SLASH_EQUAL(sql) (*(sql) == EQUAL && *((sql) - 1) == SLASH)
#define IS_EQUAL(sql) (*(sql) == EQUAL && *((sql) - 1) != SLASH)
// quote "
#define IS_SLASH_QUOTE(sql) (*(sql) == QUOTE && *((sql) - 1) == SLASH)
#define IS_QUOTE(sql) (*(sql) == QUOTE && *((sql) - 1) != SLASH)
// SLASH
#define IS_SLASH_SLASH(sql) (*(sql) == SLASH && *((sql) - 1) == SLASH)

#define IS_SLASH_LETTER(sql) (IS_SLASH_COMMA(sql) || IS_SLASH_SPACE(sql) || IS_SLASH_EQUAL(sql) || IS_SLASH_QUOTE(sql) || IS_SLASH_SLASH(sql))

#define MOVE_FORWARD_ONE(sql,len) (memmove((void*)((sql) - 1), (sql), len))

#define PROCESS_SLASH(key,keyLen)       \
for (int i = 1; i < keyLen; ++i) {      \
  if(IS_SLASH_LETTER(key+i)){           \
    MOVE_FORWARD_ONE(key+i, keyLen-i);  \
    i--;                                \
    keyLen--;                           \
  }                                     \
}

#define IS_INVALID_COL_LEN(len)   ((len) <= 0 || (len) >= TSDB_COL_NAME_LEN)
#define IS_INVALID_TABLE_LEN(len) ((len) <= 0 || (len) >= TSDB_TABLE_NAME_LEN)

#define OTD_JSON_SUB_FIELDS_NUM 2
#define OTD_JSON_FIELDS_NUM     4

#define TS              "_ts"
#define TS_LEN          3
#define VALUE           "_value"
#define VALUE_LEN       6

#define BINARY_ADD_LEN 2        // "binary"   2 means " "
#define NCHAR_ADD_LEN 3         // L"nchar"   3 means L" "

#define MAX_RETRY_TIMES 5
#define LINE_BATCH  20000
//=================================================================================================
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;

typedef enum {
  SCHEMA_ACTION_CREATE_STABLE,
  SCHEMA_ACTION_ADD_COLUMN,
  SCHEMA_ACTION_ADD_TAG,
  SCHEMA_ACTION_CHANGE_COLUMN_SIZE,
  SCHEMA_ACTION_CHANGE_TAG_SIZE,
} ESchemaAction;

typedef struct {
  char          sTableName[TSDB_TABLE_NAME_LEN];
  SArray        *tags;
  SArray        *fields;
} SCreateSTableActionInfo;

typedef struct {
  char          sTableName[TSDB_TABLE_NAME_LEN];
  SSmlKv        *field;
} SAlterSTableActionInfo;

typedef struct {
  ESchemaAction             action;
  union {
    SCreateSTableActionInfo createSTable;
    SAlterSTableActionInfo  alterSTable;
  };
} SSchemaAction;

typedef struct {
  const char  *measure;
  const char  *tags;
  const char  *cols;
  const char  *timestamp;

  int32_t measureLen;
  int32_t measureTagsLen;
  int32_t tagsLen;
  int32_t colsLen;
  int32_t timestampLen;
} SSmlLineInfo;

typedef struct {
  const char     *sTableName;   // super table name
  int32_t        sTableNameLen;
  char           childTableName[TSDB_TABLE_NAME_LEN];
  uint64_t       uid;

  SArray         *tags;

  // if info->formatData is true, elements are SArray<SSmlKv*>.
  // if info->formatData is false, elements are SHashObj<cols key string, SSmlKv*> for find by key quickly
  SArray         *cols;
} SSmlTableInfo;

typedef struct {
  SArray     *tags;       // save the origin order to create table
  SHashObj   *tagHash;    // elements are <key, index in tags>

  SArray     *cols;
  SHashObj   *colHash;

  STableMeta *tableMeta;
} SSmlSTableMeta;

typedef struct {
  int32_t   len;
  char      *buf;
} SSmlMsgBuf;

typedef struct {
  int32_t code;
  int32_t lineNum;

  int32_t numOfSTables;
  int32_t numOfCTables;
  int32_t numOfCreateSTables;

  int64_t parseTime;
  int64_t schemaTime;
  int64_t insertBindTime;
  int64_t insertRpcTime;
  int64_t endTime;
} SSmlCostInfo;

typedef struct{
  SRequestObj*    request;
  tsem_t          sem;
  TdThreadSpinlock lock;
} Params;

typedef struct {
  int64_t           id;
  Params            *params;
  bool              isLast;

  SMLProtocolType   protocol;
  int8_t            precision;
  bool              dataFormat;     // true means that the name and order of keys in each line are the same(only for influx protocol)

  SHashObj          *childTables;
  SHashObj          *superTables;
  SHashObj          *pVgHash;
  void              *exec;

  STscObj           *taos;
  SCatalog          *pCatalog;
  SRequestObj       *pRequest;
  SQuery            *pQuery;

  SSmlCostInfo      cost;
  int32_t           affectedRows;
  SSmlMsgBuf        msgBuf;
  SHashObj          *dumplicateKey;  // for dumplicate key
  SArray            *colsContainer;  // for cols parse, if dataFormat == false
} SSmlHandle;
//=================================================================================================

//=================================================================================================
static volatile int64_t linesSmlHandleId = 0;
static int64_t smlGenId() {
  int64_t id;

  do {
    id = atomic_add_fetch_64(&linesSmlHandleId, 1);
  } while (id == 0);

  return id;
}

static inline bool smlDoubleToInt64OverFlow(double num) {
  if(num >= (double)INT64_MAX || num <= (double)INT64_MIN) return true;
  return false;
}

static inline bool smlCheckDuplicateKey(const char *key, int32_t keyLen, SHashObj *pHash) {
  void *val = taosHashGet(pHash, key, keyLen);
  if (val) {
    return true;
  }
  taosHashPut(pHash, key, keyLen, key, 1);
  return false;
}

static int32_t smlBuildInvalidDataMsg(SSmlMsgBuf* pBuf, const char *msg1, const char *msg2) {
  memset(pBuf->buf, 0 , pBuf->len);
  if(msg1) strncat(pBuf->buf, msg1, pBuf->len);
  int32_t left = pBuf->len - strlen(pBuf->buf);
  if(left > 2 && msg2) {
    strncat(pBuf->buf, ":", left - 1);
    strncat(pBuf->buf, msg2, left - 2);
  }
  return TSDB_CODE_SML_INVALID_DATA;
}

static int32_t smlGenerateSchemaAction(SSchema* colField, SHashObj* colHash, SSmlKv* kv, bool isTag,
                                       SSchemaAction* action, bool* actionNeeded, SSmlHandle* info) {
  uint16_t *index = (uint16_t *)taosHashGet(colHash, kv->key, kv->keyLen);
  if (index) {
    if (colField[*index].type != kv->type) {
      uError("SML:0x%"PRIx64" point type and db type mismatch. key: %s. point type: %d, db type: %d", info->id, kv->key,
             colField[*index].type, kv->type);
      return TSDB_CODE_TSC_INVALID_VALUE;
    }

    if ((colField[*index].type == TSDB_DATA_TYPE_VARCHAR && (colField[*index].bytes - VARSTR_HEADER_SIZE) < kv->length) ||
        (colField[*index].type == TSDB_DATA_TYPE_NCHAR &&((colField[*index].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE < kv->length))) {
      if (isTag) {
        action->action = SCHEMA_ACTION_CHANGE_TAG_SIZE;
      } else {
        action->action = SCHEMA_ACTION_CHANGE_COLUMN_SIZE;
      }
      action->alterSTable.field = kv;
      *actionNeeded = true;
    }
  } else {
    if (isTag) {
      action->action = SCHEMA_ACTION_ADD_TAG;
    } else {
      action->action = SCHEMA_ACTION_ADD_COLUMN;
    }
    action->alterSTable.field = kv;
    *actionNeeded = true;
  }
  if (*actionNeeded) {
    uDebug("SML:0x%" PRIx64 " generate schema action. column name: %s, action: %d", info->id, colField->name,
             action->action);
  }
  return 0;
}

static int32_t smlFindNearestPowerOf2(int32_t length){
  int32_t result = 1;
  while(result <= length){
    result *= 2;
  }
  return result;
}

static int32_t smlBuildColumnDescription(SSmlKv* field, char* buf, int32_t bufSize, int32_t* outBytes) {
  uint8_t type = field->type;
  char    tname[TSDB_TABLE_NAME_LEN] = {0};
  memcpy(tname, field->key, field->keyLen);
  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
    int32_t bytes = smlFindNearestPowerOf2(field->length);
    int out = snprintf(buf, bufSize, "`%s` %s(%d)",
                       tname, tDataTypes[field->type].name, bytes);
    *outBytes = out;
  } else {
    int out = snprintf(buf, bufSize, "`%s` %s", tname, tDataTypes[type].name);
    *outBytes = out;
  }

  return 0;
}

static int32_t smlApplySchemaAction(SSmlHandle* info, SSchemaAction* action) {
  int32_t code = 0;
  int32_t outBytes = 0;
  char *result = (char *)taosMemoryCalloc(1, TSDB_MAX_ALLOWED_SQL_LEN);
  int32_t capacity = TSDB_MAX_ALLOWED_SQL_LEN;

  uDebug("SML:0x%"PRIx64" apply schema action. action: %d", info->id, action->action);
  switch (action->action) {
    case SCHEMA_ACTION_ADD_COLUMN: {
      int n = sprintf(result, "alter stable `%s` add column ", action->alterSTable.sTableName);
      smlBuildColumnDescription(action->alterSTable.field, result+n, capacity-n, &outBytes);
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
      code = taos_errno(res);
      const char* errStr = taos_errstr(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error: %s", info->id, errStr);
        taosMsleep(100);
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_ADD_TAG: {
      int n = sprintf(result, "alter stable `%s` add tag ", action->alterSTable.sTableName);
      smlBuildColumnDescription(action->alterSTable.field,
                             result+n, capacity-n, &outBytes);
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
      code = taos_errno(res);
      const char* errStr = taos_errstr(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
        taosMsleep(100);
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_CHANGE_COLUMN_SIZE: {
      int n = sprintf(result, "alter stable `%s` modify column ", action->alterSTable.sTableName);
      smlBuildColumnDescription(action->alterSTable.field, result+n,
                             capacity-n, &outBytes);
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
        taosMsleep(100);
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_CHANGE_TAG_SIZE: {
      int n = sprintf(result, "alter stable `%s` modify tag ", action->alterSTable.sTableName);
      smlBuildColumnDescription(action->alterSTable.field, result+n,
                             capacity-n, &outBytes);
      TAOS_RES* res = taos_query(info->taos, result); //TODO async doAsyncQuery
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
        taosMsleep(100);
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_CREATE_STABLE: {
      int n = sprintf(result, "create stable `%s` (", action->createSTable.sTableName);
      char* pos = result + n; int freeBytes = capacity - n;

      SArray *cols = action->createSTable.fields;

      for(int i = 0; i < taosArrayGetSize(cols); i++){
        SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
        smlBuildColumnDescription(kv, pos, freeBytes, &outBytes);
        pos += outBytes; freeBytes -= outBytes;
        *pos = ','; ++pos; --freeBytes;
      }

      --pos; ++freeBytes;

      outBytes = snprintf(pos, freeBytes, ") tags (");
      pos += outBytes; freeBytes -= outBytes;

      cols = action->createSTable.tags;
      for(int i = 0; i < taosArrayGetSize(cols); i++){
        SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
        smlBuildColumnDescription(kv, pos, freeBytes, &outBytes);
        pos += outBytes; freeBytes -= outBytes;
        *pos = ','; ++pos; --freeBytes;
      }
      if(taosArrayGetSize(cols) == 0){
        outBytes = snprintf(pos, freeBytes,"`%s` %s(%d)",
                            tsSmlTagName, tDataTypes[TSDB_DATA_TYPE_NCHAR].name, 1);
        pos += outBytes; freeBytes -= outBytes;
        *pos = ','; ++pos; --freeBytes;
      }
      pos--; ++freeBytes;
      outBytes = snprintf(pos, freeBytes, ")");
      TAOS_RES* res = taos_query(info->taos, result);
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" apply schema action. error : %s", info->id, taos_errstr(res));
        taosMsleep(100);
      }
      taos_free_result(res);

      break;
    }

    default:
      break;
  }

  taosMemoryFreeClear(result);
  if (code != 0) {
    uError("SML:0x%"PRIx64 " apply schema action failure. %s", info->id, tstrerror(code));
  }
  return code;
}

static int32_t smlProcessSchemaAction(SSmlHandle* info, SSchema* schemaField, SHashObj* schemaHash, SArray *cols, SSchemaAction* action, bool isTag){
  int32_t code = TSDB_CODE_SUCCESS;
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
    SSmlKv* kv = (SSmlKv*)taosArrayGetP(cols, j);
    bool actionNeeded = false;
    code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, action, &actionNeeded, info);
    if(code != TSDB_CODE_SUCCESS){
      return code;
    }
    if (actionNeeded) {
      code = smlApplySchemaAction(info, action);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t smlCheckMeta(SSchema* schema, int32_t length, SArray* cols){
  SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
  for(uint16_t i = 0; i < length; i++){
    taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
  }

  for(int32_t i = 0; i < taosArrayGetSize(cols); i++){
    SSmlKv* kv = (SSmlKv*)taosArrayGetP(cols, i);
    if(taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL){
      return -1;
    }
  }
  return 0;
}

static int32_t smlModifyDBSchemas(SSmlHandle* info) {
  int32_t code = 0;
  SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
  strcpy(pName.dbname, info->pRequest->pDb);

  SRequestConnInfo conn = {0};
  conn.pTrans = info->taos->pAppInfo->pTransporter;
  conn.requestId = info->pRequest->requestId;
  conn.requestObjRefId = info->pRequest->self;
  conn.mgmtEps = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
  
  SSmlSTableMeta** tableMetaSml = (SSmlSTableMeta**)taosHashIterate(info->superTables, NULL);
  while (tableMetaSml) {
    SSmlSTableMeta* sTableData = *tableMetaSml;
    STableMeta *pTableMeta = NULL;
    bool needCheckMeta = false;     // for multi thread

    size_t superTableLen = 0;
    void *superTable = taosHashGetKey(tableMetaSml, &superTableLen);
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
    memcpy(pName.tname, superTable, superTableLen);

    code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);

    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_INVALID_STB) {
      SSchemaAction schemaAction;
      schemaAction.action = SCHEMA_ACTION_CREATE_STABLE;
      memset(&schemaAction.createSTable, 0, sizeof(SCreateSTableActionInfo));
      memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
      schemaAction.createSTable.tags = sTableData->tags;
      schemaAction.createSTable.fields = sTableData->cols;
      code = smlApplySchemaAction(info, &schemaAction);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" smlApplySchemaAction failed. can not create %s", info->id, schemaAction.createSTable.sTableName);
        goto end;
      }
      info->cost.numOfCreateSTables++;
    }else if (code == TSDB_CODE_SUCCESS) {
      SHashObj *hashTmp = taosHashInit(pTableMeta->tableInfo.numOfTags, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
      for(uint16_t i = pTableMeta->tableInfo.numOfColumns; i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++){
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }

      SSchemaAction schemaAction;
      memset(&schemaAction, 0, sizeof(SSchemaAction));
      memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &schemaAction, true);
      if (code != TSDB_CODE_SUCCESS) {
        taosHashCleanup(hashTmp);
        goto end;
      }

      taosHashClear(hashTmp);
      for(uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns; i++){
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->cols, &schemaAction, false);
      taosHashCleanup(hashTmp);
      if (code != TSDB_CODE_SUCCESS) {
        goto end;
      }

      code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
      if (code != TSDB_CODE_SUCCESS) {
        goto end;
      }
      needCheckMeta = true;
    } else {
      uError("SML:0x%"PRIx64" load table meta error: %s", info->id, tstrerror(code));
      goto end;
    }
    if(pTableMeta) taosMemoryFree(pTableMeta);

    code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%"PRIx64" catalogGetSTableMeta failed. super table name %s", info->id, (char*)superTable);
      goto end;
    }

    if(needCheckMeta){
      code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags, sTableData->tags);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" check tag failed. super table name %s", info->id, (char*)superTable);
        goto end;
      }
      code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%"PRIx64" check cols failed. super table name %s", info->id, (char*)superTable);
        goto end;
      }
    }

    sTableData->tableMeta = pTableMeta;

    tableMetaSml = (SSmlSTableMeta**)taosHashIterate(info->superTables, tableMetaSml);
  }
  return 0;

end:
  catalogRefreshTableMeta(info->pCatalog, &conn, &pName, 1);
  return code;
}

static bool smlParseNumber(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 == 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;
}

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 = true;
    return true;
  }

  if ((len == 1) && (pVal[0] == 'f' || pVal[0] == 'F')) {
    kvVal->i = false;
    return true;
  }

  if((len == 4) && !strncasecmp(pVal, "true", len)) {
    kvVal->i = true;
    return true;
  }
  if((len == 5) && !strncasecmp(pVal, "false", len)) {
    kvVal->i = 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[0] == 'l' || pVal[0] == 'L')&& pVal[1] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

static int64_t smlGetTimeValue(const char *value, int32_t len, int8_t type) {
  char *endPtr = NULL;
  int64_t tsInt64 = taosStr2Int64(value, &endPtr, 10);
  if(value + len != endPtr){
    return -1;
  }
  double ts = tsInt64;
  switch (type) {
    case TSDB_TIME_PRECISION_HOURS:
      ts *= NANOSECOND_PER_HOUR;
      tsInt64 *= NANOSECOND_PER_HOUR;
      break;
    case TSDB_TIME_PRECISION_MINUTES:
      ts *= NANOSECOND_PER_MINUTE;
      tsInt64 *= NANOSECOND_PER_MINUTE;
      break;
    case TSDB_TIME_PRECISION_SECONDS:
      ts *= NANOSECOND_PER_SEC;
      tsInt64 *= NANOSECOND_PER_SEC;
      break;
    case TSDB_TIME_PRECISION_MILLI:
      ts *= NANOSECOND_PER_MSEC;
      tsInt64 *= NANOSECOND_PER_MSEC;
      break;
    case TSDB_TIME_PRECISION_MICRO:
      ts *= NANOSECOND_PER_USEC;
      tsInt64 *= NANOSECOND_PER_USEC;
      break;
    case TSDB_TIME_PRECISION_NANO:
      break;
    default:
      ASSERT(0);
  }
  if(ts >= (double)INT64_MAX || ts < 0){
    return -1;
  }

  return tsInt64;
}

static int8_t smlGetTsTypeByLen(int32_t len) {
  if (len == TSDB_TIME_PRECISION_SEC_DIGITS) {
    return TSDB_TIME_PRECISION_SECONDS;
  } else if (len == TSDB_TIME_PRECISION_MILLI_DIGITS) {
    return TSDB_TIME_PRECISION_MILLI;
  } else {
    return -1;
  }
}

static int8_t smlGetTsTypeByPrecision(int8_t precision) {
  switch (precision) {
    case TSDB_SML_TIMESTAMP_HOURS:
      return TSDB_TIME_PRECISION_HOURS;
    case TSDB_SML_TIMESTAMP_MILLI_SECONDS:
      return TSDB_TIME_PRECISION_MILLI;
    case TSDB_SML_TIMESTAMP_NANO_SECONDS:
    case TSDB_SML_TIMESTAMP_NOT_CONFIGURED:
      return TSDB_TIME_PRECISION_NANO;
    case TSDB_SML_TIMESTAMP_MICRO_SECONDS:
      return TSDB_TIME_PRECISION_MICRO;
    case TSDB_SML_TIMESTAMP_SECONDS:
      return TSDB_TIME_PRECISION_SECONDS;
    case TSDB_SML_TIMESTAMP_MINUTES:
      return TSDB_TIME_PRECISION_MINUTES;
    default:
      return -1;
  }
}

static int64_t smlParseInfluxTime(SSmlHandle* info, const char* data, int32_t len){
  if(len == 0 || (len == 1 && data[0] == '0')){
    return taosGetTimestampNs();
  }

  int8_t tsType = smlGetTsTypeByPrecision(info->precision);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp precision", NULL);
    return -1;
  }

  int64_t ts = smlGetTimeValue(data, len, tsType);
  if(ts == -1){
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

static int64_t smlParseOpenTsdbTime(SSmlHandle* info, const char* data, int32_t len){
  if(!data){
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
    return -1;
  }
  if(len == 1 && data[0] == '0'){
    return taosGetTimestampNs();
  }
  int8_t tsType = smlGetTsTypeByLen(len);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
    return -1;
  }
  int64_t ts = smlGetTimeValue(data, len, tsType);
  if(ts == -1){
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

static int32_t smlParseTS(SSmlHandle* info, const char* data, int32_t len, SArray *cols){
  int64_t ts = 0;
  if(info->protocol == TSDB_SML_LINE_PROTOCOL){
    ts = smlParseInfluxTime(info, data, len);
  }else if(info->protocol == TSDB_SML_TELNET_PROTOCOL){
    ts = smlParseOpenTsdbTime(info, data, len);
  }else{
    ASSERT(0);
  }

  if(ts == -1)  return TSDB_CODE_TSC_INVALID_TIME_STAMP;

  // add ts to
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
  if(!kv){
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  kv->key = TS;
  kv->keyLen = TS_LEN;
  kv->i = ts;
  kv->type = TSDB_DATA_TYPE_TIMESTAMP;
  kv->length = (int16_t)tDataTypes[kv->type].bytes;
  if(cols) taosArrayPush(cols, &kv);
  return TSDB_CODE_SUCCESS;
}

static bool smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
  //binary
  if (smlIsBinary(pVal->value, pVal->length)) {
    pVal->type = TSDB_DATA_TYPE_BINARY;
    pVal->length -= BINARY_ADD_LEN;
    pVal->value += (BINARY_ADD_LEN - 1);
    return true;
  }
  //nchar
  if (smlIsNchar(pVal->value, pVal->length)) {
    pVal->type = TSDB_DATA_TYPE_NCHAR;
    pVal->length -= NCHAR_ADD_LEN;
    pVal->value += (NCHAR_ADD_LEN - 1);
    return true;
  }

  //bool
  if (smlParseBool(pVal)) {
    pVal->type = TSDB_DATA_TYPE_BOOL;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }
  //number
  if (smlParseNumber(pVal, msg)) {
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    return true;
  }

  return false;
}

static int32_t smlParseInfluxString(const char* sql, SSmlLineInfo *elements, SSmlMsgBuf *msg){
  if(!sql) return TSDB_CODE_SML_INVALID_DATA;
  JUMP_SPACE(sql)
  if(*sql == COMMA) return TSDB_CODE_SML_INVALID_DATA;
  elements->measure = sql;

  // parse measure
  while (*sql != '\0') {
    if((sql != elements->measure) && IS_SLASH_LETTER(sql)){
      MOVE_FORWARD_ONE(sql,strlen(sql) + 1);
      continue;
    }
    if(IS_COMMA(sql)){
      break;
    }

    if(IS_SPACE(sql)){
      break;
    }
    sql++;
  }
  elements->measureLen = sql - elements->measure;
  if(IS_INVALID_TABLE_LEN(elements->measureLen)) {
    smlBuildInvalidDataMsg(msg, "measure is empty or too large than 192", NULL);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  // parse tag
  if(*sql == SPACE){
    elements->tagsLen = 0;
  }else{
    if(*sql == COMMA) sql++;
    elements->tags = sql;
    while (*sql != '\0') {
      if(IS_SPACE(sql)){
        break;
      }
      sql++;
    }
    elements->tagsLen = sql - elements->tags;
  }
  elements->measureTagsLen = sql - elements->measure;

  // parse cols
  JUMP_SPACE(sql)
  elements->cols = sql;
  bool isInQuote = false;
  while (*sql != '\0') {
    if(IS_QUOTE(sql)){
      isInQuote = !isInQuote;
    }
    if(!isInQuote && IS_SPACE(sql)){
      break;
    }
    sql++;
  }
  if(isInQuote){
    smlBuildInvalidDataMsg(msg, "only one quote", elements->cols);
    return TSDB_CODE_SML_INVALID_DATA;
  }
  elements->colsLen = sql - elements->cols;
  if(elements->colsLen == 0) {
    smlBuildInvalidDataMsg(msg, "cols is empty", NULL);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  // parse timestamp
  JUMP_SPACE(sql)
  elements->timestamp = sql;
  while (*sql != '\0') {
    if(*sql == SPACE){
      break;
    }
    sql++;
  }
  elements->timestampLen = sql - elements->timestamp;

  return TSDB_CODE_SUCCESS;
}

static void smlParseTelnetElement(const char **sql, const char **data, int32_t *len){
  while (**sql != '\0') {
    if(**sql != SPACE && !(*data)) {
      *data = *sql;
    }else if (**sql == SPACE && *data) {
      *len = *sql - *data;
      break;
    }
    (*sql)++;
  }
}

static int32_t smlParseTelnetTags(const char* data, SArray *cols, char *childTableName, SHashObj *dumplicateKey, SSmlMsgBuf *msg){
  const char *sql = data;
  size_t childTableNameLen = strlen(tsSmlChildTableName);
  while(*sql != '\0'){
    JUMP_SPACE(sql)
    if(*sql == '\0') break;

    const char *key = sql;
    int32_t keyLen = 0;

    // parse key
    while(*sql != '\0'){
      if(*sql == SPACE) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      if(*sql == EQUAL) {
        keyLen = sql - key;
        sql++;
        break;
      }
      sql++;
    }

    if(IS_INVALID_COL_LEN(keyLen)){
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
      return TSDB_CODE_SML_INVALID_DATA;
    }
    if(smlCheckDuplicateKey(key, keyLen, dumplicateKey)){
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
      return TSDB_CODE_TSC_DUP_TAG_NAMES;
    }

    // parse value
    const char *value = sql;
    int32_t valueLen = 0;
    while(*sql != '\0') {
      // parse value
      if (*sql == SPACE) {
        break;
      }
      if (*sql == EQUAL) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
    }
    valueLen = sql - value;

    if(valueLen == 0){
      smlBuildInvalidDataMsg(msg, "invalid value", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }

    //handle child table name
    if(childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0){
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, value, (valueLen < TSDB_TABLE_NAME_LEN ? valueLen : TSDB_TABLE_NAME_LEN));
      continue;
    }

    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
    if(!kv) return TSDB_CODE_OUT_OF_MEMORY;
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
    kv->length = valueLen;
    kv->type = TSDB_DATA_TYPE_NCHAR;

    if(cols) taosArrayPush(cols, &kv);
  }

  return TSDB_CODE_SUCCESS;
}

// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
static int32_t smlParseTelnetString(SSmlHandle *info, const char* sql, SSmlTableInfo *tinfo, SArray *cols){
  if(!sql) return TSDB_CODE_SML_INVALID_DATA;

  // parse metric
  smlParseTelnetElement(&sql, &tinfo->sTableName, &tinfo->sTableNameLen);
  if (!(tinfo->sTableName) || IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  // parse timestamp
  const char *timestamp = NULL;
  int32_t tLen = 0;
  smlParseTelnetElement(&sql, &timestamp, &tLen);
  if (!timestamp || tLen == 0) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  int32_t ret = smlParseTS(info, timestamp, tLen, cols);
  if (ret != TSDB_CODE_SUCCESS) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  // parse value
  const char *value = NULL;
  int32_t valueLen = 0;
  smlParseTelnetElement(&sql, &value, &valueLen);
  if (!value || valueLen == 0) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
  if(!kv) return TSDB_CODE_OUT_OF_MEMORY;
  taosArrayPush(cols, &kv);
  kv->key = VALUE;
  kv->keyLen = VALUE_LEN;
  kv->value = value;
  kv->length = valueLen;
  if(!smlParseValue(kv, &info->msgBuf)){
    return TSDB_CODE_SML_INVALID_DATA;
  }

  // parse tags
  ret = smlParseTelnetTags(sql, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
  if (ret != TSDB_CODE_SUCCESS) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseCols(const char* data, int32_t len, SArray *cols, char *childTableName, bool isTag, SHashObj *dumplicateKey, SSmlMsgBuf *msg){
  if(len == 0){
    return TSDB_CODE_SUCCESS;
  }

  size_t childTableNameLen = strlen(tsSmlChildTableName);
  const char *sql = data;
  while(sql < data + len){
    const char *key = sql;
    int32_t keyLen = 0;

    while(sql < data + len){
      // parse key
      if(IS_COMMA(sql)) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      if(IS_EQUAL(sql)) {
        keyLen = sql - key;
        sql++;
        break;
      }
      sql++;
    }

    if(IS_INVALID_COL_LEN(keyLen)){
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
      return TSDB_CODE_SML_INVALID_DATA;
    }
    if(smlCheckDuplicateKey(key, keyLen, dumplicateKey)){
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
      return TSDB_CODE_TSC_DUP_TAG_NAMES;
    }

    // parse value
    const char *value = sql;
    int32_t valueLen = 0;
    bool isInQuote = false;
    while(sql < data + len) {
      // parse value
      if(!isTag && IS_QUOTE(sql)){
        isInQuote = !isInQuote;
        sql++;
        continue;
      }
      if (!isInQuote && IS_COMMA(sql)) {
        break;
      }
      if (!isInQuote && IS_EQUAL(sql)) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
    }
    valueLen = sql - value;
    sql++;

    if(isInQuote){
      smlBuildInvalidDataMsg(msg, "only one quote", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
    if(valueLen == 0){
      smlBuildInvalidDataMsg(msg, "invalid value", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
    PROCESS_SLASH(key, keyLen)
    PROCESS_SLASH(value, valueLen)

    //handle child table name
    if(childTableName && childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0){
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, value, (valueLen < TSDB_TABLE_NAME_LEN ? valueLen : TSDB_TABLE_NAME_LEN));
      continue;
    }

    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
    if(!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if(cols) taosArrayPush(cols, &kv);

    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
    kv->length = valueLen;
    if(isTag){
      kv->type = TSDB_DATA_TYPE_NCHAR;
    }else{
      if(!smlParseValue(kv, msg)){
        return TSDB_CODE_SML_INVALID_DATA;
      }
    }
  }

  return TSDB_CODE_SUCCESS;
}

static bool smlUpdateMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols, SSmlMsgBuf *msg){
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {  //jump timestamp
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);

    int16_t *index = (int16_t *)taosHashGet(metaHash, kv->key, kv->keyLen);
    if(index){
      SSmlKv **value = (SSmlKv **)taosArrayGet(metaArray, *index);
      if(kv->type != (*value)->type){
        smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
        return false;
      }else{
        if(IS_VAR_DATA_TYPE(kv->type)){     // update string len, if bigger
          if(kv->length > (*value)->length){
            *value = kv;
          }
        }
      }
    }else{
      size_t tmp = taosArrayGetSize(metaArray);
      ASSERT(tmp <= INT16_MAX);
      int16_t size = tmp;
      taosArrayPush(metaArray, &kv);
      taosHashPut(metaHash, kv->key, kv->keyLen, &size, SHORT_BYTES);
    }
  }

  return true;
}

static void smlInsertMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols){
  for (int16_t i = 0; i < taosArrayGetSize(cols); ++i) {
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
    taosArrayPush(metaArray, &kv);
    taosHashPut(metaHash, kv->key, kv->keyLen, &i, SHORT_BYTES);
  }
}

static SSmlTableInfo* smlBuildTableInfo(){
  SSmlTableInfo *tag = (SSmlTableInfo *)taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
  if(!tag){
    return NULL;
  }

  tag->cols = taosArrayInit(16, POINTER_BYTES);
  if (tag->cols == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
  }

  tag->tags = taosArrayInit(16, POINTER_BYTES);
  if (tag->tags == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
  }
  return tag;

cleanup:
  taosMemoryFree(tag);
  return NULL;
}

static void smlDestroyTableInfo(SSmlHandle* info, SSmlTableInfo *tag){
  if(info->dataFormat){
    for(size_t i = 0; i < taosArrayGetSize(tag->cols); i++){
      SArray *kvArray = (SArray *)taosArrayGetP(tag->cols, i);
      for (int j = 0; j < taosArrayGetSize(kvArray); ++j) {
        SSmlKv *p = (SSmlKv *)taosArrayGetP(kvArray, j);
        if(info->protocol == TSDB_SML_JSON_PROTOCOL &&
            (p->type == TSDB_DATA_TYPE_NCHAR || p->type == TSDB_DATA_TYPE_BINARY)){
          taosMemoryFree((void*)p->value);
        }
        taosMemoryFree(p);
      }
      taosArrayDestroy(kvArray);
    }
  }else{
    for(size_t i = 0; i < taosArrayGetSize(tag->cols); i++){
      SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
      void** p1 = (void**)taosHashIterate(kvHash, NULL);
      while (p1) {
        taosMemoryFree(*p1);
        p1 = (void**)taosHashIterate(kvHash, p1);
      }
      taosHashCleanup(kvHash);
    }
  }
  for(size_t i = 0; i < taosArrayGetSize(tag->tags); i++){
    SSmlKv *p = (SSmlKv *)taosArrayGetP(tag->tags, i);
    if(info->protocol == TSDB_SML_JSON_PROTOCOL){
      taosMemoryFree((void*)p->key);
      if(p->type == TSDB_DATA_TYPE_NCHAR || p->type == TSDB_DATA_TYPE_BINARY){
        taosMemoryFree((void*)p->value);
      }
    }
    taosMemoryFree(p);
  }
  if(info->protocol == TSDB_SML_JSON_PROTOCOL && tag->sTableName){
    taosMemoryFree((void*)tag->sTableName);
  }
  taosArrayDestroy(tag->cols);
  taosArrayDestroy(tag->tags);
  taosMemoryFree(tag);
}

static int32_t smlKvTimeArrayCompare(const void* key1, const void* key2) {
  SArray *s1 = *(SArray **)key1;
  SArray *s2 = *(SArray **)key2;
  SSmlKv *kv1 = (SSmlKv *)taosArrayGetP(s1, 0);
  SSmlKv *kv2 = (SSmlKv *)taosArrayGetP(s2, 0);
  ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP);
  ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP);
  if (kv1->i < kv2->i) {
    return -1;
  } else if (kv1->i > kv2->i) {
    return 1;
  } else {
    return 0;
  }
}

static int32_t smlKvTimeHashCompare(const void* key1, const void* key2) {
  SHashObj *s1 = *(SHashObj **)key1;
  SHashObj *s2 = *(SHashObj **)key2;
  SSmlKv *kv1 = (SSmlKv *)taosHashGet(s1, TS, TS_LEN);
  SSmlKv *kv2 = (SSmlKv *)taosHashGet(s2, TS, TS_LEN);
  ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP);
  ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP);
  if (kv1->i < kv2->i) {
    return -1;
  } else if (kv1->i > kv2->i) {
    return 1;
  } else {
    return 0;
  }
}

static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
  if(dataFormat){
    void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GE);
    if(p == NULL){
      taosArrayPush(oneTable->cols, &cols);
    }else{      // to make the sort stable for update data
      SArray *sa = (SArray *)p;
      SSmlKv *cur = (SSmlKv*)taosArrayGet(sa, 0);
      SSmlKv *dCur = (SSmlKv*)taosArrayGet(cols, 0);
      if(cur->i > dCur->i){
        taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
      }else{
        ASSERT(cur->i == dCur->i);
        int32_t index = TARRAY_ELEM_IDX(oneTable->cols, p) + 1;
        for(; index < taosArrayGetSize(oneTable->cols); index++){
          SArray *tmp = (SArray *)taosArrayGet(oneTable->cols, index);
          SSmlKv *curTs = (SSmlKv*)taosArrayGet(tmp, 0);
          if(curTs->i > dCur->i){
            break;
          }
        }
        taosArrayInsert(oneTable->cols, index, &cols);
      }
    }
    return TSDB_CODE_SUCCESS;
  }

  SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if(!kvHash){
    uError("SML:smlDealCols failed to allocate memory");
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
  for(size_t i = 0; i < taosArrayGetSize(cols); i++){
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
    taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
  }

  void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GE);
  if(p == NULL){
    taosArrayPush(oneTable->cols, &kvHash);
  }else{    // to make the sort stable for update data
    SHashObj *sa = (SHashObj *)p;
    SSmlKv *cur = (SSmlKv *)taosHashGet(sa, TS, TS_LEN);
    SSmlKv *dCur = (SSmlKv*)taosArrayGet(cols, 0);
    if(cur->i > dCur->i){
      taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
    }else{
      ASSERT(cur->i == dCur->i);
      int32_t index = TARRAY_ELEM_IDX(oneTable->cols, p) + 1;
      for(; index < taosArrayGetSize(oneTable->cols); index++){
        SHashObj *tmp = (SHashObj *)taosArrayGet(oneTable->cols, index);
        SSmlKv *curTs = (SSmlKv *)taosHashGet(tmp, TS, TS_LEN);
        if(curTs->i > dCur->i){
          break;
        }
      }
      taosArrayInsert(oneTable->cols, index, &cols);
    }
  }
  return TSDB_CODE_SUCCESS;
}

static SSmlSTableMeta* smlBuildSTableMeta(){
  SSmlSTableMeta* meta = (SSmlSTableMeta*)taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if(!meta){
    return NULL;
  }
  meta->tagHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (meta->tagHash == NULL) {
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }

  meta->colHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (meta->colHash == NULL) {
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }

  meta->tags = taosArrayInit(32, POINTER_BYTES);
  if (meta->tags == NULL) {
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }

  meta->cols = taosArrayInit(32, POINTER_BYTES);
  if (meta->cols == NULL) {
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }
  return meta;

cleanup:
  taosMemoryFree(meta);
  return NULL;
}

static void smlDestroySTableMeta(SSmlSTableMeta *meta){
  taosHashCleanup(meta->tagHash);
  taosHashCleanup(meta->colHash);
  taosArrayDestroy(meta->tags);
  taosArrayDestroy(meta->cols);
  taosMemoryFree(meta->tableMeta);
  taosMemoryFree(meta);
}

static void smlDestroyCols(SArray *cols) {
  if (!cols) return;
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
    void *kv = taosArrayGetP(cols, i);
    taosMemoryFree(kv);
  }
}

static void smlDestroyInfo(SSmlHandle* info){
  if(!info) return;
  qDestroyQuery(info->pQuery);
  smlDestroyHandle(info->exec);

  // destroy info->childTables
  void** p1 = (void**)taosHashIterate(info->childTables, NULL);
  while (p1) {
    smlDestroyTableInfo(info, (SSmlTableInfo*)(*p1));
    p1 = (void**)taosHashIterate(info->childTables, p1);
  }
  taosHashCleanup(info->childTables);

  // destroy info->superTables
  p1 = (void**)taosHashIterate(info->superTables, NULL);
  while (p1) {
    smlDestroySTableMeta((SSmlSTableMeta*)(*p1));
    p1 = (void**)taosHashIterate(info->superTables, p1);
  }
  taosHashCleanup(info->superTables);

  // destroy info->pVgHash
  taosHashCleanup(info->pVgHash);
  taosHashCleanup(info->dumplicateKey);
  if(!info->dataFormat){
    taosArrayDestroy(info->colsContainer);
  }
  destroyRequest(info->pRequest);
  taosMemoryFreeClear(info);
}

static SSmlHandle* smlBuildSmlInfo(TAOS* taos, SRequestObj* request, SMLProtocolType protocol, int8_t precision){
  int32_t code = TSDB_CODE_SUCCESS;
  SSmlHandle* info = (SSmlHandle*)taosMemoryCalloc(1, sizeof(SSmlHandle));
  if (NULL == info) {
    return NULL;
  }
  info->id          = smlGenId();

  info->pQuery      = (SQuery *)taosMemoryCalloc(1, sizeof(SQuery));
  if (NULL == info->pQuery) {
    uError("SML:0x%"PRIx64" create info->pQuery error", info->id);
    goto cleanup;
  }
  info->pQuery->execMode      = QUERY_EXEC_MODE_SCHEDULE;
  info->pQuery->haveResultSet = false;
  info->pQuery->msgType       = TDMT_VND_SUBMIT;
  info->pQuery->pRoot         = (SNode*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT);
  if(NULL == info->pQuery->pRoot){
    uError("SML:0x%"PRIx64" create info->pQuery->pRoot error", info->id);
    goto cleanup;
  }
  ((SVnodeModifOpStmt*)(info->pQuery->pRoot))->payloadType = PAYLOAD_TYPE_KV;

  info->taos        = (STscObj *)taos;
  code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
  if(code != TSDB_CODE_SUCCESS){
    uError("SML:0x%"PRIx64" get catalog error %d", info->id, code);
    goto cleanup;
  }

  info->precision   = precision;
  info->protocol    = protocol;
  if(protocol == TSDB_SML_LINE_PROTOCOL){
    info->dataFormat = tsSmlDataFormat;
  }else{
    info->dataFormat = true;
  }
  info->pRequest    = request;
  info->msgBuf.buf  = info->pRequest->msgBuf;
  info->msgBuf.len  = ERROR_MSG_BUF_DEFAULT_SIZE;

  info->exec        = smlInitHandle(info->pQuery);
  info->childTables = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  info->superTables = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  info->pVgHash     = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);

  info->dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if(!info->dataFormat){
    info->colsContainer = taosArrayInit(32, POINTER_BYTES);
    if(NULL == info->colsContainer){
      uError("SML:0x%"PRIx64" create info failed", info->id);
      goto cleanup;
    }
  }
  if(NULL == info->exec || NULL == info->childTables
      || NULL == info->superTables || NULL == info->pVgHash
      || NULL == info->dumplicateKey){
    uError("SML:0x%"PRIx64" create info failed", info->id);
    goto cleanup;
  }

  return info;
cleanup:
  smlDestroyInfo(info);
  return NULL;
}

/************* TSDB_SML_JSON_PROTOCOL function start **************/
static int32_t smlJsonCreateSring(const char **output, char *input, int32_t inputLen){
  *output = (const char *)taosMemoryMalloc(inputLen);
  if (*output == NULL){
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

  memcpy((void*)(*output), input, inputLen);
  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseMetricFromJSON(SSmlHandle *info, cJSON *root, SSmlTableInfo *tinfo) {
  cJSON *metric = cJSON_GetObjectItem(root, "metric");
  if (!cJSON_IsString(metric)) {
    return  TSDB_CODE_TSC_INVALID_JSON;
  }

  tinfo->sTableNameLen = strlen(metric->valuestring);
  if (IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
    uError("OTD:0x%"PRIx64" Metric lenght is 0 or large than 192", info->id);
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
  }

  return smlJsonCreateSring(&tinfo->sTableName, metric->valuestring, tinfo->sTableNameLen);
}

static int32_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int64_t *tsVal) {
  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 (!cJSON_IsNumber(value)) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

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

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

  size_t typeLen = strlen(type->valuestring);
  if (typeLen == 1 && (type->valuestring[0] == 's' || type->valuestring[0] == 'S')) {
    //seconds
    timeDouble = timeDouble * 1e9;
    if(smlDoubleToInt64OverFlow(timeDouble)){
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
      return TSDB_CODE_TSC_INVALID_TIME_STAMP;
    }
    *tsVal = timeDouble;
  } else if (typeLen == 2 && (type->valuestring[1] == 's' || type->valuestring[1] == 'S')) {
    switch (type->valuestring[0]) {
      case 'm':
      case 'M':
        //milliseconds
        timeDouble = timeDouble * 1e6;
        if(smlDoubleToInt64OverFlow(timeDouble)){
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
          return TSDB_CODE_TSC_INVALID_TIME_STAMP;
        }
        *tsVal = timeDouble;
        break;
      case 'u':
      case 'U':
        //microseconds
        timeDouble = timeDouble * 1e3;
        if(smlDoubleToInt64OverFlow(timeDouble)){
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
          return TSDB_CODE_TSC_INVALID_TIME_STAMP;
        }
        *tsVal = timeDouble;
        break;
      case 'n':
      case 'N':
        //nanoseconds
        *tsVal = timeDouble;
        break;
      default:
        return TSDB_CODE_TSC_INVALID_JSON;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  return TSDB_CODE_SUCCESS;
}

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

static int32_t smlParseTSFromJSON(SSmlHandle *info, cJSON *root, SArray *cols) {
  //Timestamp must be the first KV to parse
  int64_t tsVal = 0;

  cJSON *timestamp = cJSON_GetObjectItem(root, "timestamp");
  if (cJSON_IsNumber(timestamp)) {
    //timestamp value 0 indicates current system time
    double timeDouble = timestamp->valuedouble;
    if(smlDoubleToInt64OverFlow(timeDouble)){
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
      return TSDB_CODE_TSC_INVALID_TIME_STAMP;
    }

    if(timeDouble < 0){
      return TSDB_CODE_TSC_INVALID_TIME_STAMP;
    }
    uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);
    if (tsLen == TSDB_TIME_PRECISION_SEC_DIGITS) {
      timeDouble = timeDouble * 1e9;
      if(smlDoubleToInt64OverFlow(timeDouble)){
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
        return TSDB_CODE_TSC_INVALID_TIME_STAMP;
      }
      tsVal = timeDouble;
    } else if (tsLen == TSDB_TIME_PRECISION_MILLI_DIGITS) {
      timeDouble = timeDouble * 1e6;
      if(smlDoubleToInt64OverFlow(timeDouble)){
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
        return TSDB_CODE_TSC_INVALID_TIME_STAMP;
      }
      tsVal = timeDouble;
    } else if(timeDouble == 0){
      tsVal = taosGetTimestampNs();
    }else {
      return TSDB_CODE_TSC_INVALID_TIME_STAMP;
    }
  } else if (cJSON_IsObject(timestamp)) {
    int32_t ret = smlParseTSFromJSONObj(info, timestamp, &tsVal);
    if (ret != TSDB_CODE_SUCCESS) {
      uError("SML:0x%"PRIx64" Failed to parse timestamp from JSON Obj", info->id);
      return ret;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  // add ts to
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
  if(!kv){
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  kv->key = TS;
  kv->keyLen = TS_LEN;
  kv->i = tsVal;
  kv->type = TSDB_DATA_TYPE_TIMESTAMP;
  kv->length = (int16_t)tDataTypes[kv->type].bytes;
  if(cols) taosArrayPush(cols, &kv);
  return TSDB_CODE_SUCCESS;

}

static int32_t smlConvertJSONBool(SSmlKv *pVal, char* typeStr, cJSON *value) {
  if (strcasecmp(typeStr, "bool") != 0) {
    uError("OTD: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("OTD: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("OTD: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("OTD: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("OTD: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("OTD: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, "nchar") == 0) {
    pVal->type = TSDB_DATA_TYPE_NCHAR;
  } else {
    uError("OTD:invalid type(%s) for JSON String", typeStr);
    return TSDB_CODE_TSC_INVALID_JSON_TYPE;
  }
  pVal->length = (int16_t)strlen(value->valuestring);
  return smlJsonCreateSring(&pVal->value, value->valuestring, pVal->length);
}

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: {
      /* set default JSON type to binary/nchar according to
       * user configured parameter tsDefaultJSONStrType
       */

      char *tsDefaultJSONStrType = "nchar";   //todo
      smlConvertJSONString(kv, tsDefaultJSONStrType, root);
      break;
    }
    case cJSON_Object: {
      int32_t ret = smlParseValueFromJSONObj(root, kv);
      if (ret != TSDB_CODE_SUCCESS) {
        uError("OTD:Failed to parse value from JSON Obj");
        return ret;
      }
      break;
    }
    default:
      return TSDB_CODE_TSC_INVALID_JSON;
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseColsFromJSON(cJSON *root, SArray *cols) {
  cJSON *metricVal = cJSON_GetObjectItem(root, "value");
  if (metricVal == NULL) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
  if(!kv){
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  if(cols) taosArrayPush(cols, &kv);

  kv->key = VALUE;
  kv->keyLen = VALUE_LEN;
  int32_t ret = smlParseValueFromJSON(metricVal, kv);
  if (ret != TSDB_CODE_SUCCESS) {
    return ret;
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseTagsFromJSON(cJSON *root, SArray *pKVs, char *childTableName, SHashObj *dumplicateKey, SSmlMsgBuf *msg) {
  int32_t ret = TSDB_CODE_SUCCESS;

  cJSON *tags = cJSON_GetObjectItem(root, "tags");
  if (tags == NULL || tags->type != cJSON_Object) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  size_t childTableNameLen = strlen(tsSmlChildTableName);
  int32_t tagNum = cJSON_GetArraySize(tags);
  for (int32_t i = 0; i < tagNum; ++i) {
    cJSON *tag = cJSON_GetArrayItem(tags, i);
    if (tag == NULL) {
      return TSDB_CODE_TSC_INVALID_JSON;
    }
    size_t keyLen = strlen(tag->string);
    if (IS_INVALID_COL_LEN(keyLen)) {
      uError("OTD:Tag key length is 0 or too large than 64");
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
    }
    //check duplicate keys
    if (smlCheckDuplicateKey(tag->string, keyLen, dumplicateKey)) {
      return TSDB_CODE_TSC_DUP_TAG_NAMES;
    }

    //handle child table name
    if(childTableNameLen != 0 && strcmp(tag->string, tsSmlChildTableName) == 0){
      if (!cJSON_IsString(tag)) {
        uError("OTD:ID must be JSON string");
        return TSDB_CODE_TSC_INVALID_JSON;
      }
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, tag->valuestring, TSDB_TABLE_NAME_LEN);
      continue;
    }

    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
    if(!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if(pKVs) taosArrayPush(pKVs, &kv);

    //key
    kv->keyLen = keyLen;
    ret = smlJsonCreateSring(&kv->key, tag->string, kv->keyLen);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
    //value
    ret = smlParseValueFromJSON(tag, kv);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
  }

  return ret;

}

static int32_t smlParseJSONString(SSmlHandle *info, cJSON *root, SSmlTableInfo *tinfo, SArray *cols) {
  int32_t ret = TSDB_CODE_SUCCESS;

  if (!cJSON_IsObject(root)) {
    uError("OTD:0x%"PRIx64" data point needs to be JSON object", info->id);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

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

  //Parse metric
  ret = smlParseMetricFromJSON(info, root, tinfo);
  if (ret != TSDB_CODE_SUCCESS) {
    uError("OTD:0x%"PRIx64" Unable to parse metric from JSON payload", info->id);
    return ret;
  }
  uDebug("OTD:0x%"PRIx64" Parse metric from JSON payload finished", info->id);

  //Parse timestamp
  ret = smlParseTSFromJSON(info, root, cols);
  if (ret) {
    uError("OTD:0x%"PRIx64" Unable to parse timestamp from JSON payload", info->id);
    return ret;
  }
  uDebug("OTD:0x%"PRIx64" Parse timestamp from JSON payload finished", info->id);

  //Parse metric value
  ret = smlParseColsFromJSON(root, cols);
  if (ret) {
    uError("OTD:0x%"PRIx64" Unable to parse metric value from JSON payload", info->id);
    return ret;
  }
  uDebug("OTD:0x%"PRIx64" Parse metric value from JSON payload finished", info->id);

  //Parse tags
  ret = smlParseTagsFromJSON(root, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
  if (ret) {
    uError("OTD:0x%"PRIx64" Unable to parse tags from JSON payload", info->id);
    return ret;
  }
  uDebug("OTD:0x%"PRIx64" Parse tags from JSON payload finished", info->id);

  return TSDB_CODE_SUCCESS;
}
/************* TSDB_SML_JSON_PROTOCOL function end **************/



static int32_t smlParseInfluxLine(SSmlHandle* info, const char* sql) {
  SSmlLineInfo elements = {0};
  int ret = smlParseInfluxString(sql, &elements, &info->msgBuf);
  if(ret != TSDB_CODE_SUCCESS){
    uError("SML:0x%"PRIx64" smlParseInfluxLine failed", info->id);
    return ret;
  }

  SArray *cols = NULL;
  if(info->dataFormat){   // if dataFormat, cols need new memory to save data
    cols = taosArrayInit(16, POINTER_BYTES);
    if (cols == NULL) {
      uError("SML:0x%"PRIx64" smlParseInfluxLine failed to allocate memory", info->id);
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
  }else{      // if dataFormat is false, cols do not need to save data, there is another new memory to save data
    cols = info->colsContainer;
  }

  ret = smlParseTS(info, elements.timestamp, elements.timestampLen, cols);
  if(ret != TSDB_CODE_SUCCESS){
    uError("SML:0x%"PRIx64" smlParseTS failed", info->id);
    if(info->dataFormat) taosArrayDestroy(cols);
    return ret;
  }
  ret = smlParseCols(elements.cols, elements.colsLen, cols, NULL, false, info->dumplicateKey, &info->msgBuf);
  if(ret != TSDB_CODE_SUCCESS){
    uError("SML:0x%"PRIx64" smlParseCols parse cloums fields failed", info->id);
    smlDestroyCols(cols);
    if(info->dataFormat) taosArrayDestroy(cols);
    return ret;
  }
  if(taosArrayGetSize(cols) > TSDB_MAX_COLUMNS){
    smlBuildInvalidDataMsg(&info->msgBuf, "too many columns than 4096", NULL);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  bool hasTable = true;
  SSmlTableInfo *tinfo = NULL;
  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashGet(info->childTables, elements.measure, elements.measureTagsLen);
  if(!oneTable){
    tinfo = smlBuildTableInfo();
    if(!tinfo){
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    taosHashPut(info->childTables, elements.measure, elements.measureTagsLen, &tinfo, POINTER_BYTES);
    oneTable = &tinfo;
    hasTable = false;
  }

  ret = smlDealCols(*oneTable, info->dataFormat, cols);
  if(ret != TSDB_CODE_SUCCESS){
    return ret;
  }

  if(!hasTable){
    ret = smlParseCols(elements.tags, elements.tagsLen, (*oneTable)->tags, (*oneTable)->childTableName, true, info->dumplicateKey, &info->msgBuf);
    if(ret != TSDB_CODE_SUCCESS){
      uError("SML:0x%"PRIx64" smlParseCols parse tag fields failed", info->id);
      return ret;
    }

    if(taosArrayGetSize((*oneTable)->tags) > TSDB_MAX_TAGS){
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
      return TSDB_CODE_SML_INVALID_DATA;
    }

    (*oneTable)->sTableName = elements.measure;
    (*oneTable)->sTableNameLen = elements.measureLen;
    if(strlen((*oneTable)->childTableName) == 0){
      RandTableName rName = { (*oneTable)->tags, (*oneTable)->sTableName, (uint8_t)(*oneTable)->sTableNameLen,
                             (*oneTable)->childTableName, 0 };

      buildChildTableName(&rName);
      (*oneTable)->uid = rName.uid;
    }else{
      (*oneTable)->uid = *(uint64_t*)((*oneTable)->childTableName);
    }

  }

  SSmlSTableMeta** tableMeta = (SSmlSTableMeta**)taosHashGet(info->superTables, elements.measure, elements.measureLen);
  if(tableMeta){  // update meta
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
    if(!hasTable && ret){
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
    if(!ret){
      uError("SML:0x%"PRIx64" smlUpdateMeta failed", info->id);
      return TSDB_CODE_SML_INVALID_DATA;
    }
  }else{
    SSmlSTableMeta *meta = smlBuildSTableMeta();
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
    taosHashPut(info->superTables, elements.measure, elements.measureLen, &meta, POINTER_BYTES);
  }

  if(!info->dataFormat){
    taosArrayClear(info->colsContainer);
  }
  taosHashClear(info->dumplicateKey);
  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseTelnetLine(SSmlHandle* info, void *data) {
  int ret = TSDB_CODE_SUCCESS;
  SSmlTableInfo *tinfo = smlBuildTableInfo();
  if(!tinfo){
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

  SArray *cols = taosArrayInit(16, POINTER_BYTES);
  if (cols == NULL) {
    uError("SML:0x%"PRIx64" smlParseTelnetLine failed to allocate memory", info->id);
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

  if(info->protocol == TSDB_SML_TELNET_PROTOCOL){
    ret = smlParseTelnetString(info, (const char*)data, tinfo, cols);
  }else if(info->protocol == TSDB_SML_JSON_PROTOCOL){
    ret = smlParseJSONString(info, (cJSON *)data, tinfo, cols);
  }else{
    ASSERT(0);
  }
  if(ret != TSDB_CODE_SUCCESS){
    uError("SML:0x%"PRIx64" smlParseTelnetLine failed", info->id);
    smlDestroyTableInfo(info, tinfo);
    smlDestroyCols(cols);
    taosArrayDestroy(cols);
    return ret;
  }

  if(taosArrayGetSize(tinfo->tags) <= 0 || taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS){
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate tags length:[1,128]", NULL);
    smlDestroyTableInfo(info, tinfo);
    smlDestroyCols(cols);
    taosArrayDestroy(cols);
    return TSDB_CODE_SML_INVALID_DATA;
  }
  taosHashClear(info->dumplicateKey);

  if(strlen(tinfo->childTableName) == 0){
    RandTableName rName = { tinfo->tags, tinfo->sTableName, (uint8_t)tinfo->sTableNameLen,
                           tinfo->childTableName, 0 };
    buildChildTableName(&rName);
    tinfo->uid = rName.uid;
  }else{
    tinfo->uid = *(uint64_t*)(tinfo->childTableName);  // generate uid by name simple
  }


  bool hasTable = true;
  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashGet(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName));
  if(!oneTable) {
    taosHashPut(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName), &tinfo, POINTER_BYTES);
    oneTable = &tinfo;
    hasTable = false;
  }else{
    smlDestroyTableInfo(info, tinfo);
  }

  taosArrayPush((*oneTable)->cols, &cols);
  SSmlSTableMeta** tableMeta = (SSmlSTableMeta** )taosHashGet(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen);
  if(tableMeta){  // update meta
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
    if(!hasTable && ret){
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
    if(!ret){
      uError("SML:0x%"PRIx64" smlUpdateMeta failed", info->id);
      return TSDB_CODE_SML_INVALID_DATA;
    }
  }else{
    SSmlSTableMeta *meta = smlBuildSTableMeta();
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
    taosHashPut(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen, &meta, POINTER_BYTES);
  }

  return TSDB_CODE_SUCCESS;
}

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

  if (payload == NULL) {
    uError("SML:0x%"PRIx64" empty JSON Payload", info->id);
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  cJSON *root = cJSON_Parse(payload);
  if (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_IsObject(root)) {
    payloadNum = 1;
  } else if (cJSON_IsArray(root)) {
    payloadNum = cJSON_GetArraySize(root);
  } else {
    uError("SML:0x%"PRIx64" Invalid JSON Payload", info->id);
    ret = TSDB_CODE_TSC_INVALID_JSON;
    goto end;
  }

  for (int32_t i = 0; i < payloadNum; ++i) {
    cJSON *dataPoint = (payloadNum == 1 && cJSON_IsObject(root)) ? root : cJSON_GetArrayItem(root, i);
    ret = smlParseTelnetLine(info, dataPoint);
    if(ret != TSDB_CODE_SUCCESS){
      uError("SML:0x%"PRIx64" Invalid JSON Payload", info->id);
      goto end;
    }
  }

end:
  cJSON_Delete(root);
  return ret;
}

static int32_t smlInsertData(SSmlHandle* info) {
  int32_t code = TSDB_CODE_SUCCESS;

  SSmlTableInfo** oneTable = (SSmlTableInfo**)taosHashIterate(info->childTables, NULL);
  while (oneTable) {
    SSmlTableInfo* tableData = *oneTable;

    SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
    strcpy(pName.dbname, info->pRequest->pDb);
    memcpy(pName.tname, tableData->childTableName, strlen(tableData->childTableName));

    SRequestConnInfo conn = {0};
    conn.pTrans = info->taos->pAppInfo->pTransporter;
    conn.requestId = info->pRequest->requestId;
    conn.requestObjRefId = info->pRequest->self;
    conn.mgmtEps = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
                             
    SVgroupInfo vg;
    code = catalogGetTableHashVgroup(info->pCatalog, &conn, &pName, &vg);
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%"PRIx64" catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName);
      return code;
    }
    taosHashPut(info->pVgHash, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg));

    SSmlSTableMeta** pMeta = (SSmlSTableMeta** )taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen);
    ASSERT (NULL != pMeta && NULL != *pMeta);

    // use tablemeta of stable to save vgid and uid of child table
    (*pMeta)->tableMeta->vgId = vg.vgId;
    (*pMeta)->tableMeta->uid = tableData->uid; // one table merge data block together according uid

    code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat,
                       (*pMeta)->tableMeta, tableData->childTableName, info->msgBuf.buf, info->msgBuf.len);
    if(code != TSDB_CODE_SUCCESS){
      uError("SML:0x%"PRIx64" smlBindData failed", info->id);
      return code;
    }
    oneTable = (SSmlTableInfo**)taosHashIterate(info->childTables, oneTable);
  }

  code = smlBuildOutput(info->exec, info->pVgHash);
  if (code != TSDB_CODE_SUCCESS) {
    uError("SML:0x%"PRIx64" smlBuildOutput failed", info->id);
    return code;
  }
  info->cost.insertRpcTime = taosGetTimestampUs();

  //launchQueryImpl(info->pRequest, info->pQuery, false, NULL);
//  info->affectedRows = taos_affected_rows(info->pRequest);
//  return info->pRequest->code;

  launchAsyncQuery(info->pRequest, info->pQuery);
  return TSDB_CODE_SUCCESS;
}

static void smlPrintStatisticInfo(SSmlHandle *info){
  uError("SML:0x%"PRIx64" smlInsertLines result, code:%d,lineNum:%d,stable num:%d,ctable num:%d,create stable num:%d \
        parse cost:%"PRId64",schema cost:%"PRId64",bind cost:%"PRId64",rpc cost:%"PRId64",total cost:%"PRId64"", info->id, info->cost.code,
         info->cost.lineNum, info->cost.numOfSTables, info->cost.numOfCTables, info->cost.numOfCreateSTables,
         info->cost.schemaTime-info->cost.parseTime, info->cost.insertBindTime-info->cost.schemaTime,
         info->cost.insertRpcTime-info->cost.insertBindTime, info->cost.endTime-info->cost.insertRpcTime,
         info->cost.endTime-info->cost.parseTime);
}

static int32_t smlParseLine(SSmlHandle *info, char* lines[], int numLines){
  int32_t code = TSDB_CODE_SUCCESS;
  if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
    code = smlParseJSON(info, *lines);
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, *lines);
      return code;
    }
    return code;
  }

  for (int32_t i = 0; i < numLines; ++i) {
    if(info->protocol == TSDB_SML_LINE_PROTOCOL){
      code = smlParseInfluxLine(info, lines[i]);
    }else if(info->protocol == TSDB_SML_TELNET_PROTOCOL){
      code = smlParseTelnetLine(info, lines[i]);
    }else{
      ASSERT(0);
    }
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, lines[i]);
      return code;
    }
  }
  return code;
}

static int smlProcess(SSmlHandle *info, char* lines[], int numLines) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t retryNum = 0;

  info->cost.parseTime = taosGetTimestampUs();

  code = smlParseLine(info, lines, numLines);
  if (code != 0) {
    uError("SML:0x%"PRIx64" smlParseLine error : %s", info->id, tstrerror(code));
    return code;
  }

  info->cost.lineNum = numLines;
  info->cost.numOfSTables = taosHashGetSize(info->superTables);
  info->cost.numOfCTables = taosHashGetSize(info->childTables);

  info->cost.schemaTime = taosGetTimestampUs();

  do{
    code = smlModifyDBSchemas(info);
    if (code == 0) break;
  } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);

  if (code != 0) {
    uError("SML:0x%"PRIx64" smlModifyDBSchemas error : %s", info->id, tstrerror(code));
    return code;
  }

  info->cost.insertBindTime = taosGetTimestampUs();
  code = smlInsertData(info);
  if (code != 0) {
    uError("SML:0x%"PRIx64" smlInsertData error : %s", info->id, tstrerror(code));
    return code;
  }

  return code;
}

static int32_t isSchemalessDb(STscObj *taos, SRequestObj* request){
  SCatalog* catalog = NULL;
  int32_t code = catalogGetHandle(((STscObj *)taos)->pAppInfo->clusterId, &catalog);
  if(code != TSDB_CODE_SUCCESS){
    uError("SML get catalog error %d", code);
    return code;
  }

  SName          name;
  tNameSetDbName(&name, taos->acctId, taos->db, strlen(taos->db));
  char dbFname[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(&name, dbFname);
  SDbCfgInfo pInfo = {0};

  SRequestConnInfo conn = {0};
  conn.pTrans = taos->pAppInfo->pTransporter;
  conn.requestId = request->requestId;
  conn.requestObjRefId = request->self;
  conn.mgmtEps = getEpSet_s(&taos->pAppInfo->mgmtEp);

  code = catalogGetDBCfg(catalog, &conn, dbFname, &pInfo);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
  taosArrayDestroy(pInfo.pRetensions);
  
  if (!pInfo.schemaless){
    return TSDB_CODE_SML_INVALID_DB_CONF;
  }
  return TSDB_CODE_SUCCESS;
}

static void smlInsertCallback(void* param, void* res, int32_t code) {
  SRequestObj *pRequest = (SRequestObj *)res;
  SSmlHandle* info = (SSmlHandle *)param;

  uDebug("SML:0x%"PRIx64" result. code:%d, msg:%s", info->id, pRequest->code, pRequest->msgBuf);
  // lock
  if(code != TSDB_CODE_SUCCESS){
    taosThreadSpinLock(&info->params->lock);
    info->params->request->code = code;
    taosThreadSpinUnlock(&info->params->lock);
  }
  // unlock

  printf("SML:0x%" PRIx64 " insert finished, code: %d, total: %d\n", info->id, code, info->affectedRows);
  Params *pParam = info->params;
  bool isLast = info->isLast;
  info->cost.endTime = taosGetTimestampUs();
  info->cost.code = code;
  smlPrintStatisticInfo(info);
  smlDestroyInfo(info);

  if(isLast){
    tsem_post(&pParam->sem);
  }
}

/**
 * taos_schemaless_insert() parse and insert data points into database according to
 * different protocol.
 *
 * @param $lines input array may contain multiple lines, each line indicates a data point.
 *               If protocol=2 is used input array should contain single JSON
 *               string(e.g. char *lines[] = {"$JSON_string"}). If need to insert
 *               multiple data points in JSON format, should include them in $JSON_string
 *               as a JSON array.
 * @param $numLines indicates how many data points in $lines.
 *                  If protocol = 2 is used this param will be ignored as $lines should
 *                  contain single JSON string.
 * @param $protocol indicates which protocol to use for parsing:
 *                  0 - influxDB line protocol
 *                  1 - OpenTSDB telnet line protocol
 *                  2 - OpenTSDB JSON format protocol
 * @return return zero for successful insertion. Otherwise return none-zero error code of
 *         failure reason.
 *
 */

TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision) {
  SRequestObj* request = (SRequestObj*)createRequest((STscObj *)taos, TSDB_SQL_INSERT);
  if(!request){
    uError("SML:taos_schemaless_insert error request is null");
    return NULL;
  }

  ((STscObj *)taos)->schemalessType = 1;
  SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};

  int cnt = ceil(((double)numLines)/LINE_BATCH);
  Params params;
  params.request = request;
  tsem_init(&params.sem, 0, 0);
  taosThreadSpinInit(&(params.lock), 0);

  if(request->pDb == NULL){
    request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
    smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
    goto end;
  }

  if(isSchemalessDb(((STscObj *)taos), request) != TSDB_CODE_SUCCESS){
    request->code = TSDB_CODE_SML_INVALID_DB_CONF;
    smlBuildInvalidDataMsg(&msg, "Cannot write data to a non schemaless database", NULL);
    goto end;
  }

  if (!lines) {
    request->code = TSDB_CODE_SML_INVALID_DATA;
    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
    goto end;
  }

  if(protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL){
    request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
    smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
    goto end;
  }

  if(protocol == TSDB_SML_LINE_PROTOCOL && (precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)){
    request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
    smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
    goto end;
  }

  for (int i = 0; i < cnt; ++i) {
    SRequestObj* req = (SRequestObj*)createRequest((STscObj *)taos, TSDB_SQL_INSERT);
    if(!req){
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error request is null");
      goto end;
    }
    SSmlHandle* info = smlBuildSmlInfo(taos, req, (SMLProtocolType)protocol, precision);
    if(!info){
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error SSmlHandle is null");
      goto end;
    }

    int32_t perBatch = LINE_BATCH;

    if(numLines > perBatch){
      numLines -= perBatch;
      info->isLast = false;
    }else{
      perBatch = numLines;
      numLines = 0;
      info->isLast = true;
    }

    info->params = &params;
    info->affectedRows = perBatch;
    info->pRequest->body.queryFp = smlInsertCallback;
    info->pRequest->body.param   = info;
    int32_t code = smlProcess(info, lines, perBatch);
    lines += perBatch;
    if (code != TSDB_CODE_SUCCESS){
      info->pRequest->body.queryFp(info, req, code);
    }
  }
  tsem_wait(&params.sem);

end:
  taosThreadSpinDestroy(&params.lock);
  tsem_destroy(&params.sem);
//  ((STscObj *)taos)->schemalessType = 0;
  ((STscObj *)taos)->schemalessType = 1;
  uDebug("resultend:%s", request->msgBuf);
  return (TAOS_RES*)request;
}