clientSml.c 87.7 KB
Newer Older
wmmhello's avatar
wmmhello 已提交
1 2 3 4 5
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

X
Xiaoyu Wang 已提交
6 7 8 9 10
#include "cJSON.h"
#include "catalog.h"
#include "clientInt.h"
#include "osSemaphore.h"
#include "osThread.h"
wmmhello's avatar
wmmhello 已提交
11 12
#include "query.h"
#include "taos.h"
wmmhello's avatar
wmmhello 已提交
13
#include "taoserror.h"
X
Xiaoyu Wang 已提交
14
#include "tcommon.h"
wmmhello's avatar
wmmhello 已提交
15
#include "tdef.h"
X
Xiaoyu Wang 已提交
16
#include "tglobal.h"
wmmhello's avatar
wmmhello 已提交
17 18
#include "tlog.h"
#include "tmsg.h"
X
Xiaoyu Wang 已提交
19
#include "tname.h"
wmmhello's avatar
wmmhello 已提交
20 21
#include "ttime.h"
#include "ttypes.h"
wmmhello's avatar
wmmhello 已提交
22

wmmhello's avatar
wmmhello 已提交
23
//=================================================================================================
wmmhello's avatar
wmmhello 已提交
24 25 26 27 28 29 30

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

dengyihao's avatar
dengyihao 已提交
31 32 33 34 35 36
#define JUMP_SPACE(sql, sqlEnd) \
  while (sql < sqlEnd) {        \
    if (*sql == SPACE)          \
      sql++;                    \
    else                        \
      break;                    \
X
Xiaoyu Wang 已提交
37
  }
wmmhello's avatar
wmmhello 已提交
38
// comma ,
X
Xiaoyu Wang 已提交
39 40
#define IS_SLASH_COMMA(sql) (*(sql) == COMMA && *((sql)-1) == SLASH)
#define IS_COMMA(sql)       (*(sql) == COMMA && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
41
// space
X
Xiaoyu Wang 已提交
42 43
#define IS_SLASH_SPACE(sql) (*(sql) == SPACE && *((sql)-1) == SLASH)
#define IS_SPACE(sql)       (*(sql) == SPACE && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
44
// equal =
X
Xiaoyu Wang 已提交
45 46
#define IS_SLASH_EQUAL(sql) (*(sql) == EQUAL && *((sql)-1) == SLASH)
#define IS_EQUAL(sql)       (*(sql) == EQUAL && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
47
// quote "
X
Xiaoyu Wang 已提交
48 49
#define IS_SLASH_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) == SLASH)
#define IS_QUOTE(sql)       (*(sql) == QUOTE && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
50
// SLASH
X
Xiaoyu Wang 已提交
51
#define IS_SLASH_SLASH(sql) (*(sql) == SLASH && *((sql)-1) == SLASH)
wmmhello's avatar
wmmhello 已提交
52

X
Xiaoyu Wang 已提交
53 54
#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))
wmmhello's avatar
wmmhello 已提交
55

X
Xiaoyu Wang 已提交
56
#define MOVE_FORWARD_ONE(sql, len) (memmove((void *)((sql)-1), (sql), len))
wmmhello's avatar
wmmhello 已提交
57

X
Xiaoyu Wang 已提交
58 59 60 61 62 63 64 65
#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--;                              \
    }                                        \
  }
wmmhello's avatar
wmmhello 已提交
66

67 68 69
#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)

70 71 72
#define OTD_JSON_SUB_FIELDS_NUM 2
#define OTD_JSON_FIELDS_NUM     4

X
Xiaoyu Wang 已提交
73 74 75 76
#define TS        "_ts"
#define TS_LEN    3
#define VALUE     "_value"
#define VALUE_LEN 6
77

X
Xiaoyu Wang 已提交
78 79
#define BINARY_ADD_LEN 2  // "binary"   2 means " "
#define NCHAR_ADD_LEN  3  // L"nchar"   3 means L" "
wmmhello's avatar
wmmhello 已提交
80 81

#define MAX_RETRY_TIMES 5
wmmhello's avatar
wmmhello 已提交
82 83 84 85
//=================================================================================================
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;

typedef enum {
86
  SCHEMA_ACTION_NULL,
wmmhello's avatar
wmmhello 已提交
87 88 89 90 91
  SCHEMA_ACTION_CREATE_STABLE,
  SCHEMA_ACTION_ADD_COLUMN,
  SCHEMA_ACTION_ADD_TAG,
  SCHEMA_ACTION_CHANGE_COLUMN_SIZE,
  SCHEMA_ACTION_CHANGE_TAG_SIZE,
wmmhello's avatar
wmmhello 已提交
92 93 94
} ESchemaAction;

typedef struct {
X
Xiaoyu Wang 已提交
95 96 97 98
  const char *measure;
  const char *tags;
  const char *cols;
  const char *timestamp;
wmmhello's avatar
wmmhello 已提交
99 100 101 102 103 104 105 106 107

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

typedef struct {
X
Xiaoyu Wang 已提交
108 109 110 111
  const char *sTableName;  // super table name
  int32_t     sTableNameLen;
  char        childTableName[TSDB_TABLE_NAME_LEN];
  uint64_t    uid;
wmmhello's avatar
wmmhello 已提交
112

X
Xiaoyu Wang 已提交
113
  SArray *tags;
wmmhello's avatar
wmmhello 已提交
114

115 116
  // 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
X
Xiaoyu Wang 已提交
117
  SArray *cols;
wmmhello's avatar
wmmhello 已提交
118 119 120
} SSmlTableInfo;

typedef struct {
X
Xiaoyu Wang 已提交
121 122
  SArray   *tags;     // save the origin order to create table
  SHashObj *tagHash;  // elements are <key, index in tags>
123

X
Xiaoyu Wang 已提交
124 125
  SArray   *cols;
  SHashObj *colHash;
126

wmmhello's avatar
wmmhello 已提交
127 128 129 130
  STableMeta *tableMeta;
} SSmlSTableMeta;

typedef struct {
X
Xiaoyu Wang 已提交
131 132
  int32_t len;
  char   *buf;
wmmhello's avatar
wmmhello 已提交
133 134
} SSmlMsgBuf;

135 136 137 138 139 140 141
typedef struct {
  int32_t code;
  int32_t lineNum;

  int32_t numOfSTables;
  int32_t numOfCTables;
  int32_t numOfCreateSTables;
142 143
  int32_t numOfAlterColSTables;
  int32_t numOfAlterTagSTables;
144 145 146 147 148 149 150 151

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

X
Xiaoyu Wang 已提交
152 153 154
typedef struct {
  SRequestObj     *request;
  tsem_t           sem;
155 156
  int32_t          cnt;
  int32_t          total;
wmmhello's avatar
wmmhello 已提交
157 158 159
  TdThreadSpinlock lock;
} Params;

wmmhello's avatar
wmmhello 已提交
160
typedef struct {
X
Xiaoyu Wang 已提交
161 162 163 164 165
  int64_t id;
  Params *params;

  SMLProtocolType protocol;
  int8_t          precision;
166 167 168
  bool            dataFormat;  // true means that the name and order of keys in each line are the same(only for influx protocol)
  bool            isRawLine;
  int32_t         ttl;
X
Xiaoyu Wang 已提交
169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184

  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
wmmhello's avatar
wmmhello 已提交
185 186

  cJSON       *root;  // for parse json
wmmhello's avatar
wmmhello 已提交
187
} SSmlHandle;
wmmhello's avatar
wmmhello 已提交
188 189
//=================================================================================================

wmmhello's avatar
wmmhello 已提交
190
//=================================================================================================
191
static volatile int64_t linesSmlHandleId = 0;
X
Xiaoyu Wang 已提交
192 193
static int64_t          smlGenId() {
           int64_t id;
wmmhello's avatar
wmmhello 已提交
194

X
Xiaoyu Wang 已提交
195 196
           do {
             id = atomic_add_fetch_64(&linesSmlHandleId, 1);
wmmhello's avatar
wmmhello 已提交
197 198
  } while (id == 0);

X
Xiaoyu Wang 已提交
199
           return id;
wmmhello's avatar
wmmhello 已提交
200 201
}

202
static inline bool smlDoubleToInt64OverFlow(double num) {
X
Xiaoyu Wang 已提交
203
  if (num >= (double)INT64_MAX || num <= (double)INT64_MIN) return true;
204 205 206 207 208 209 210 211 212 213 214 215
  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;
}

X
Xiaoyu Wang 已提交
216
static int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2) {
217
  if (pBuf->buf) {
218 219 220 221 222 223 224
    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);
    }
wmmhello's avatar
wmmhello 已提交
225
  }
wmmhello's avatar
wmmhello 已提交
226 227 228
  return TSDB_CODE_SML_INVALID_DATA;
}

X
Xiaoyu Wang 已提交
229
static int32_t smlGenerateSchemaAction(SSchema *colField, SHashObj *colHash, SSmlKv *kv, bool isTag,
230
                                       ESchemaAction *action, SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
231
  uint16_t *index = colHash ? (uint16_t *)taosHashGet(colHash, kv->key, kv->keyLen) : NULL;
232 233
  if (index) {
    if (colField[*index].type != kv->type) {
X
Xiaoyu Wang 已提交
234 235
      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);
236 237 238
      return TSDB_CODE_TSC_INVALID_VALUE;
    }

X
Xiaoyu Wang 已提交
239 240 241 242
    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))) {
243
      if (isTag) {
wmmhello's avatar
wmmhello 已提交
244
        *action = SCHEMA_ACTION_CHANGE_TAG_SIZE;
245
      } else {
wmmhello's avatar
wmmhello 已提交
246
        *action = SCHEMA_ACTION_CHANGE_COLUMN_SIZE;
247 248 249 250
      }
    }
  } else {
    if (isTag) {
wmmhello's avatar
wmmhello 已提交
251
      *action = SCHEMA_ACTION_ADD_TAG;
252
    } else {
wmmhello's avatar
wmmhello 已提交
253
      *action = SCHEMA_ACTION_ADD_COLUMN;
254 255
    }
  }
wmmhello's avatar
wmmhello 已提交
256 257 258
  return 0;
}

wmmhello's avatar
wmmhello 已提交
259
static int32_t smlFindNearestPowerOf2(int32_t length, uint8_t type) {
wmmhello's avatar
wmmhello 已提交
260
  int32_t result = 1;
X
Xiaoyu Wang 已提交
261
  while (result <= length) {
wmmhello's avatar
wmmhello 已提交
262 263
    result *= 2;
  }
264
  if (type == TSDB_DATA_TYPE_BINARY && result > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
wmmhello's avatar
wmmhello 已提交
265
    result = TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE;
266
  } else if (type == TSDB_DATA_TYPE_NCHAR && result > (TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
wmmhello's avatar
wmmhello 已提交
267 268
    result = (TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
  }
wmmhello's avatar
wmmhello 已提交
269

270
  if (type == TSDB_DATA_TYPE_NCHAR) {
271
    result = result * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
272
  } else if (type == TSDB_DATA_TYPE_BINARY) {
273
    result = result + VARSTR_HEADER_SIZE;
wmmhello's avatar
wmmhello 已提交
274
  }
275
  return result;
wmmhello's avatar
wmmhello 已提交
276 277
}

X
Xiaoyu Wang 已提交
278
static int32_t smlProcessSchemaAction(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
279
                                      ESchemaAction *action, bool isTag) {
280 281
  int32_t code = TSDB_CODE_SUCCESS;
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
282
    if (j == 0 && !isTag) continue;
X
Xiaoyu Wang 已提交
283
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, j);
284
    code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, action, info);
X
Xiaoyu Wang 已提交
285
    if (code != TSDB_CODE_SUCCESS) {
286 287 288 289 290 291
      return code;
    }
  }
  return TSDB_CODE_SUCCESS;
}

292
static int32_t smlCheckMeta(SSchema *schema, int32_t length, SArray *cols, bool isTag) {
293
  SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
294 295
  int32_t   i = 0;
  for (; i < length; i++) {
296 297 298
    taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
  }

299
  if (isTag) {
300 301 302 303 304
    i = 0;
  } else {
    i = 1;
  }
  for (; i < taosArrayGetSize(cols); i++) {
X
Xiaoyu Wang 已提交
305 306
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
    if (taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL) {
wmmhello's avatar
wmmhello 已提交
307
      taosHashCleanup(hashTmp);
308 309 310
      return -1;
    }
  }
311
  taosHashCleanup(hashTmp);
312 313 314
  return 0;
}

315
static int32_t getBytes(uint8_t type, int32_t length) {
316 317 318 319 320 321 322
  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
    return smlFindNearestPowerOf2(length, type);
  } else {
    return tDataTypes[type].bytes;
  }
}

323 324
static int32_t smlBuildFieldsList(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
                                  SArray *results, int32_t numOfCols, bool isTag) {
wmmhello's avatar
wmmhello 已提交
325
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
326
    SSmlKv       *kv = (SSmlKv *)taosArrayGetP(cols, j);
wmmhello's avatar
wmmhello 已提交
327 328
    ESchemaAction action = SCHEMA_ACTION_NULL;
    smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, &action, info);
329
    if (action == SCHEMA_ACTION_ADD_COLUMN || action == SCHEMA_ACTION_ADD_TAG) {
wmmhello's avatar
wmmhello 已提交
330 331 332 333 334
      SField field = {0};
      field.type = kv->type;
      field.bytes = getBytes(kv->type, kv->length);
      memcpy(field.name, kv->key, kv->keyLen);
      taosArrayPush(results, &field);
335
    } else if (action == SCHEMA_ACTION_CHANGE_COLUMN_SIZE || action == SCHEMA_ACTION_CHANGE_TAG_SIZE) {
wmmhello's avatar
wmmhello 已提交
336
      uint16_t *index = (uint16_t *)taosHashGet(schemaHash, kv->key, kv->keyLen);
337 338
      uint16_t  newIndex = *index;
      if (isTag) newIndex -= numOfCols;
wmmhello's avatar
wmmhello 已提交
339 340 341 342 343 344 345
      SField *field = (SField *)taosArrayGet(results, newIndex);
      field->bytes = getBytes(kv->type, kv->length);
    }
  }
  return TSDB_CODE_SUCCESS;
}

346 347 348 349 350
// static int32_t smlSendMetaMsg(SSmlHandle *info, SName *pName, SSmlSTableMeta *sTableData,
//                               int32_t colVer, int32_t tagVer, int8_t source, uint64_t suid){
static int32_t smlSendMetaMsg(SSmlHandle *info, SName *pName, SArray *pColumns, SArray *pTags, STableMeta *pTableMeta,
                              ESchemaAction action) {
  SRequestObj   *pRequest = NULL;
351 352 353 354
  SMCreateStbReq pReq = {0};
  int32_t        code = TSDB_CODE_SUCCESS;
  SCmdMsgInfo    pCmdMsg = {0};

wmmhello's avatar
wmmhello 已提交
355 356 357 358 359 360
  // put front for free
  pReq.numOfColumns = taosArrayGetSize(pColumns);
  pReq.pColumns = pColumns;
  pReq.numOfTags = taosArrayGetSize(pTags);
  pReq.pTags = pTags;

dengyihao's avatar
dengyihao 已提交
361
  code = buildRequest(info->taos->id, "", 0, NULL, false, &pRequest, 0);
362 363 364 365
  if (code != TSDB_CODE_SUCCESS) {
    goto end;
  }

wmmhello's avatar
wmmhello 已提交
366
  pRequest->syncQuery = true;
367 368 369 370 371
  if (!pRequest->pDb) {
    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
    goto end;
  }

372
  if (action == SCHEMA_ACTION_CREATE_STABLE) {
wmmhello's avatar
wmmhello 已提交
373 374 375 376
    pReq.colVer = 1;
    pReq.tagVer = 1;
    pReq.suid = 0;
    pReq.source = TD_REQ_FROM_APP;
377
  } else if (action == SCHEMA_ACTION_ADD_TAG || action == SCHEMA_ACTION_CHANGE_TAG_SIZE) {
wmmhello's avatar
wmmhello 已提交
378 379 380 381
    pReq.colVer = pTableMeta->sversion;
    pReq.tagVer = pTableMeta->tversion + 1;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
382
  } else if (action == SCHEMA_ACTION_ADD_COLUMN || action == SCHEMA_ACTION_CHANGE_COLUMN_SIZE) {
wmmhello's avatar
wmmhello 已提交
383 384 385 386 387 388
    pReq.colVer = pTableMeta->sversion + 1;
    pReq.tagVer = pTableMeta->tversion;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
  }

389
  if (pReq.numOfTags == 0) {
wmmhello's avatar
wmmhello 已提交
390 391 392 393 394 395 396 397
    pReq.numOfTags = 1;
    SField field = {0};
    field.type = TSDB_DATA_TYPE_NCHAR;
    field.bytes = 1;
    strcpy(field.name, tsSmlTagName);
    taosArrayPush(pReq.pTags, &field);
  }

398 399 400 401 402 403 404 405 406 407 408 409 410 411
  pReq.commentLen = -1;
  pReq.igExists = true;
  tNameExtractFullName(pName, pReq.name);

  pCmdMsg.epSet = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
  pCmdMsg.msgType = TDMT_MND_CREATE_STB;
  pCmdMsg.msgLen = tSerializeSMCreateStbReq(NULL, 0, &pReq);
  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);
  if (NULL == pCmdMsg.pMsg) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto end;
  }
  tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq);

D
dapan1121 已提交
412 413
  SQuery pQuery;
  memset(&pQuery, 0, sizeof(pQuery));
414 415 416 417 418 419 420
  pQuery.execMode = QUERY_EXEC_MODE_RPC;
  pQuery.pCmdMsg = &pCmdMsg;
  pQuery.msgType = pQuery.pCmdMsg->msgType;
  pQuery.stableQuery = true;

  launchQueryImpl(pRequest, &pQuery, true, NULL);

421
  if (pRequest->code == TSDB_CODE_SUCCESS) {
422 423 424 425 426 427 428 429 430 431 432
    catalogRemoveTableMeta(info->pCatalog, pName);
  }
  code = pRequest->code;
  taosMemoryFree(pCmdMsg.pMsg);

end:
  destroyRequest(pRequest);
  tFreeSMCreateStbReq(&pReq);
  return code;
}

X
Xiaoyu Wang 已提交
433
static int32_t smlModifyDBSchemas(SSmlHandle *info) {
434 435 436
  int32_t     code = 0;
  SHashObj   *hashTmp = NULL;
  STableMeta *pTableMeta = NULL;
wmmhello's avatar
wmmhello 已提交
437

438
  SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
wmmhello's avatar
wmmhello 已提交
439
  tstrncpy(pName.dbname, info->pRequest->pDb, sizeof(pName.dbname));
wmmhello's avatar
wmmhello 已提交
440

D
dapan1121 已提交
441 442 443 444 445
  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);
X
Xiaoyu Wang 已提交
446 447

  SSmlSTableMeta **tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, NULL);
wmmhello's avatar
wmmhello 已提交
448
  while (tableMetaSml) {
X
Xiaoyu Wang 已提交
449 450
    SSmlSTableMeta *sTableData = *tableMetaSml;
    bool            needCheckMeta = false;  // for multi thread
wmmhello's avatar
wmmhello 已提交
451

wmmhello's avatar
wmmhello 已提交
452
    size_t superTableLen = 0;
X
Xiaoyu Wang 已提交
453
    void  *superTable = taosHashGetKey(tableMetaSml, &superTableLen);
454
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
wmmhello's avatar
wmmhello 已提交
455
    memcpy(pName.tname, superTable, superTableLen);
wmmhello's avatar
wmmhello 已提交
456

D
dapan1121 已提交
457
    code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
wmmhello's avatar
wmmhello 已提交
458

459
    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_STB_NOT_EXIST) {
460 461
      SArray *pColumns = taosArrayInit(taosArrayGetSize(sTableData->cols), sizeof(SField));
      SArray *pTags = taosArrayInit(taosArrayGetSize(sTableData->tags), sizeof(SField));
wmmhello's avatar
wmmhello 已提交
462 463 464 465
      smlBuildFieldsList(info, NULL, NULL, sTableData->tags, pTags, 0, true);
      smlBuildFieldsList(info, NULL, NULL, sTableData->cols, pColumns, 0, false);

      code = smlSendMetaMsg(info, &pName, pColumns, pTags, NULL, SCHEMA_ACTION_CREATE_STABLE);
466
      if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
467
        uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, pName.tname);
468
        goto end;
wmmhello's avatar
wmmhello 已提交
469
      }
470
      info->cost.numOfCreateSTables++;
wmmhello's avatar
wmmhello 已提交
471 472 473 474 475 476 477
      taosMemoryFreeClear(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, pName.tname);
        goto end;
      }
X
Xiaoyu Wang 已提交
478
    } else if (code == TSDB_CODE_SUCCESS) {
479 480
      hashTmp = taosHashInit(pTableMeta->tableInfo.numOfTags, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true,
                             HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
481 482
      for (uint16_t i = pTableMeta->tableInfo.numOfColumns;
           i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
483 484
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
wmmhello's avatar
wmmhello 已提交
485

486 487
      ESchemaAction action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &action, true);
488
      if (code != TSDB_CODE_SUCCESS) {
489
        goto end;
490
      }
491 492 493 494 495
      if (action != SCHEMA_ACTION_NULL) {
        SArray *pColumns =
            taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
        SArray *pTags =
            taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
wmmhello's avatar
wmmhello 已提交
496 497 498 499 500 501

        for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
          SField field = {0};
          field.type = pTableMeta->schema[i].type;
          field.bytes = pTableMeta->schema[i].bytes;
          strcpy(field.name, pTableMeta->schema[i].name);
502
          if (i < pTableMeta->tableInfo.numOfColumns) {
wmmhello's avatar
wmmhello 已提交
503
            taosArrayPush(pColumns, &field);
504
          } else {
wmmhello's avatar
wmmhello 已提交
505 506 507
            taosArrayPush(pTags, &field);
          }
        }
508 509
        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->tags, pTags,
                           pTableMeta->tableInfo.numOfColumns, true);
wmmhello's avatar
wmmhello 已提交
510 511

        code = smlSendMetaMsg(info, &pName, pColumns, pTags, pTableMeta, action);
512
        if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
513
          uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, pName.tname);
514 515 516
          goto end;
        }

517
        info->cost.numOfAlterTagSTables++;
wmmhello's avatar
wmmhello 已提交
518 519 520 521 522 523 524 525 526
        taosMemoryFreeClear(pTableMeta);
        code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
        if (code != TSDB_CODE_SUCCESS) {
          goto end;
        }
        code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
        if (code != TSDB_CODE_SUCCESS) {
          goto end;
        }
wmmhello's avatar
wmmhello 已提交
527
      }
528 529

      taosHashClear(hashTmp);
wmmhello's avatar
wmmhello 已提交
530
      for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns; i++) {
531 532
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
533 534
      action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->cols, &action, false);
535
      if (code != TSDB_CODE_SUCCESS) {
536
        goto end;
wmmhello's avatar
wmmhello 已提交
537
      }
538 539 540 541 542
      if (action != SCHEMA_ACTION_NULL) {
        SArray *pColumns =
            taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
        SArray *pTags =
            taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
wmmhello's avatar
wmmhello 已提交
543 544 545 546 547 548

        for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
          SField field = {0};
          field.type = pTableMeta->schema[i].type;
          field.bytes = pTableMeta->schema[i].bytes;
          strcpy(field.name, pTableMeta->schema[i].name);
549
          if (i < pTableMeta->tableInfo.numOfColumns) {
wmmhello's avatar
wmmhello 已提交
550
            taosArrayPush(pColumns, &field);
551
          } else {
wmmhello's avatar
wmmhello 已提交
552 553 554 555
            taosArrayPush(pTags, &field);
          }
        }

556 557
        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->cols, pColumns,
                           pTableMeta->tableInfo.numOfColumns, false);
wmmhello's avatar
wmmhello 已提交
558 559

        code = smlSendMetaMsg(info, &pName, pColumns, pTags, pTableMeta, action);
560
        if (code != TSDB_CODE_SUCCESS) {
561
          uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, pName.tname);
562 563
          goto end;
        }
wmmhello's avatar
wmmhello 已提交
564

565
        info->cost.numOfAlterColSTables++;
wmmhello's avatar
wmmhello 已提交
566
        taosMemoryFreeClear(pTableMeta);
wmmhello's avatar
wmmhello 已提交
567 568 569 570
        code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
        if (code != TSDB_CODE_SUCCESS) {
          goto end;
        }
571 572 573 574 575
        code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " catalogGetSTableMeta failed. super table name %s", info->id, pName.tname);
          goto end;
        }
wmmhello's avatar
wmmhello 已提交
576
      }
wmmhello's avatar
wmmhello 已提交
577

578
      needCheckMeta = true;
wmmhello's avatar
wmmhello 已提交
579 580
      taosHashCleanup(hashTmp);
      hashTmp = NULL;
wmmhello's avatar
wmmhello 已提交
581
    } else {
X
Xiaoyu Wang 已提交
582
      uError("SML:0x%" PRIx64 " load table meta error: %s", info->id, tstrerror(code));
583
      goto end;
wmmhello's avatar
wmmhello 已提交
584
    }
585

X
Xiaoyu Wang 已提交
586 587
    if (needCheckMeta) {
      code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags,
588
                          sTableData->tags, true);
589
      if (code != TSDB_CODE_SUCCESS) {
590
        uError("SML:0x%" PRIx64 " check tag failed. super table name %s", info->id, pName.tname);
591 592
        goto end;
      }
593
      code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols, false);
594
      if (code != TSDB_CODE_SUCCESS) {
595
        uError("SML:0x%" PRIx64 " check cols failed. super table name %s", info->id, pName.tname);
596 597 598 599
        goto end;
      }
    }

600
    sTableData->tableMeta = pTableMeta;
wmmhello's avatar
wmmhello 已提交
601

X
Xiaoyu Wang 已提交
602
    tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, tableMetaSml);
wmmhello's avatar
wmmhello 已提交
603 604
  }
  return 0;
605 606

end:
wmmhello's avatar
wmmhello 已提交
607 608
  taosHashCleanup(hashTmp);
  taosMemoryFreeClear(pTableMeta);
wmmhello's avatar
wmmhello 已提交
609
//  catalogRefreshTableMeta(info->pCatalog, &conn, &pName, 1);
610
  return code;
wmmhello's avatar
wmmhello 已提交
611 612
}

X
Xiaoyu Wang 已提交
613
static bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
614
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
615 616 617 618
  int32_t     len = kvVal->length;
  char       *endptr = NULL;
  double      result = taosStr2Double(pVal, &endptr);
  if (pVal == endptr) {
619
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
620 621 622
    return false;
  }

623
  int32_t left = len - (endptr - pVal);
X
Xiaoyu Wang 已提交
624
  if (left == 0 || (left == 3 && strncasecmp(endptr, "f64", left) == 0)) {
625 626
    kvVal->type = TSDB_DATA_TYPE_DOUBLE;
    kvVal->d = result;
X
Xiaoyu Wang 已提交
627 628
  } else if ((left == 3 && strncasecmp(endptr, "f32", left) == 0)) {
    if (!IS_VALID_FLOAT(result)) {
629 630
      smlBuildInvalidDataMsg(msg, "float out of range[-3.402823466e+38,3.402823466e+38]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
631
    }
632 633
    kvVal->type = TSDB_DATA_TYPE_FLOAT;
    kvVal->f = (float)result;
X
Xiaoyu Wang 已提交
634 635
  } else if ((left == 1 && *endptr == 'i') || (left == 3 && strncasecmp(endptr, "i64", left) == 0)) {
    if (smlDoubleToInt64OverFlow(result)) {
wmmhello's avatar
wmmhello 已提交
636 637
      errno = 0;
      int64_t tmp = taosStr2Int64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
638
      if (errno == ERANGE) {
wmmhello's avatar
wmmhello 已提交
639 640 641 642 643 644
        smlBuildInvalidDataMsg(msg, "big int out of range[-9223372036854775808,9223372036854775807]", pVal);
        return false;
      }
      kvVal->type = TSDB_DATA_TYPE_BIGINT;
      kvVal->i = tmp;
      return true;
wmmhello's avatar
wmmhello 已提交
645
    }
646
    kvVal->type = TSDB_DATA_TYPE_BIGINT;
wmmhello's avatar
wmmhello 已提交
647
    kvVal->i = (int64_t)result;
wmmhello's avatar
wmmhello 已提交
648
  } else if ((left == 1 && *endptr == 'u') || (left == 3 && strncasecmp(endptr, "u64", left) == 0)) {
X
Xiaoyu Wang 已提交
649
    if (result >= (double)UINT64_MAX || result < 0) {
wmmhello's avatar
wmmhello 已提交
650 651
      errno = 0;
      uint64_t tmp = taosStr2UInt64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
652
      if (errno == ERANGE || result < 0) {
wmmhello's avatar
wmmhello 已提交
653 654 655 656 657 658
        smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", pVal);
        return false;
      }
      kvVal->type = TSDB_DATA_TYPE_UBIGINT;
      kvVal->u = tmp;
      return true;
659
    }
660
    kvVal->type = TSDB_DATA_TYPE_UBIGINT;
wmmhello's avatar
wmmhello 已提交
661
    kvVal->u = result;
X
Xiaoyu Wang 已提交
662 663
  } else if (left == 3 && strncasecmp(endptr, "i32", left) == 0) {
    if (!IS_VALID_INT(result)) {
664 665
      smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
666
    }
667 668
    kvVal->type = TSDB_DATA_TYPE_INT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
669 670
  } else if (left == 3 && strncasecmp(endptr, "u32", left) == 0) {
    if (!IS_VALID_UINT(result)) {
671 672
      smlBuildInvalidDataMsg(msg, "unsigned int out of range[0,4294967295]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
673
    }
674 675
    kvVal->type = TSDB_DATA_TYPE_UINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
676 677
  } else if (left == 3 && strncasecmp(endptr, "i16", left) == 0) {
    if (!IS_VALID_SMALLINT(result)) {
678 679
      smlBuildInvalidDataMsg(msg, "small int our of range[-32768,32767]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
680
    }
681 682
    kvVal->type = TSDB_DATA_TYPE_SMALLINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
683 684
  } else if (left == 3 && strncasecmp(endptr, "u16", left) == 0) {
    if (!IS_VALID_USMALLINT(result)) {
685 686
      smlBuildInvalidDataMsg(msg, "unsigned small int out of rang[0,65535]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
687
    }
688 689
    kvVal->type = TSDB_DATA_TYPE_USMALLINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
690 691
  } else if (left == 2 && strncasecmp(endptr, "i8", left) == 0) {
    if (!IS_VALID_TINYINT(result)) {
692 693
      smlBuildInvalidDataMsg(msg, "tiny int out of range[-128,127]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
694
    }
695 696
    kvVal->type = TSDB_DATA_TYPE_TINYINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
697 698
  } else if (left == 2 && strncasecmp(endptr, "u8", left) == 0) {
    if (!IS_VALID_UTINYINT(result)) {
699 700
      smlBuildInvalidDataMsg(msg, "unsigned tiny int out of range[0,255]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
701
    }
702 703
    kvVal->type = TSDB_DATA_TYPE_UTINYINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
704
  } else {
705
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
706 707
    return false;
  }
708
  return true;
wmmhello's avatar
wmmhello 已提交
709 710
}

wmmhello's avatar
wmmhello 已提交
711
static bool smlParseBool(SSmlKv *kvVal) {
wmmhello's avatar
wmmhello 已提交
712
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
713
  int32_t     len = kvVal->length;
714
  if ((len == 1) && (pVal[0] == 't' || pVal[0] == 'T')) {
wmmhello's avatar
wmmhello 已提交
715
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
716 717 718
    return true;
  }

719
  if ((len == 1) && (pVal[0] == 'f' || pVal[0] == 'F')) {
wmmhello's avatar
wmmhello 已提交
720
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
721 722 723
    return true;
  }

X
Xiaoyu Wang 已提交
724
  if ((len == 4) && !strncasecmp(pVal, "true", len)) {
wmmhello's avatar
wmmhello 已提交
725
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
726 727
    return true;
  }
X
Xiaoyu Wang 已提交
728
  if ((len == 5) && !strncasecmp(pVal, "false", len)) {
wmmhello's avatar
wmmhello 已提交
729
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
730 731 732 733 734
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
735
static bool smlIsBinary(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
736
  // binary: "abc"
wmmhello's avatar
wmmhello 已提交
737 738 739 740 741 742 743 744 745
  if (len < 2) {
    return false;
  }
  if (pVal[0] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
746
static bool smlIsNchar(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
747
  // nchar: L"abc"
wmmhello's avatar
wmmhello 已提交
748 749 750
  if (len < 3) {
    return false;
  }
X
Xiaoyu Wang 已提交
751
  if ((pVal[0] == 'l' || pVal[0] == 'L') && pVal[1] == '"' && pVal[len - 1] == '"') {
wmmhello's avatar
wmmhello 已提交
752 753 754 755 756
    return true;
  }
  return false;
}

757
static int64_t smlGetTimeValue(const char *value, int32_t len, int8_t type) {
X
Xiaoyu Wang 已提交
758
  char   *endPtr = NULL;
wafwerar's avatar
wafwerar 已提交
759
  int64_t tsInt64 = taosStr2Int64(value, &endPtr, 10);
X
Xiaoyu Wang 已提交
760
  if (value + len != endPtr) {
761
    return -1;
wmmhello's avatar
wmmhello 已提交
762
  }
763
  double ts = tsInt64;
764 765
  switch (type) {
    case TSDB_TIME_PRECISION_HOURS:
wmmhello's avatar
wmmhello 已提交
766 767
      ts *= NANOSECOND_PER_HOUR;
      tsInt64 *= NANOSECOND_PER_HOUR;
768 769
      break;
    case TSDB_TIME_PRECISION_MINUTES:
wmmhello's avatar
wmmhello 已提交
770 771
      ts *= NANOSECOND_PER_MINUTE;
      tsInt64 *= NANOSECOND_PER_MINUTE;
772 773
      break;
    case TSDB_TIME_PRECISION_SECONDS:
wmmhello's avatar
wmmhello 已提交
774 775
      ts *= NANOSECOND_PER_SEC;
      tsInt64 *= NANOSECOND_PER_SEC;
776 777
      break;
    case TSDB_TIME_PRECISION_MILLI:
wmmhello's avatar
wmmhello 已提交
778 779
      ts *= NANOSECOND_PER_MSEC;
      tsInt64 *= NANOSECOND_PER_MSEC;
780 781
      break;
    case TSDB_TIME_PRECISION_MICRO:
wmmhello's avatar
wmmhello 已提交
782 783
      ts *= NANOSECOND_PER_USEC;
      tsInt64 *= NANOSECOND_PER_USEC;
784 785 786 787 788
      break;
    case TSDB_TIME_PRECISION_NANO:
      break;
    default:
      ASSERT(0);
wmmhello's avatar
wmmhello 已提交
789
  }
X
Xiaoyu Wang 已提交
790
  if (ts >= (double)INT64_MAX || ts < 0) {
791
    return -1;
wmmhello's avatar
wmmhello 已提交
792 793
  }

794
  return tsInt64;
795 796 797 798 799 800
}

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) {
801
    return TSDB_TIME_PRECISION_MILLI;
802 803
  } else {
    return -1;
wmmhello's avatar
wmmhello 已提交
804
  }
805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823
}

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;
wmmhello's avatar
wmmhello 已提交
824
  }
825 826
}

X
Xiaoyu Wang 已提交
827 828
static int64_t smlParseInfluxTime(SSmlHandle *info, const char *data, int32_t len) {
  if (len == 0 || (len == 1 && data[0] == '0')) {
829
    return taosGetTimestampNs();
wmmhello's avatar
wmmhello 已提交
830 831
  }

832 833 834 835
  int8_t tsType = smlGetTsTypeByPrecision(info->precision);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp precision", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
836
  }
837 838

  int64_t ts = smlGetTimeValue(data, len, tsType);
X
Xiaoyu Wang 已提交
839
  if (ts == -1) {
840 841
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
wmmhello's avatar
wmmhello 已提交
842
  }
843 844 845
  return ts;
}

X
Xiaoyu Wang 已提交
846 847
static int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len) {
  if (!data) {
848 849
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
850
  }
X
Xiaoyu Wang 已提交
851
  if (len == 1 && data[0] == '0') {
852 853
    return taosGetTimestampNs();
  }
854 855
  int8_t tsType = smlGetTsTypeByLen(len);
  if (tsType == -1) {
X
Xiaoyu Wang 已提交
856 857
    smlBuildInvalidDataMsg(&info->msgBuf,
                           "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
858 859 860
    return -1;
  }
  int64_t ts = smlGetTimeValue(data, len, tsType);
X
Xiaoyu Wang 已提交
861
  if (ts == -1) {
862 863 864 865 866 867
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

X
Xiaoyu Wang 已提交
868
static int32_t smlParseTS(SSmlHandle *info, const char *data, int32_t len, SArray *cols) {
869
  int64_t ts = 0;
X
Xiaoyu Wang 已提交
870
  if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
871
    //    uError("SML:data:%s,len:%d", data, len);
872
    ts = smlParseInfluxTime(info, data, len);
X
Xiaoyu Wang 已提交
873
  } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
874
    ts = smlParseOpenTsdbTime(info, data, len);
X
Xiaoyu Wang 已提交
875
  } else {
876
    ASSERT(0);
877
  }
wmmhello's avatar
wmmhello 已提交
878
  uDebug("SML:0x%" PRIx64 " smlParseTS:%" PRId64, info->id, ts);
879

880 881 882 883
  if (ts <= 0) {
    uError("SML:0x%" PRIx64 " smlParseTS error:%" PRId64, info->id, ts);
    return TSDB_CODE_INVALID_TIMESTAMP;
  }
884 885 886

  // add ts to
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
887
  if (!kv) {
888 889 890 891 892 893 894 895
    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;
wmmhello's avatar
wmmhello 已提交
896 897
  taosArrayPush(cols, &kv);

898 899 900
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
901
static int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
X
Xiaoyu Wang 已提交
902
  // binary
wmmhello's avatar
wmmhello 已提交
903
  if (smlIsBinary(pVal->value, pVal->length)) {
904
    pVal->type = TSDB_DATA_TYPE_BINARY;
wmmhello's avatar
wmmhello 已提交
905
    pVal->length -= BINARY_ADD_LEN;
906
    if (pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
wmmhello's avatar
wmmhello 已提交
907 908
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
909
    pVal->value += (BINARY_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
910
    return TSDB_CODE_SUCCESS;
911
  }
X
Xiaoyu Wang 已提交
912
  // nchar
wmmhello's avatar
wmmhello 已提交
913
  if (smlIsNchar(pVal->value, pVal->length)) {
914
    pVal->type = TSDB_DATA_TYPE_NCHAR;
wmmhello's avatar
wmmhello 已提交
915
    pVal->length -= NCHAR_ADD_LEN;
916
    if (pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
wmmhello's avatar
wmmhello 已提交
917 918
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
919
    pVal->value += (NCHAR_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
920
    return TSDB_CODE_SUCCESS;
921 922
  }

X
Xiaoyu Wang 已提交
923
  // bool
924 925 926
  if (smlParseBool(pVal)) {
    pVal->type = TSDB_DATA_TYPE_BOOL;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
927
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
928
  }
X
Xiaoyu Wang 已提交
929
  // number
930
  if (smlParseNumber(pVal, msg)) {
wmmhello's avatar
wmmhello 已提交
931
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
932
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
933 934
  }

wmmhello's avatar
wmmhello 已提交
935
  return TSDB_CODE_TSC_INVALID_VALUE;
wmmhello's avatar
wmmhello 已提交
936 937
}

wmmhello's avatar
wmmhello 已提交
938
static int32_t smlParseInfluxString(const char *sql, const char *sqlEnd, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
X
Xiaoyu Wang 已提交
939
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
940
  JUMP_SPACE(sql, sqlEnd)
X
Xiaoyu Wang 已提交
941
  if (*sql == COMMA) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
942
  elements->measure = sql;
wmmhello's avatar
wmmhello 已提交
943

wmmhello's avatar
wmmhello 已提交
944
  // parse measure
wmmhello's avatar
wmmhello 已提交
945
  while (sql < sqlEnd) {
X
Xiaoyu Wang 已提交
946
    if ((sql != elements->measure) && IS_SLASH_LETTER(sql)) {
wmmhello's avatar
wmmhello 已提交
947 948
      MOVE_FORWARD_ONE(sql, sqlEnd - sql);
      sqlEnd--;
wmmhello's avatar
wmmhello 已提交
949 950
      continue;
    }
X
Xiaoyu Wang 已提交
951
    if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
952 953 954
      break;
    }

X
Xiaoyu Wang 已提交
955
    if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
956 957
      break;
    }
wmmhello's avatar
wmmhello 已提交
958 959
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
960
  elements->measureLen = sql - elements->measure;
X
Xiaoyu Wang 已提交
961
  if (IS_INVALID_TABLE_LEN(elements->measureLen)) {
962
    smlBuildInvalidDataMsg(msg, "measure is empty or too large than 192", NULL);
wmmhello's avatar
wmmhello 已提交
963
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
wmmhello's avatar
wmmhello 已提交
964
  }
wmmhello's avatar
wmmhello 已提交
965

wmmhello's avatar
wmmhello 已提交
966
  // parse tag
X
Xiaoyu Wang 已提交
967
  if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
968
    elements->tagsLen = 0;
X
Xiaoyu Wang 已提交
969 970
  } else {
    if (*sql == COMMA) sql++;
wmmhello's avatar
wmmhello 已提交
971
    elements->tags = sql;
wmmhello's avatar
wmmhello 已提交
972
    while (sql < sqlEnd) {
X
Xiaoyu Wang 已提交
973
      if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
974 975 976
        break;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
977
    }
wmmhello's avatar
wmmhello 已提交
978
    elements->tagsLen = sql - elements->tags;
979
  }
wmmhello's avatar
wmmhello 已提交
980
  elements->measureTagsLen = sql - elements->measure;
wmmhello's avatar
wmmhello 已提交
981

wmmhello's avatar
wmmhello 已提交
982
  // parse cols
wmmhello's avatar
wmmhello 已提交
983
  JUMP_SPACE(sql, sqlEnd)
wmmhello's avatar
wmmhello 已提交
984
  elements->cols = sql;
wmmhello's avatar
wmmhello 已提交
985
  bool isInQuote = false;
wmmhello's avatar
wmmhello 已提交
986
  while (sql < sqlEnd) {
X
Xiaoyu Wang 已提交
987
    if (IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
988 989
      isInQuote = !isInQuote;
    }
X
Xiaoyu Wang 已提交
990
    if (!isInQuote && IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
991 992 993 994
      break;
    }
    sql++;
  }
X
Xiaoyu Wang 已提交
995
  if (isInQuote) {
996 997 998
    smlBuildInvalidDataMsg(msg, "only one quote", elements->cols);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
999
  elements->colsLen = sql - elements->cols;
X
Xiaoyu Wang 已提交
1000
  if (elements->colsLen == 0) {
wmmhello's avatar
wmmhello 已提交
1001 1002 1003
    smlBuildInvalidDataMsg(msg, "cols is empty", NULL);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
1004

wmmhello's avatar
wmmhello 已提交
1005
  // parse timestamp
wmmhello's avatar
wmmhello 已提交
1006
  JUMP_SPACE(sql, sqlEnd)
wmmhello's avatar
wmmhello 已提交
1007
  elements->timestamp = sql;
wmmhello's avatar
wmmhello 已提交
1008 1009
  while (sql < sqlEnd) {
    if (isspace(*sql)) {
wmmhello's avatar
wmmhello 已提交
1010 1011 1012 1013
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
1014
  elements->timestampLen = sql - elements->timestamp;
wmmhello's avatar
wmmhello 已提交
1015 1016 1017 1018

  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1019 1020
static void smlParseTelnetElement(const char **sql, const char *sqlEnd, const char **data, int32_t *len) {
  while (*sql < sqlEnd) {
X
Xiaoyu Wang 已提交
1021
    if (**sql != SPACE && !(*data)) {
1022
      *data = *sql;
X
Xiaoyu Wang 已提交
1023
    } else if (**sql == SPACE && *data) {
1024 1025 1026 1027 1028 1029 1030
      *len = *sql - *data;
      break;
    }
    (*sql)++;
  }
}

dengyihao's avatar
dengyihao 已提交
1031 1032 1033
static int32_t smlParseTelnetTags(const char *data, const char *sqlEnd, SArray *cols, char *childTableName,
                                  SHashObj *dumplicateKey, SSmlMsgBuf *msg) {
  if (!cols) return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
1034
  const char *sql = data;
X
Xiaoyu Wang 已提交
1035
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
wmmhello's avatar
wmmhello 已提交
1036 1037
  while (sql < sqlEnd) {
    JUMP_SPACE(sql, sqlEnd)
X
Xiaoyu Wang 已提交
1038
    if (*sql == '\0') break;
wmmhello's avatar
wmmhello 已提交
1039

wmmhello's avatar
wmmhello 已提交
1040
    const char *key = sql;
X
Xiaoyu Wang 已提交
1041
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
1042 1043

    // parse key
wmmhello's avatar
wmmhello 已提交
1044
    while (sql < sqlEnd) {
X
Xiaoyu Wang 已提交
1045
      if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
1046 1047 1048
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
1049
      if (*sql == EQUAL) {
wmmhello's avatar
wmmhello 已提交
1050 1051
        keyLen = sql - key;
        sql++;
1052 1053
        break;
      }
wmmhello's avatar
wmmhello 已提交
1054
      sql++;
1055
    }
wmmhello's avatar
wmmhello 已提交
1056

X
Xiaoyu Wang 已提交
1057
    if (IS_INVALID_COL_LEN(keyLen)) {
1058
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1059
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
1060
    }
X
Xiaoyu Wang 已提交
1061
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1062
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1063
      return TSDB_CODE_TSC_DUP_NAMES;
1064 1065 1066
    }

    // parse value
wmmhello's avatar
wmmhello 已提交
1067
    const char *value = sql;
X
Xiaoyu Wang 已提交
1068
    int32_t     valueLen = 0;
wmmhello's avatar
wmmhello 已提交
1069
    while (sql < sqlEnd) {
wmmhello's avatar
wmmhello 已提交
1070 1071
      // parse value
      if (*sql == SPACE) {
1072 1073
        break;
      }
wmmhello's avatar
wmmhello 已提交
1074 1075 1076 1077 1078
      if (*sql == EQUAL) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
1079
    }
wmmhello's avatar
wmmhello 已提交
1080
    valueLen = sql - value;
wmmhello's avatar
wmmhello 已提交
1081

X
Xiaoyu Wang 已提交
1082
    if (valueLen == 0) {
1083
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1084
      return TSDB_CODE_TSC_INVALID_VALUE;
1085
    }
wmmhello's avatar
wmmhello 已提交
1086

X
Xiaoyu Wang 已提交
1087 1088
    // handle child table name
    if (childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1089 1090 1091 1092 1093
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, value, (valueLen < TSDB_TABLE_NAME_LEN ? valueLen : TSDB_TABLE_NAME_LEN));
      continue;
    }

1094
    if (valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
wmmhello's avatar
wmmhello 已提交
1095 1096 1097
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }

1098 1099
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1100
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1101 1102 1103
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1104
    kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1105
    kv->type = TSDB_DATA_TYPE_NCHAR;
1106

wmmhello's avatar
wmmhello 已提交
1107
    taosArrayPush(cols, &kv);
1108 1109 1110 1111
  }

  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
1112

1113
// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
dengyihao's avatar
dengyihao 已提交
1114 1115
static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, const char *sqlEnd, SSmlTableInfo *tinfo,
                                    SArray *cols) {
X
Xiaoyu Wang 已提交
1116
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
1117 1118

  // parse metric
wmmhello's avatar
wmmhello 已提交
1119
  smlParseTelnetElement(&sql, sqlEnd, &tinfo->sTableName, &tinfo->sTableNameLen);
1120
  if (!(tinfo->sTableName) || IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
1121
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1122
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
1123 1124 1125 1126
  }

  // parse timestamp
  const char *timestamp = NULL;
X
Xiaoyu Wang 已提交
1127
  int32_t     tLen = 0;
wmmhello's avatar
wmmhello 已提交
1128
  smlParseTelnetElement(&sql, sqlEnd, &timestamp, &tLen);
1129 1130 1131 1132 1133 1134 1135 1136
  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);
wmmhello's avatar
wmmhello 已提交
1137
    return ret;
1138 1139 1140 1141
  }

  // parse value
  const char *value = NULL;
X
Xiaoyu Wang 已提交
1142
  int32_t     valueLen = 0;
wmmhello's avatar
wmmhello 已提交
1143
  smlParseTelnetElement(&sql, sqlEnd, &value, &valueLen);
1144 1145
  if (!value || valueLen == 0) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
wmmhello's avatar
wmmhello 已提交
1146
    return TSDB_CODE_TSC_INVALID_VALUE;
1147 1148 1149
  }

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1150
  if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1151 1152 1153 1154
  taosArrayPush(cols, &kv);
  kv->key = VALUE;
  kv->keyLen = VALUE_LEN;
  kv->value = value;
wmmhello's avatar
wmmhello 已提交
1155
  kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1156 1157
  if ((ret = smlParseValue(kv, &info->msgBuf)) != TSDB_CODE_SUCCESS) {
    return ret;
1158 1159 1160
  }

  // parse tags
wmmhello's avatar
wmmhello 已提交
1161
  ret = smlParseTelnetTags(sql, sqlEnd, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
1162 1163
  if (ret != TSDB_CODE_SUCCESS) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1164
    return ret;
1165 1166 1167 1168 1169
  }

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1170 1171 1172
static int32_t smlParseCols(const char *data, int32_t len, SArray *cols, char *childTableName, bool isTag,
                            SHashObj *dumplicateKey, SSmlMsgBuf *msg) {
  if (len == 0) {
wmmhello's avatar
wmmhello 已提交
1173
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1174 1175
  }

X
Xiaoyu Wang 已提交
1176
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
wmmhello's avatar
wmmhello 已提交
1177
  const char *sql = data;
X
Xiaoyu Wang 已提交
1178
  while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1179
    const char *key = sql;
X
Xiaoyu Wang 已提交
1180
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
1181

X
Xiaoyu Wang 已提交
1182
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1183
      // parse key
X
Xiaoyu Wang 已提交
1184
      if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1185 1186 1187
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
1188
      if (IS_EQUAL(sql)) {
wmmhello's avatar
wmmhello 已提交
1189 1190
        keyLen = sql - key;
        sql++;
wmmhello's avatar
wmmhello 已提交
1191 1192
        break;
      }
wmmhello's avatar
wmmhello 已提交
1193
      sql++;
wmmhello's avatar
wmmhello 已提交
1194
    }
wmmhello's avatar
wmmhello 已提交
1195

X
Xiaoyu Wang 已提交
1196
    if (IS_INVALID_COL_LEN(keyLen)) {
wmmhello's avatar
wmmhello 已提交
1197
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1198
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
wmmhello's avatar
wmmhello 已提交
1199
    }
X
Xiaoyu Wang 已提交
1200
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1201
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1202
      return TSDB_CODE_TSC_DUP_NAMES;
1203 1204
    }

wmmhello's avatar
wmmhello 已提交
1205
    // parse value
wmmhello's avatar
wmmhello 已提交
1206
    const char *value = sql;
X
Xiaoyu Wang 已提交
1207 1208 1209
    int32_t     valueLen = 0;
    bool        isInQuote = false;
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1210
      // parse value
X
Xiaoyu Wang 已提交
1211
      if (!isTag && IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
1212
        isInQuote = !isInQuote;
wmmhello's avatar
wmmhello 已提交
1213 1214
        sql++;
        continue;
wmmhello's avatar
wmmhello 已提交
1215
      }
wmmhello's avatar
wmmhello 已提交
1216
      if (!isInQuote && IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1217 1218
        break;
      }
wmmhello's avatar
wmmhello 已提交
1219 1220 1221 1222 1223
      if (!isInQuote && IS_EQUAL(sql)) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
1224
    }
wmmhello's avatar
wmmhello 已提交
1225 1226 1227
    valueLen = sql - value;
    sql++;

X
Xiaoyu Wang 已提交
1228
    if (isInQuote) {
wmmhello's avatar
wmmhello 已提交
1229 1230 1231
      smlBuildInvalidDataMsg(msg, "only one quote", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
X
Xiaoyu Wang 已提交
1232
    if (valueLen == 0) {
wmmhello's avatar
wmmhello 已提交
1233
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1234 1235
      return TSDB_CODE_SML_INVALID_DATA;
    }
wmmhello's avatar
wmmhello 已提交
1236 1237
    PROCESS_SLASH(key, keyLen)
    PROCESS_SLASH(value, valueLen)
wmmhello's avatar
wmmhello 已提交
1238

X
Xiaoyu Wang 已提交
1239 1240
    // handle child table name
    if (childTableName && childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1241 1242 1243 1244 1245
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, value, (valueLen < TSDB_TABLE_NAME_LEN ? valueLen : TSDB_TABLE_NAME_LEN));
      continue;
    }

wmmhello's avatar
wmmhello 已提交
1246
    // add kv to SSmlKv
wmmhello's avatar
wmmhello 已提交
1247
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1248 1249
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if (cols) taosArrayPush(cols, &kv);
1250

wmmhello's avatar
wmmhello 已提交
1251 1252 1253
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1254
    kv->length = valueLen;
X
Xiaoyu Wang 已提交
1255
    if (isTag) {
1256
      if (valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
1257 1258
        return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
      }
wmmhello's avatar
wmmhello 已提交
1259
      kv->type = TSDB_DATA_TYPE_NCHAR;
X
Xiaoyu Wang 已提交
1260
    } else {
wmmhello's avatar
wmmhello 已提交
1261 1262 1263
      int32_t ret = smlParseValue(kv, msg);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
wmmhello's avatar
wmmhello 已提交
1264
      }
wmmhello's avatar
wmmhello 已提交
1265 1266
    }
  }
wmmhello's avatar
wmmhello 已提交
1267

wmmhello's avatar
wmmhello 已提交
1268 1269 1270
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1271 1272
static int32_t smlUpdateMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols, SSmlMsgBuf *msg) {
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
wmmhello's avatar
wmmhello 已提交
1273
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
wmmhello's avatar
wmmhello 已提交
1274

wmmhello's avatar
wmmhello 已提交
1275
    int16_t *index = (int16_t *)taosHashGet(metaHash, kv->key, kv->keyLen);
X
Xiaoyu Wang 已提交
1276
    if (index) {
wmmhello's avatar
wmmhello 已提交
1277
      SSmlKv **value = (SSmlKv **)taosArrayGet(metaArray, *index);
X
Xiaoyu Wang 已提交
1278
      if (kv->type != (*value)->type) {
wmmhello's avatar
wmmhello 已提交
1279
        smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
wmmhello's avatar
wmmhello 已提交
1280
        return TSDB_CODE_SML_NOT_SAME_TYPE;
X
Xiaoyu Wang 已提交
1281 1282 1283
      } else {
        if (IS_VAR_DATA_TYPE(kv->type)) {  // update string len, if bigger
          if (kv->length > (*value)->length) {
wmmhello's avatar
wmmhello 已提交
1284
            *value = kv;
1285 1286 1287
          }
        }
      }
X
Xiaoyu Wang 已提交
1288
    } else {
wmmhello's avatar
wmmhello 已提交
1289 1290 1291 1292 1293
      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);
1294
    }
wmmhello's avatar
wmmhello 已提交
1295
  }
wmmhello's avatar
wmmhello 已提交
1296

wmmhello's avatar
wmmhello 已提交
1297
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1298 1299
}

X
Xiaoyu Wang 已提交
1300
static void smlInsertMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols) {
wmmhello's avatar
wmmhello 已提交
1301 1302 1303 1304
  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);
wmmhello's avatar
wmmhello 已提交
1305 1306 1307
  }
}

X
Xiaoyu Wang 已提交
1308
static SSmlTableInfo *smlBuildTableInfo() {
1309
  SSmlTableInfo *tag = (SSmlTableInfo *)taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
X
Xiaoyu Wang 已提交
1310
  if (!tag) {
1311
    return NULL;
wmmhello's avatar
wmmhello 已提交
1312
  }
1313 1314 1315 1316 1317

  tag->cols = taosArrayInit(16, POINTER_BYTES);
  if (tag->cols == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1318
  }
1319 1320 1321 1322 1323

  tag->tags = taosArrayInit(16, POINTER_BYTES);
  if (tag->tags == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1324
  }
1325 1326 1327 1328 1329
  return tag;

cleanup:
  taosMemoryFree(tag);
  return NULL;
wmmhello's avatar
wmmhello 已提交
1330 1331
}

X
Xiaoyu Wang 已提交
1332 1333 1334
static void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag) {
  if (info->dataFormat) {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1335 1336
      SArray *kvArray = (SArray *)taosArrayGetP(tag->cols, i);
      for (int j = 0; j < taosArrayGetSize(kvArray); ++j) {
wmmhello's avatar
wmmhello 已提交
1337
        SSmlKv *p = (SSmlKv *)taosArrayGetP(kvArray, j);
1338 1339 1340 1341
        taosMemoryFree(p);
      }
      taosArrayDestroy(kvArray);
    }
X
Xiaoyu Wang 已提交
1342 1343
  } else {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1344
      SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
X
Xiaoyu Wang 已提交
1345
      void    **p1 = (void **)taosHashIterate(kvHash, NULL);
1346 1347
      while (p1) {
        taosMemoryFree(*p1);
X
Xiaoyu Wang 已提交
1348
        p1 = (void **)taosHashIterate(kvHash, p1);
1349 1350 1351 1352
      }
      taosHashCleanup(kvHash);
    }
  }
X
Xiaoyu Wang 已提交
1353
  for (size_t i = 0; i < taosArrayGetSize(tag->tags); i++) {
wmmhello's avatar
wmmhello 已提交
1354 1355 1356
    SSmlKv *p = (SSmlKv *)taosArrayGetP(tag->tags, i);
    taosMemoryFree(p);
  }
1357 1358 1359 1360 1361
  taosArrayDestroy(tag->cols);
  taosArrayDestroy(tag->tags);
  taosMemoryFree(tag);
}

X
Xiaoyu Wang 已提交
1362
static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) {
1363 1364
  SArray *s1 = *(SArray **)key1;
  SArray *s2 = *(SArray **)key2;
1365 1366 1367 1368 1369 1370 1371 1372 1373 1374 1375 1376 1377
  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;
  }
}

X
Xiaoyu Wang 已提交
1378
static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
1379 1380
  SHashObj *s1 = *(SHashObj **)key1;
  SHashObj *s2 = *(SHashObj **)key2;
dengyihao's avatar
dengyihao 已提交
1381 1382 1383
  SSmlKv  **kv1pp = (SSmlKv **)taosHashGet(s1, TS, TS_LEN);
  SSmlKv  **kv2pp = (SSmlKv **)taosHashGet(s2, TS, TS_LEN);
  if (!kv1pp || !kv2pp) {
wmmhello's avatar
wmmhello 已提交
1384 1385 1386
    uError("smlKvTimeHashCompare kv is null");
    return -1;
  }
dengyihao's avatar
dengyihao 已提交
1387 1388 1389
  SSmlKv *kv1 = *kv1pp;
  SSmlKv *kv2 = *kv2pp;
  if (!kv1 || kv1->type != TSDB_DATA_TYPE_TIMESTAMP) {
wmmhello's avatar
wmmhello 已提交
1390 1391 1392
    uError("smlKvTimeHashCompare kv1");
    return -1;
  }
dengyihao's avatar
dengyihao 已提交
1393
  if (!kv2 || kv2->type != TSDB_DATA_TYPE_TIMESTAMP) {
wmmhello's avatar
wmmhello 已提交
1394 1395 1396
    uError("smlKvTimeHashCompare kv2");
    return -1;
  }
1397 1398 1399 1400 1401 1402 1403 1404 1405
  if (kv1->i < kv2->i) {
    return -1;
  } else if (kv1->i > kv2->i) {
    return 1;
  } else {
    return 0;
  }
}

1406 1407
static int32_t smlDealCols(SSmlTableInfo *oneTable, bool dataFormat, SArray *cols) {
  if (dataFormat) {
1408
    void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GT);
1409
    if (p == NULL) {
1410
      taosArrayPush(oneTable->cols, &cols);
1411
    } else {
1412
      taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
1413
    }
1414 1415 1416 1417
    return TSDB_CODE_SUCCESS;
  }

  SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1418
  if (!kvHash) {
1419
    uError("SML:smlDealCols failed to allocate memory");
S
Shengliang Guan 已提交
1420
    return TSDB_CODE_OUT_OF_MEMORY;
1421
  }
X
Xiaoyu Wang 已提交
1422
  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
1423
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
wmmhello's avatar
wmmhello 已提交
1424
    taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
1425 1426
  }

1427
  void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GT);
1428
  if (p == NULL) {
1429
    taosArrayPush(oneTable->cols, &kvHash);
1430
  } else {
1431
    taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &kvHash);
1432
  }
1433 1434 1435
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1436 1437 1438
static SSmlSTableMeta *smlBuildSTableMeta() {
  SSmlSTableMeta *meta = (SSmlSTableMeta *)taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if (!meta) {
1439 1440 1441 1442 1443 1444 1445 1446
    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;
  }

wmmhello's avatar
wmmhello 已提交
1447 1448
  meta->colHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (meta->colHash == NULL) {
1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470
    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;
}

X
Xiaoyu Wang 已提交
1471
static void smlDestroySTableMeta(SSmlSTableMeta *meta) {
1472
  taosHashCleanup(meta->tagHash);
wmmhello's avatar
wmmhello 已提交
1473
  taosHashCleanup(meta->colHash);
1474 1475 1476
  taosArrayDestroy(meta->tags);
  taosArrayDestroy(meta->cols);
  taosMemoryFree(meta->tableMeta);
wmmhello's avatar
wmmhello 已提交
1477
  taosMemoryFree(meta);
1478 1479
}

wmmhello's avatar
wmmhello 已提交
1480
static void smlDestroyCols(SArray *cols) {
1481 1482
  if (!cols) return;
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
wmmhello's avatar
wmmhello 已提交
1483
    void *kv = taosArrayGetP(cols, i);
1484 1485 1486 1487
    taosMemoryFree(kv);
  }
}

X
Xiaoyu Wang 已提交
1488 1489
static void smlDestroyInfo(SSmlHandle *info) {
  if (!info) return;
1490 1491 1492 1493
  qDestroyQuery(info->pQuery);
  smlDestroyHandle(info->exec);

  // destroy info->childTables
X
Xiaoyu Wang 已提交
1494
  void **p1 = (void **)taosHashIterate(info->childTables, NULL);
1495
  while (p1) {
X
Xiaoyu Wang 已提交
1496 1497
    smlDestroyTableInfo(info, (SSmlTableInfo *)(*p1));
    p1 = (void **)taosHashIterate(info->childTables, p1);
1498 1499 1500 1501
  }
  taosHashCleanup(info->childTables);

  // destroy info->superTables
X
Xiaoyu Wang 已提交
1502
  p1 = (void **)taosHashIterate(info->superTables, NULL);
1503
  while (p1) {
X
Xiaoyu Wang 已提交
1504 1505
    smlDestroySTableMeta((SSmlSTableMeta *)(*p1));
    p1 = (void **)taosHashIterate(info->superTables, p1);
1506 1507 1508 1509 1510 1511
  }
  taosHashCleanup(info->superTables);

  // destroy info->pVgHash
  taosHashCleanup(info->pVgHash);
  taosHashCleanup(info->dumplicateKey);
X
Xiaoyu Wang 已提交
1512
  if (!info->dataFormat) {
wmmhello's avatar
wmmhello 已提交
1513 1514
    taosArrayDestroy(info->colsContainer);
  }
wmmhello's avatar
wmmhello 已提交
1515
  destroyRequest(info->pRequest);
wmmhello's avatar
wmmhello 已提交
1516 1517

  cJSON_Delete(info->root);
1518 1519 1520
  taosMemoryFreeClear(info);
}

1521 1522 1523
static SSmlHandle *smlBuildSmlInfo(STscObj *pTscObj, SRequestObj *request, SMLProtocolType protocol, int8_t precision) {
  int32_t     code = TSDB_CODE_SUCCESS;
  SSmlHandle *info = (SSmlHandle *)taosMemoryCalloc(1, sizeof(SSmlHandle));
1524 1525 1526
  if (NULL == info) {
    return NULL;
  }
X
Xiaoyu Wang 已提交
1527
  info->id = smlGenId();
1528

1529
  info->pQuery = (SQuery *)nodesMakeNode(QUERY_NODE_QUERY);
1530
  if (NULL == info->pQuery) {
X
Xiaoyu Wang 已提交
1531
    uError("SML:0x%" PRIx64 " create info->pQuery error", info->id);
1532 1533
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
1534
  info->pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
1535
  info->pQuery->haveResultSet = false;
X
Xiaoyu Wang 已提交
1536 1537 1538 1539
  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);
1540 1541 1542
    goto cleanup;
  }

1543 1544
  if (pTscObj) {
    info->taos = pTscObj;
1545 1546 1547 1548 1549
    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;
    }
wmmhello's avatar
wmmhello 已提交
1550
  }
1551

X
Xiaoyu Wang 已提交
1552 1553 1554
  info->precision = precision;
  info->protocol = protocol;
  if (protocol == TSDB_SML_LINE_PROTOCOL) {
1555
    info->dataFormat = tsSmlDataFormat;
X
Xiaoyu Wang 已提交
1556
  } else {
1557 1558
    info->dataFormat = true;
  }
1559

1560
  if (request) {
1561 1562 1563
    info->pRequest = request;
    info->msgBuf.buf = info->pRequest->msgBuf;
    info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
1564
    info->pRequest->stmtType = info->pQuery->pRoot->type;
1565
  }
1566

X
Xiaoyu Wang 已提交
1567
  info->exec = smlInitHandle(info->pQuery);
1568 1569
  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);
X
Xiaoyu Wang 已提交
1570
  info->pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
1571 1572

  info->dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1573
  if (!info->dataFormat) {
1574
    info->colsContainer = taosArrayInit(32, POINTER_BYTES);
X
Xiaoyu Wang 已提交
1575 1576
    if (NULL == info->colsContainer) {
      uError("SML:0x%" PRIx64 " create info failed", info->id);
1577 1578 1579
      goto cleanup;
    }
  }
X
Xiaoyu Wang 已提交
1580 1581 1582
  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);
1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595
    goto cleanup;
  }

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

/************* TSDB_SML_JSON_PROTOCOL function start **************/
static int32_t smlParseMetricFromJSON(SSmlHandle *info, cJSON *root, SSmlTableInfo *tinfo) {
  cJSON *metric = cJSON_GetObjectItem(root, "metric");
  if (!cJSON_IsString(metric)) {
X
Xiaoyu Wang 已提交
1596
    return TSDB_CODE_TSC_INVALID_JSON;
1597 1598 1599
  }

  tinfo->sTableNameLen = strlen(metric->valuestring);
1600
  if (IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
X
Xiaoyu Wang 已提交
1601
    uError("OTD:0x%" PRIx64 " Metric lenght is 0 or large than 192", info->id);
1602 1603 1604
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
  }

wmmhello's avatar
wmmhello 已提交
1605 1606
  tinfo->sTableName = metric->valuestring;
  return TSDB_CODE_SUCCESS;
1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625
}

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;
X
Xiaoyu Wang 已提交
1626
  if (smlDoubleToInt64OverFlow(timeDouble)) {
1627
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1628
    return TSDB_CODE_INVALID_TIMESTAMP;
1629
  }
wmmhello's avatar
wmmhello 已提交
1630 1631 1632 1633 1634 1635 1636 1637

  if (timeDouble == 0) {
    *tsVal = taosGetTimestampNs();
    return TSDB_CODE_SUCCESS;
  }

  if (timeDouble < 0) {
    return TSDB_CODE_INVALID_TIMESTAMP;
1638 1639
  }

1640
  *tsVal = timeDouble;
1641
  size_t typeLen = strlen(type->valuestring);
wmmhello's avatar
wmmhello 已提交
1642
  if (typeLen == 1 && (type->valuestring[0] == 's' || type->valuestring[0] == 'S')) {
X
Xiaoyu Wang 已提交
1643
    // seconds
1644 1645
    *tsVal = *tsVal * NANOSECOND_PER_SEC;
    timeDouble = timeDouble * NANOSECOND_PER_SEC;
X
Xiaoyu Wang 已提交
1646
    if (smlDoubleToInt64OverFlow(timeDouble)) {
1647
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1648
      return TSDB_CODE_INVALID_TIMESTAMP;
1649
    }
wmmhello's avatar
wmmhello 已提交
1650
  } else if (typeLen == 2 && (type->valuestring[1] == 's' || type->valuestring[1] == 'S')) {
1651 1652
    switch (type->valuestring[0]) {
      case 'm':
wmmhello's avatar
wmmhello 已提交
1653
      case 'M':
X
Xiaoyu Wang 已提交
1654
        // milliseconds
1655 1656
        *tsVal = *tsVal * NANOSECOND_PER_MSEC;
        timeDouble = timeDouble * NANOSECOND_PER_MSEC;
X
Xiaoyu Wang 已提交
1657
        if (smlDoubleToInt64OverFlow(timeDouble)) {
1658
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1659
          return TSDB_CODE_INVALID_TIMESTAMP;
1660 1661 1662
        }
        break;
      case 'u':
wmmhello's avatar
wmmhello 已提交
1663
      case 'U':
X
Xiaoyu Wang 已提交
1664
        // microseconds
1665 1666
        *tsVal = *tsVal * NANOSECOND_PER_USEC;
        timeDouble = timeDouble * NANOSECOND_PER_USEC;
X
Xiaoyu Wang 已提交
1667
        if (smlDoubleToInt64OverFlow(timeDouble)) {
1668
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1669
          return TSDB_CODE_INVALID_TIMESTAMP;
1670 1671 1672
        }
        break;
      case 'n':
wmmhello's avatar
wmmhello 已提交
1673
      case 'N':
1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694
        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) {
X
Xiaoyu Wang 已提交
1695
  // Timestamp must be the first KV to parse
1696 1697 1698 1699
  int64_t tsVal = 0;

  cJSON *timestamp = cJSON_GetObjectItem(root, "timestamp");
  if (cJSON_IsNumber(timestamp)) {
X
Xiaoyu Wang 已提交
1700
    // timestamp value 0 indicates current system time
1701
    double timeDouble = timestamp->valuedouble;
X
Xiaoyu Wang 已提交
1702
    if (smlDoubleToInt64OverFlow(timeDouble)) {
1703
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1704
      return TSDB_CODE_INVALID_TIMESTAMP;
1705
    }
wmmhello's avatar
wmmhello 已提交
1706

X
Xiaoyu Wang 已提交
1707
    if (timeDouble < 0) {
wmmhello's avatar
wmmhello 已提交
1708
      return TSDB_CODE_INVALID_TIMESTAMP;
1709
    }
1710

1711
    uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);
1712
    tsVal = (int64_t)timeDouble;
1713
    if (tsLen == TSDB_TIME_PRECISION_SEC_DIGITS) {
1714 1715
      tsVal = tsVal * NANOSECOND_PER_SEC;
      timeDouble = timeDouble * NANOSECOND_PER_SEC;
X
Xiaoyu Wang 已提交
1716
      if (smlDoubleToInt64OverFlow(timeDouble)) {
1717
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1718
        return TSDB_CODE_INVALID_TIMESTAMP;
1719 1720
      }
    } else if (tsLen == TSDB_TIME_PRECISION_MILLI_DIGITS) {
1721 1722
      tsVal = tsVal * NANOSECOND_PER_MSEC;
      timeDouble = timeDouble * NANOSECOND_PER_MSEC;
X
Xiaoyu Wang 已提交
1723
      if (smlDoubleToInt64OverFlow(timeDouble)) {
1724
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1725
        return TSDB_CODE_INVALID_TIMESTAMP;
1726
      }
X
Xiaoyu Wang 已提交
1727
    } else if (timeDouble == 0) {
wmmhello's avatar
wmmhello 已提交
1728
      tsVal = taosGetTimestampNs();
X
Xiaoyu Wang 已提交
1729
    } else {
wmmhello's avatar
wmmhello 已提交
1730
      return TSDB_CODE_INVALID_TIMESTAMP;
1731 1732 1733 1734
    }
  } else if (cJSON_IsObject(timestamp)) {
    int32_t ret = smlParseTSFromJSONObj(info, timestamp, &tsVal);
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
1735
      uError("SML:0x%" PRIx64 " Failed to parse timestamp from JSON Obj", info->id);
1736 1737 1738 1739
      return ret;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
1740 1741
  }

wmmhello's avatar
wmmhello 已提交
1742
  // add ts to
wmmhello's avatar
wmmhello 已提交
1743
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1744
  if (!kv) {
wmmhello's avatar
wmmhello 已提交
1745 1746 1747
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  kv->key = TS;
1748 1749
  kv->keyLen = TS_LEN;
  kv->i = tsVal;
wmmhello's avatar
wmmhello 已提交
1750
  kv->type = TSDB_DATA_TYPE_TIMESTAMP;
wmmhello's avatar
wmmhello 已提交
1751
  kv->length = (int16_t)tDataTypes[kv->type].bytes;
wmmhello's avatar
wmmhello 已提交
1752
  taosArrayPush(cols, &kv);
wmmhello's avatar
wmmhello 已提交
1753 1754 1755
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1756
static int32_t smlConvertJSONBool(SSmlKv *pVal, char *typeStr, cJSON *value) {
1757 1758 1759 1760 1761 1762 1763
  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;
wmmhello's avatar
wmmhello 已提交
1764

1765 1766
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
1767

X
Xiaoyu Wang 已提交
1768 1769 1770
static int32_t smlConvertJSONNumber(SSmlKv *pVal, char *typeStr, cJSON *value) {
  // tinyint
  if (strcasecmp(typeStr, "i8") == 0 || strcasecmp(typeStr, "tinyint") == 0) {
1771 1772 1773 1774 1775 1776 1777 1778 1779
    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;
  }
X
Xiaoyu Wang 已提交
1780 1781
  // smallint
  if (strcasecmp(typeStr, "i16") == 0 || strcasecmp(typeStr, "smallint") == 0) {
1782 1783 1784 1785 1786 1787 1788 1789 1790
    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;
  }
X
Xiaoyu Wang 已提交
1791 1792
  // int
  if (strcasecmp(typeStr, "i32") == 0 || strcasecmp(typeStr, "int") == 0) {
1793 1794 1795 1796 1797 1798 1799 1800 1801
    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;
  }
X
Xiaoyu Wang 已提交
1802 1803
  // bigint
  if (strcasecmp(typeStr, "i64") == 0 || strcasecmp(typeStr, "bigint") == 0) {
1804 1805
    pVal->type = TSDB_DATA_TYPE_BIGINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
X
Xiaoyu Wang 已提交
1806
    if (value->valuedouble >= (double)INT64_MAX) {
1807
      pVal->i = INT64_MAX;
X
Xiaoyu Wang 已提交
1808
    } else if (value->valuedouble <= (double)INT64_MIN) {
1809
      pVal->i = INT64_MIN;
X
Xiaoyu Wang 已提交
1810
    } else {
1811
      pVal->i = value->valuedouble;
1812 1813 1814
    }
    return TSDB_CODE_SUCCESS;
  }
X
Xiaoyu Wang 已提交
1815 1816
  // float
  if (strcasecmp(typeStr, "f32") == 0 || strcasecmp(typeStr, "float") == 0) {
1817 1818 1819
    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;
wmmhello's avatar
wmmhello 已提交
1820
    }
1821 1822 1823 1824 1825
    pVal->type = TSDB_DATA_TYPE_FLOAT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    pVal->f = value->valuedouble;
    return TSDB_CODE_SUCCESS;
  }
X
Xiaoyu Wang 已提交
1826 1827
  // double
  if (strcasecmp(typeStr, "f64") == 0 || strcasecmp(typeStr, "double") == 0) {
1828 1829 1830 1831
    pVal->type = TSDB_DATA_TYPE_DOUBLE;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
    pVal->d = value->valuedouble;
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1832 1833
  }

X
Xiaoyu Wang 已提交
1834
  // if reach here means type is unsupported
1835 1836 1837
  uError("OTD:invalid type(%s) for JSON Number", typeStr);
  return TSDB_CODE_TSC_INVALID_JSON_TYPE;
}
1838

X
Xiaoyu Wang 已提交
1839
static int32_t smlConvertJSONString(SSmlKv *pVal, char *typeStr, cJSON *value) {
1840 1841 1842 1843 1844 1845 1846 1847 1848
  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);
wmmhello's avatar
wmmhello 已提交
1849

1850
  if (pVal->type == TSDB_DATA_TYPE_BINARY && pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
wmmhello's avatar
wmmhello 已提交
1851 1852
    return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
  }
1853 1854
  if (pVal->type == TSDB_DATA_TYPE_NCHAR &&
      pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
wmmhello's avatar
wmmhello 已提交
1855 1856 1857
    return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
  }

wmmhello's avatar
wmmhello 已提交
1858 1859
  pVal->value = value->valuestring;
  return TSDB_CODE_SUCCESS;
1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885
}

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;
wmmhello's avatar
wmmhello 已提交
1886
      }
1887
      break;
wmmhello's avatar
wmmhello 已提交
1888
    }
1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901 1902 1903 1904
    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;
wmmhello's avatar
wmmhello 已提交
1905
  }
1906 1907

  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1908 1909
}

1910 1911 1912 1913 1914 1915 1916 1917
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;
wmmhello's avatar
wmmhello 已提交
1918
    }
1919 1920 1921 1922 1923 1924 1925 1926 1927 1928
    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
       */
wmmhello's avatar
wmmhello 已提交
1929

X
Xiaoyu Wang 已提交
1930
      char *tsDefaultJSONStrType = "nchar";  // todo
1931 1932
      smlConvertJSONString(kv, tsDefaultJSONStrType, root);
      break;
wmmhello's avatar
wmmhello 已提交
1933
    }
1934 1935 1936 1937 1938 1939 1940 1941 1942 1943
    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;
wmmhello's avatar
wmmhello 已提交
1944
  }
1945 1946 1947 1948 1949

  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseColsFromJSON(cJSON *root, SArray *cols) {
dengyihao's avatar
dengyihao 已提交
1950
  if (!cols) return TSDB_CODE_OUT_OF_MEMORY;
1951 1952 1953 1954 1955 1956
  cJSON *metricVal = cJSON_GetObjectItem(root, "value");
  if (metricVal == NULL) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1957
  if (!kv) {
1958 1959
    return TSDB_CODE_OUT_OF_MEMORY;
  }
wmmhello's avatar
wmmhello 已提交
1960
  taosArrayPush(cols, &kv);
1961 1962 1963 1964 1965 1966 1967 1968

  kv->key = VALUE;
  kv->keyLen = VALUE_LEN;
  int32_t ret = smlParseValueFromJSON(metricVal, kv);
  if (ret != TSDB_CODE_SUCCESS) {
    return ret;
  }
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1969 1970
}

X
Xiaoyu Wang 已提交
1971 1972
static int32_t smlParseTagsFromJSON(cJSON *root, SArray *pKVs, char *childTableName, SHashObj *dumplicateKey,
                                    SSmlMsgBuf *msg) {
1973
  int32_t ret = TSDB_CODE_SUCCESS;
dengyihao's avatar
dengyihao 已提交
1974
  if (!pKVs) {
wmmhello's avatar
wmmhello 已提交
1975 1976
    return TSDB_CODE_OUT_OF_MEMORY;
  }
1977 1978 1979 1980
  cJSON *tags = cJSON_GetObjectItem(root, "tags");
  if (tags == NULL || tags->type != cJSON_Object) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }
wmmhello's avatar
wmmhello 已提交
1981

X
Xiaoyu Wang 已提交
1982
  size_t  childTableNameLen = strlen(tsSmlChildTableName);
1983 1984 1985 1986 1987
  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;
wmmhello's avatar
wmmhello 已提交
1988
    }
1989 1990 1991 1992 1993
    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;
    }
X
Xiaoyu Wang 已提交
1994
    // check duplicate keys
1995
    if (smlCheckDuplicateKey(tag->string, keyLen, dumplicateKey)) {
wmmhello's avatar
wmmhello 已提交
1996
      return TSDB_CODE_TSC_DUP_NAMES;
wmmhello's avatar
wmmhello 已提交
1997 1998
    }

X
Xiaoyu Wang 已提交
1999 2000
    // handle child table name
    if (childTableNameLen != 0 && strcmp(tag->string, tsSmlChildTableName) == 0) {
2001 2002 2003 2004 2005
      if (!cJSON_IsString(tag)) {
        uError("OTD:ID must be JSON string");
        return TSDB_CODE_TSC_INVALID_JSON;
      }
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
wmmhello's avatar
wmmhello 已提交
2006
      tstrncpy(childTableName, tag->valuestring, TSDB_TABLE_NAME_LEN);
2007 2008 2009
      continue;
    }

2010 2011
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
2012
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2013
    taosArrayPush(pKVs, &kv);
2014

X
Xiaoyu Wang 已提交
2015
    // key
2016
    kv->keyLen = keyLen;
wmmhello's avatar
wmmhello 已提交
2017 2018
    kv->key = tag->string;

X
Xiaoyu Wang 已提交
2019
    // value
2020 2021 2022 2023
    ret = smlParseValueFromJSON(tag, kv);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
wmmhello's avatar
wmmhello 已提交
2024 2025
  }

2026
  return ret;
wmmhello's avatar
wmmhello 已提交
2027 2028
}

2029 2030
static int32_t smlParseJSONString(SSmlHandle *info, cJSON *root, SSmlTableInfo *tinfo, SArray *cols) {
  int32_t ret = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2031

2032
  if (!cJSON_IsObject(root)) {
X
Xiaoyu Wang 已提交
2033
    uError("OTD:0x%" PRIx64 " data point needs to be JSON object", info->id);
2034
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
2035
  }
2036

2037
  int32_t size = cJSON_GetArraySize(root);
X
Xiaoyu Wang 已提交
2038
  // outmost json fields has to be exactly 4
2039
  if (size != OTD_JSON_FIELDS_NUM) {
X
Xiaoyu Wang 已提交
2040
    uError("OTD:0x%" PRIx64 " Invalid number of JSON fields in data point %d", info->id, size);
2041
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
2042
  }
2043

X
Xiaoyu Wang 已提交
2044
  // Parse metric
2045 2046
  ret = smlParseMetricFromJSON(info, root, tinfo);
  if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2047
    uError("OTD:0x%" PRIx64 " Unable to parse metric from JSON payload", info->id);
2048
    return ret;
wmmhello's avatar
wmmhello 已提交
2049
  }
X
Xiaoyu Wang 已提交
2050
  uDebug("OTD:0x%" PRIx64 " Parse metric from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2051

X
Xiaoyu Wang 已提交
2052
  // Parse timestamp
2053 2054
  ret = smlParseTSFromJSON(info, root, cols);
  if (ret) {
X
Xiaoyu Wang 已提交
2055
    uError("OTD:0x%" PRIx64 " Unable to parse timestamp from JSON payload", info->id);
2056
    return ret;
wmmhello's avatar
wmmhello 已提交
2057
  }
X
Xiaoyu Wang 已提交
2058
  uDebug("OTD:0x%" PRIx64 " Parse timestamp from JSON payload finished", info->id);
2059

X
Xiaoyu Wang 已提交
2060
  // Parse metric value
2061 2062
  ret = smlParseColsFromJSON(root, cols);
  if (ret) {
X
Xiaoyu Wang 已提交
2063
    uError("OTD:0x%" PRIx64 " Unable to parse metric value from JSON payload", info->id);
2064
    return ret;
2065
  }
X
Xiaoyu Wang 已提交
2066
  uDebug("OTD:0x%" PRIx64 " Parse metric value from JSON payload finished", info->id);
2067

X
Xiaoyu Wang 已提交
2068
  // Parse tags
2069
  ret = smlParseTagsFromJSON(root, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
2070
  if (ret) {
X
Xiaoyu Wang 已提交
2071
    uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
2072
    return ret;
2073
  }
X
Xiaoyu Wang 已提交
2074
  uDebug("OTD:0x%" PRIx64 " Parse tags from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2075

2076
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2077
}
2078
/************* TSDB_SML_JSON_PROTOCOL function end **************/
wmmhello's avatar
wmmhello 已提交
2079

wmmhello's avatar
wmmhello 已提交
2080
static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql, const int len) {
wmmhello's avatar
wmmhello 已提交
2081
  SSmlLineInfo elements = {0};
2082
  uDebug("SML:0x%" PRIx64 " smlParseInfluxLine raw:%d, len:%d, sql:%s", info->id, info->isRawLine, len, (info->isRawLine ? "rawdata" : sql));
2083

wmmhello's avatar
wmmhello 已提交
2084
  int ret = smlParseInfluxString(sql, sql + len, &elements, &info->msgBuf);
X
Xiaoyu Wang 已提交
2085 2086
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseInfluxLine failed", info->id);
wmmhello's avatar
wmmhello 已提交
2087 2088 2089
    return ret;
  }

2090
  SArray *cols = NULL;
X
Xiaoyu Wang 已提交
2091
  if (info->dataFormat) {  // if dataFormat, cols need new memory to save data
2092 2093
    cols = taosArrayInit(16, POINTER_BYTES);
    if (cols == NULL) {
X
Xiaoyu Wang 已提交
2094
      uError("SML:0x%" PRIx64 " smlParseInfluxLine failed to allocate memory", info->id);
S
Shengliang Guan 已提交
2095
      return TSDB_CODE_OUT_OF_MEMORY;
2096
    }
X
Xiaoyu Wang 已提交
2097
  } else {  // if dataFormat is false, cols do not need to save data, there is another new memory to save data
2098
    cols = info->colsContainer;
wmmhello's avatar
wmmhello 已提交
2099
  }
wmmhello's avatar
wmmhello 已提交
2100

wmmhello's avatar
wmmhello 已提交
2101
  ret = smlParseTS(info, elements.timestamp, elements.timestampLen, cols);
X
Xiaoyu Wang 已提交
2102 2103 2104
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseTS failed", info->id);
    if (info->dataFormat) taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2105 2106
    return ret;
  }
2107
  ret = smlParseCols(elements.cols, elements.colsLen, cols, NULL, false, info->dumplicateKey, &info->msgBuf);
X
Xiaoyu Wang 已提交
2108 2109
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseCols parse cloums fields failed", info->id);
wmmhello's avatar
wmmhello 已提交
2110
    smlDestroyCols(cols);
X
Xiaoyu Wang 已提交
2111
    if (info->dataFormat) taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2112 2113 2114
    return ret;
  }

X
Xiaoyu Wang 已提交
2115 2116 2117 2118 2119
  bool            hasTable = true;
  SSmlTableInfo  *tinfo = NULL;
  SSmlTableInfo **oneTable =
      (SSmlTableInfo **)taosHashGet(info->childTables, elements.measure, elements.measureTagsLen);
  if (!oneTable) {
2120
    tinfo = smlBuildTableInfo();
X
Xiaoyu Wang 已提交
2121
    if (!tinfo) {
wmmhello's avatar
wmmhello 已提交
2122
      smlDestroyCols(cols);
wmmhello's avatar
wmmhello 已提交
2123
      if (info->dataFormat) taosArrayDestroy(cols);
S
Shengliang Guan 已提交
2124
      return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2125
    }
2126
    taosHashPut(info->childTables, elements.measure, elements.measureTagsLen, &tinfo, POINTER_BYTES);
2127
    oneTable = &tinfo;
2128 2129 2130 2131
    hasTable = false;
  }

  ret = smlDealCols(*oneTable, info->dataFormat, cols);
X
Xiaoyu Wang 已提交
2132
  if (ret != TSDB_CODE_SUCCESS) {
2133 2134
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2135

X
Xiaoyu Wang 已提交
2136 2137 2138 2139 2140
  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);
wmmhello's avatar
wmmhello 已提交
2141 2142 2143
      return ret;
    }

X
Xiaoyu Wang 已提交
2144
    if (taosArrayGetSize((*oneTable)->tags) > TSDB_MAX_TAGS) {
wmmhello's avatar
wmmhello 已提交
2145
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
wmmhello's avatar
wmmhello 已提交
2146
      return TSDB_CODE_PAR_INVALID_TAGS_NUM;
wmmhello's avatar
wmmhello 已提交
2147 2148
    }

2149 2150 2151 2152 2153
    if (taosArrayGetSize(cols) + taosArrayGetSize((*oneTable)->tags) > TSDB_MAX_COLUMNS) {
      smlBuildInvalidDataMsg(&info->msgBuf, "too many columns than 4096", NULL);
      return TSDB_CODE_PAR_TOO_MANY_COLUMNS;
    }

2154 2155
    (*oneTable)->sTableName = elements.measure;
    (*oneTable)->sTableNameLen = elements.measureLen;
X
Xiaoyu Wang 已提交
2156 2157 2158
    if (strlen((*oneTable)->childTableName) == 0) {
      RandTableName rName = {(*oneTable)->tags, (*oneTable)->sTableName, (uint8_t)(*oneTable)->sTableNameLen,
                             (*oneTable)->childTableName, 0};
2159 2160 2161

      buildChildTableName(&rName);
      (*oneTable)->uid = rName.uid;
X
Xiaoyu Wang 已提交
2162 2163
    } else {
      (*oneTable)->uid = *(uint64_t *)((*oneTable)->childTableName);
2164
    }
2165
  }
wmmhello's avatar
wmmhello 已提交
2166

X
Xiaoyu Wang 已提交
2167 2168
  SSmlSTableMeta **tableMeta = (SSmlSTableMeta **)taosHashGet(info->superTables, elements.measure, elements.measureLen);
  if (tableMeta) {  // update meta
wmmhello's avatar
wmmhello 已提交
2169
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
2170
    if (!hasTable && ret == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2171 2172
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
wmmhello's avatar
wmmhello 已提交
2173
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2174
      uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
wmmhello's avatar
wmmhello 已提交
2175
      return ret;
wmmhello's avatar
wmmhello 已提交
2176
    }
X
Xiaoyu Wang 已提交
2177
  } else {
2178
    SSmlSTableMeta *meta = smlBuildSTableMeta();
wmmhello's avatar
wmmhello 已提交
2179 2180
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
2181
    taosHashPut(info->superTables, elements.measure, elements.measureLen, &meta, POINTER_BYTES);
wmmhello's avatar
wmmhello 已提交
2182
  }
2183

X
Xiaoyu Wang 已提交
2184
  if (!info->dataFormat) {
2185 2186 2187
    taosArrayClear(info->colsContainer);
  }
  taosHashClear(info->dumplicateKey);
wmmhello's avatar
wmmhello 已提交
2188 2189 2190
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
2191
static int32_t smlParseTelnetLine(SSmlHandle *info, void *data, const int len) {
X
Xiaoyu Wang 已提交
2192
  int            ret = TSDB_CODE_SUCCESS;
2193
  SSmlTableInfo *tinfo = smlBuildTableInfo();
X
Xiaoyu Wang 已提交
2194
  if (!tinfo) {
S
Shengliang Guan 已提交
2195
    return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2196 2197
  }

2198 2199
  SArray *cols = taosArrayInit(16, POINTER_BYTES);
  if (cols == NULL) {
X
Xiaoyu Wang 已提交
2200
    uError("SML:0x%" PRIx64 " smlParseTelnetLine failed to allocate memory", info->id);
S
Shengliang Guan 已提交
2201
    return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2202 2203
  }

X
Xiaoyu Wang 已提交
2204
  if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
dengyihao's avatar
dengyihao 已提交
2205
    ret = smlParseTelnetString(info, (const char *)data, (char *)data + len, tinfo, cols);
X
Xiaoyu Wang 已提交
2206
  } else if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
2207
    ret = smlParseJSONString(info, (cJSON *)data, tinfo, cols);
X
Xiaoyu Wang 已提交
2208
  } else {
2209 2210
    ASSERT(0);
  }
X
Xiaoyu Wang 已提交
2211 2212
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseTelnetLine failed", info->id);
wmmhello's avatar
wmmhello 已提交
2213
    smlDestroyTableInfo(info, tinfo);
wmmhello's avatar
wmmhello 已提交
2214
    smlDestroyCols(cols);
2215 2216
    taosArrayDestroy(cols);
    return ret;
wmmhello's avatar
wmmhello 已提交
2217
  }
wmmhello's avatar
wmmhello 已提交
2218

X
Xiaoyu Wang 已提交
2219
  if (taosArrayGetSize(tinfo->tags) <= 0 || taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS) {
2220
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate tags length:[1,128]", NULL);
wmmhello's avatar
wmmhello 已提交
2221
    smlDestroyTableInfo(info, tinfo);
wmmhello's avatar
wmmhello 已提交
2222
    smlDestroyCols(cols);
wmmhello's avatar
wmmhello 已提交
2223
    taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2224
    return TSDB_CODE_PAR_INVALID_TAGS_NUM;
wmmhello's avatar
wmmhello 已提交
2225
  }
2226 2227
  taosHashClear(info->dumplicateKey);

X
Xiaoyu Wang 已提交
2228 2229
  if (strlen(tinfo->childTableName) == 0) {
    RandTableName rName = {tinfo->tags, tinfo->sTableName, (uint8_t)tinfo->sTableNameLen, tinfo->childTableName, 0};
2230 2231
    buildChildTableName(&rName);
    tinfo->uid = rName.uid;
X
Xiaoyu Wang 已提交
2232 2233
  } else {
    tinfo->uid = *(uint64_t *)(tinfo->childTableName);  // generate uid by name simple
2234 2235
  }

X
Xiaoyu Wang 已提交
2236 2237 2238 2239
  bool            hasTable = true;
  SSmlTableInfo **oneTable =
      (SSmlTableInfo **)taosHashGet(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName));
  if (!oneTable) {
2240
    taosHashPut(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName), &tinfo, POINTER_BYTES);
2241
    oneTable = &tinfo;
2242
    hasTable = false;
X
Xiaoyu Wang 已提交
2243
  } else {
wmmhello's avatar
wmmhello 已提交
2244
    smlDestroyTableInfo(info, tinfo);
wmmhello's avatar
wmmhello 已提交
2245
  }
wmmhello's avatar
wmmhello 已提交
2246

2247
  taosArrayPush((*oneTable)->cols, &cols);
X
Xiaoyu Wang 已提交
2248 2249 2250
  SSmlSTableMeta **tableMeta =
      (SSmlSTableMeta **)taosHashGet(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen);
  if (tableMeta) {  // update meta
wmmhello's avatar
wmmhello 已提交
2251
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
2252
    if (!hasTable && ret == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2253 2254
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
wmmhello's avatar
wmmhello 已提交
2255
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2256
      uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
wmmhello's avatar
wmmhello 已提交
2257
      return ret;
2258
    }
X
Xiaoyu Wang 已提交
2259
  } else {
2260
    SSmlSTableMeta *meta = smlBuildSTableMeta();
wmmhello's avatar
wmmhello 已提交
2261 2262
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
2263
    taosHashPut(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen, &meta, POINTER_BYTES);
wmmhello's avatar
wmmhello 已提交
2264 2265
  }

2266 2267
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
2268

X
Xiaoyu Wang 已提交
2269
static int32_t smlParseJSON(SSmlHandle *info, char *payload) {
2270 2271
  int32_t payloadNum = 0;
  int32_t ret = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2272

2273
  if (payload == NULL) {
X
Xiaoyu Wang 已提交
2274
    uError("SML:0x%" PRIx64 " empty JSON Payload", info->id);
2275
    return TSDB_CODE_TSC_INVALID_JSON;
2276
  }
2277

wmmhello's avatar
wmmhello 已提交
2278 2279
  info->root = cJSON_Parse(payload);
  if (info->root == NULL) {
X
Xiaoyu Wang 已提交
2280
    uError("SML:0x%" PRIx64 " parse json failed:%s", info->id, payload);
2281 2282
    return TSDB_CODE_TSC_INVALID_JSON;
  }
X
Xiaoyu Wang 已提交
2283
  // multiple data points must be sent in JSON array
wmmhello's avatar
wmmhello 已提交
2284
  if (cJSON_IsObject(info->root)) {
2285
    payloadNum = 1;
wmmhello's avatar
wmmhello 已提交
2286 2287
  } else if (cJSON_IsArray(info->root)) {
    payloadNum = cJSON_GetArraySize(info->root);
2288
  } else {
X
Xiaoyu Wang 已提交
2289
    uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2290 2291
    ret = TSDB_CODE_TSC_INVALID_JSON;
    goto end;
wmmhello's avatar
wmmhello 已提交
2292
  }
wmmhello's avatar
wmmhello 已提交
2293

2294
  for (int32_t i = 0; i < payloadNum; ++i) {
wmmhello's avatar
wmmhello 已提交
2295
    cJSON *dataPoint = (payloadNum == 1 && cJSON_IsObject(info->root)) ? info->root : cJSON_GetArrayItem(info->root, i);
wmmhello's avatar
wmmhello 已提交
2296
    ret = smlParseTelnetLine(info, dataPoint, -1);
X
Xiaoyu Wang 已提交
2297 2298
    if (ret != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2299 2300 2301 2302 2303 2304
      goto end;
    }
  }

end:
  return ret;
wmmhello's avatar
wmmhello 已提交
2305
}
2306

X
Xiaoyu Wang 已提交
2307
static int32_t smlInsertData(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
2308 2309
  int32_t code = TSDB_CODE_SUCCESS;

X
Xiaoyu Wang 已提交
2310
  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL);
wmmhello's avatar
wmmhello 已提交
2311
  while (oneTable) {
X
Xiaoyu Wang 已提交
2312
    SSmlTableInfo *tableData = *oneTable;
wmmhello's avatar
wmmhello 已提交
2313 2314

    SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
wmmhello's avatar
wmmhello 已提交
2315
    tstrncpy(pName.dbname, info->pRequest->pDb, sizeof(pName.dbname));
wmmhello's avatar
wmmhello 已提交
2316
    memcpy(pName.tname, tableData->childTableName, strlen(tableData->childTableName));
D
dapan1121 已提交
2317 2318 2319 2320 2321 2322

    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);
X
Xiaoyu Wang 已提交
2323

wmmhello's avatar
wmmhello 已提交
2324
    SVgroupInfo vg;
D
dapan1121 已提交
2325
    code = catalogGetTableHashVgroup(info->pCatalog, &conn, &pName, &vg);
2326
    if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2327
      uError("SML:0x%" PRIx64 " catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName);
wmmhello's avatar
wmmhello 已提交
2328 2329
      return code;
    }
X
Xiaoyu Wang 已提交
2330
    taosHashPut(info->pVgHash, (const char *)&vg.vgId, sizeof(vg.vgId), (char *)&vg, sizeof(vg));
wmmhello's avatar
wmmhello 已提交
2331

X
Xiaoyu Wang 已提交
2332 2333 2334
    SSmlSTableMeta **pMeta =
        (SSmlSTableMeta **)taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen);
    ASSERT(NULL != pMeta && NULL != *pMeta);
wmmhello's avatar
wmmhello 已提交
2335

2336
    // use tablemeta of stable to save vgid and uid of child table
wmmhello's avatar
wmmhello 已提交
2337
    (*pMeta)->tableMeta->vgId = vg.vgId;
X
Xiaoyu Wang 已提交
2338
    (*pMeta)->tableMeta->uid = tableData->uid;  // one table merge data block together according uid
wmmhello's avatar
wmmhello 已提交
2339

2340
    code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat,
2341
                       (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen,
2342
                       info->ttl, info->msgBuf.buf, info->msgBuf.len);
X
Xiaoyu Wang 已提交
2343 2344
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
wmmhello's avatar
wmmhello 已提交
2345 2346
      return code;
    }
X
Xiaoyu Wang 已提交
2347
    oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, oneTable);
wmmhello's avatar
wmmhello 已提交
2348
  }
wmmhello's avatar
wmmhello 已提交
2349

2350 2351
  code = smlBuildOutput(info->exec, info->pVgHash);
  if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2352
    uError("SML:0x%" PRIx64 " smlBuildOutput failed", info->id);
2353 2354
    return code;
  }
2355 2356
  info->cost.insertRpcTime = taosGetTimestampUs();

X
Xiaoyu Wang 已提交
2357 2358 2359
  // launchQueryImpl(info->pRequest, info->pQuery, false, NULL);
  //  info->affectedRows = taos_affected_rows(info->pRequest);
  //  return info->pRequest->code;
wmmhello's avatar
wmmhello 已提交
2360

2361 2362 2363
  SAppClusterSummary *pActivity = &info->taos->pAppInfo->summary;
  atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);

2364 2365 2366 2367 2368 2369
  SSqlCallbackWrapper *pWrapper = (SSqlCallbackWrapper *)taosMemoryCalloc(1, sizeof(SSqlCallbackWrapper));
  if (pWrapper == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  pWrapper->pRequest = info->pRequest;
  launchAsyncQuery(info->pRequest, info->pQuery, NULL, pWrapper);
wmmhello's avatar
wmmhello 已提交
2370
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2371 2372
}

X
Xiaoyu Wang 已提交
2373 2374
static void smlPrintStatisticInfo(SSmlHandle *info) {
  uError("SML:0x%" PRIx64
2375
         " smlInsertLines result, code:%d,lineNum:%d,stable num:%d,ctable num:%d,create stable num:%d,alter stable tag num:%d,alter stable col num:%d \
X
Xiaoyu Wang 已提交
2376 2377 2378
        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,
2379 2380
         info->cost.numOfCreateSTables, info->cost.numOfAlterTagSTables, info->cost.numOfAlterColSTables,
         info->cost.schemaTime - info->cost.parseTime,
X
Xiaoyu Wang 已提交
2381 2382
         info->cost.insertBindTime - info->cost.schemaTime, info->cost.insertRpcTime - info->cost.insertBindTime,
         info->cost.endTime - info->cost.insertRpcTime, info->cost.endTime - info->cost.parseTime);
2383 2384
}

dengyihao's avatar
dengyihao 已提交
2385
static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char *rawLineEnd, int numLines) {
wmmhello's avatar
wmmhello 已提交
2386
  int32_t code = TSDB_CODE_SUCCESS;
2387
  if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
dengyihao's avatar
dengyihao 已提交
2388
    if (lines) {
wmmhello's avatar
wmmhello 已提交
2389
      code = smlParseJSON(info, *lines);
dengyihao's avatar
dengyihao 已提交
2390
    } else if (rawLine) {
wmmhello's avatar
wmmhello 已提交
2391 2392
      code = smlParseJSON(info, rawLine);
    }
2393
    if (code != TSDB_CODE_SUCCESS) {
dengyihao's avatar
dengyihao 已提交
2394
      uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, lines ? *lines : rawLine);
2395 2396
      return code;
    }
wmmhello's avatar
wmmhello 已提交
2397
    return code;
wmmhello's avatar
wmmhello 已提交
2398
  }
wmmhello's avatar
wmmhello 已提交
2399

wmmhello's avatar
wmmhello 已提交
2400
  for (int32_t i = 0; i < numLines; ++i) {
wmmhello's avatar
wmmhello 已提交
2401
    char *tmp = NULL;
dengyihao's avatar
dengyihao 已提交
2402 2403
    int   len = 0;
    if (lines) {
wmmhello's avatar
wmmhello 已提交
2404 2405
      tmp = lines[i];
      len = strlen(tmp);
dengyihao's avatar
dengyihao 已提交
2406
    } else if (rawLine) {
wmmhello's avatar
wmmhello 已提交
2407
      tmp = rawLine;
dengyihao's avatar
dengyihao 已提交
2408 2409
      while (rawLine < rawLineEnd) {
        if (*(rawLine++) == '\n') {
wmmhello's avatar
wmmhello 已提交
2410 2411 2412 2413
          break;
        }
        len++;
      }
dengyihao's avatar
dengyihao 已提交
2414
      if (info->protocol == TSDB_SML_LINE_PROTOCOL && tmp[0] == '#') {  // this line is comment
wmmhello's avatar
wmmhello 已提交
2415 2416
        continue;
      }
wmmhello's avatar
wmmhello 已提交
2417 2418
    }

X
Xiaoyu Wang 已提交
2419
    if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
wmmhello's avatar
wmmhello 已提交
2420
      code = smlParseInfluxLine(info, tmp, len);
X
Xiaoyu Wang 已提交
2421
    } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
wmmhello's avatar
wmmhello 已提交
2422
      code = smlParseTelnetLine(info, tmp, len);
X
Xiaoyu Wang 已提交
2423
    } else {
2424 2425
      ASSERT(0);
    }
wmmhello's avatar
wmmhello 已提交
2426
    if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2427
      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, tmp);
2428
      return code;
wmmhello's avatar
wmmhello 已提交
2429 2430
    }
  }
2431 2432 2433
  return code;
}

dengyihao's avatar
dengyihao 已提交
2434
static int smlProcess(SSmlHandle *info, char *lines[], char *rawLine, char *rawLineEnd, int numLines) {
2435
  int32_t code = TSDB_CODE_SUCCESS;
2436 2437
  int32_t retryNum = 0;

2438 2439
  info->cost.parseTime = taosGetTimestampUs();

wmmhello's avatar
wmmhello 已提交
2440
  code = smlParseLine(info, lines, rawLine, rawLineEnd, numLines);
2441
  if (code != 0) {
X
Xiaoyu Wang 已提交
2442
    uError("SML:0x%" PRIx64 " smlParseLine error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2443
    return code;
2444
  }
wmmhello's avatar
wmmhello 已提交
2445

2446 2447 2448 2449 2450
  info->cost.lineNum = numLines;
  info->cost.numOfSTables = taosHashGetSize(info->superTables);
  info->cost.numOfCTables = taosHashGetSize(info->childTables);

  info->cost.schemaTime = taosGetTimestampUs();
2451

X
Xiaoyu Wang 已提交
2452
  do {
2453 2454
    code = smlModifyDBSchemas(info);
    if (code == 0) break;
wmmhello's avatar
wmmhello 已提交
2455
  } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);
2456

wmmhello's avatar
wmmhello 已提交
2457
  if (code != 0) {
X
Xiaoyu Wang 已提交
2458
    uError("SML:0x%" PRIx64 " smlModifyDBSchemas error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2459
    return code;
wmmhello's avatar
wmmhello 已提交
2460
  }
wmmhello's avatar
wmmhello 已提交
2461

2462
  info->cost.insertBindTime = taosGetTimestampUs();
wmmhello's avatar
wmmhello 已提交
2463 2464
  code = smlInsertData(info);
  if (code != 0) {
X
Xiaoyu Wang 已提交
2465
    uError("SML:0x%" PRIx64 " smlInsertData error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2466
    return code;
wmmhello's avatar
wmmhello 已提交
2467 2468 2469 2470 2471
  }

  return code;
}

X
Xiaoyu Wang 已提交
2472
static int32_t isSchemalessDb(STscObj *taos, SRequestObj *request) {
2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489 2490 2491 2492 2493 2494 2495 2496 2497 2498 2499 2500
  //  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;
  //  }
wmmhello's avatar
wmmhello 已提交
2501 2502 2503
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2504
static void smlInsertCallback(void *param, void *res, int32_t code) {
wmmhello's avatar
wmmhello 已提交
2505
  SRequestObj *pRequest = (SRequestObj *)res;
X
Xiaoyu Wang 已提交
2506
  SSmlHandle  *info = (SSmlHandle *)param;
2507
  int32_t      rows = taos_affected_rows(pRequest);
wmmhello's avatar
wmmhello 已提交
2508

X
Xiaoyu Wang 已提交
2509
  uDebug("SML:0x%" PRIx64 " result. code:%d, msg:%s", info->id, pRequest->code, pRequest->msgBuf);
2510
  Params *pParam = info->params;
wmmhello's avatar
wmmhello 已提交
2511
  // lock
2512 2513
  taosThreadSpinLock(&pParam->lock);
  pParam->cnt++;
X
Xiaoyu Wang 已提交
2514
  if (code != TSDB_CODE_SUCCESS) {
2515 2516
    pParam->request->code = code;
    pParam->request->body.resInfo.numOfRows += rows;
2517
  } else {
2518 2519
    pParam->request->body.resInfo.numOfRows += info->affectedRows;
  }
2520 2521 2522
  // unlock
  taosThreadSpinUnlock(&pParam->lock);

2523 2524
  if (pParam->cnt == pParam->total) {
    tsem_post(&pParam->sem);
wmmhello's avatar
wmmhello 已提交
2525
  }
wmmhello's avatar
wmmhello 已提交
2526
  uDebug("SML:0x%" PRIx64 " insert finished, code: %d, rows: %d, total: %d", info->id, code, rows, info->affectedRows);
wmmhello's avatar
wmmhello 已提交
2527 2528 2529
  info->cost.endTime = taosGetTimestampUs();
  info->cost.code = code;
  smlPrintStatisticInfo(info);
wmmhello's avatar
wmmhello 已提交
2530 2531 2532
  smlDestroyInfo(info);
}

dengyihao's avatar
dengyihao 已提交
2533
TAOS_RES *taos_schemaless_insert_inner(SRequestObj *request, char *lines[], char *rawLine, char *rawLineEnd,
2534
                                       int numLines, int protocol, int precision, int32_t ttl) {
2535 2536
  int      batchs = 0;
  STscObj *pTscObj = request->pTscObj;
2537

2538
  pTscObj->schemalessType = 1;
wmmhello's avatar
wmmhello 已提交
2539
  SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
wmmhello's avatar
wmmhello 已提交
2540

2541
  Params params = {0};
wmmhello's avatar
wmmhello 已提交
2542
  params.request = request;
wmmhello's avatar
wmmhello 已提交
2543 2544 2545
  tsem_init(&params.sem, 0, 0);
  taosThreadSpinInit(&(params.lock), 0);

X
Xiaoyu Wang 已提交
2546
  if (request->pDb == NULL) {
wmmhello's avatar
wmmhello 已提交
2547
    request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
wmmhello's avatar
wmmhello 已提交
2548
    smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
wmmhello's avatar
wmmhello 已提交
2549 2550 2551
    goto end;
  }

2552
  if (isSchemalessDb(pTscObj, request) != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2553
    request->code = TSDB_CODE_SML_INVALID_DB_CONF;
wmmhello's avatar
wmmhello 已提交
2554
    smlBuildInvalidDataMsg(&msg, "Cannot write data to a non schemaless database", NULL);
wmmhello's avatar
wmmhello 已提交
2555 2556 2557
    goto end;
  }

X
Xiaoyu Wang 已提交
2558
  if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
2559
    request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
wmmhello's avatar
wmmhello 已提交
2560
    smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
2561
    goto end;
wmmhello's avatar
wmmhello 已提交
2562 2563
  }

X
Xiaoyu Wang 已提交
2564 2565
  if (protocol == TSDB_SML_LINE_PROTOCOL &&
      (precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)) {
2566
    request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
wmmhello's avatar
wmmhello 已提交
2567
    smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
2568 2569 2570
    goto end;
  }

2571
  if (protocol == TSDB_SML_JSON_PROTOCOL) {
wmmhello's avatar
wmmhello 已提交
2572
    numLines = 1;
2573
  } else if (numLines <= 0) {
wmmhello's avatar
wmmhello 已提交
2574 2575 2576 2577 2578
    request->code = TSDB_CODE_SML_INVALID_DATA;
    smlBuildInvalidDataMsg(&msg, "line num is invalid", NULL);
    goto end;
  }

2579
  batchs = ceil(((double)numLines) / tsSmlBatchSize);
2580
  params.total = batchs;
wmmhello's avatar
wmmhello 已提交
2581
  for (int i = 0; i < batchs; ++i) {
dengyihao's avatar
dengyihao 已提交
2582
    SRequestObj *req = (SRequestObj *)createRequest(pTscObj->id, TSDB_SQL_INSERT, 0);
2583
    if (!req) {
wmmhello's avatar
wmmhello 已提交
2584 2585 2586 2587
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error request is null");
      goto end;
    }
2588 2589
    SSmlHandle *info = smlBuildSmlInfo(pTscObj, req, (SMLProtocolType)protocol, precision);
    if (!info) {
wmmhello's avatar
wmmhello 已提交
2590 2591 2592 2593 2594
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error SSmlHandle is null");
      goto end;
    }

wmmhello's avatar
wmmhello 已提交
2595
    info->isRawLine = (rawLine == NULL);
2596
    info->ttl       = ttl;
wmmhello's avatar
wmmhello 已提交
2597

2598
    int32_t perBatch = tsSmlBatchSize;
wmmhello's avatar
wmmhello 已提交
2599

X
Xiaoyu Wang 已提交
2600
    if (numLines > perBatch) {
wmmhello's avatar
wmmhello 已提交
2601
      numLines -= perBatch;
X
Xiaoyu Wang 已提交
2602
    } else {
wmmhello's avatar
wmmhello 已提交
2603 2604 2605 2606
      perBatch = numLines;
      numLines = 0;
    }

wmmhello's avatar
wmmhello 已提交
2607
    info->params = &params;
wmmhello's avatar
wmmhello 已提交
2608 2609
    info->affectedRows = perBatch;
    info->pRequest->body.queryFp = smlInsertCallback;
X
Xiaoyu Wang 已提交
2610
    info->pRequest->body.param = info;
wmmhello's avatar
wmmhello 已提交
2611
    int32_t code = smlProcess(info, lines, rawLine, rawLineEnd, perBatch);
dengyihao's avatar
dengyihao 已提交
2612
    if (lines) {
wmmhello's avatar
wmmhello 已提交
2613 2614
      lines += perBatch;
    }
dengyihao's avatar
dengyihao 已提交
2615
    if (rawLine) {
wmmhello's avatar
wmmhello 已提交
2616
      int num = 0;
dengyihao's avatar
dengyihao 已提交
2617 2618
      while (rawLine < rawLineEnd) {
        if (*(rawLine++) == '\n') {
wmmhello's avatar
wmmhello 已提交
2619 2620
          num++;
        }
dengyihao's avatar
dengyihao 已提交
2621
        if (num == perBatch) {
wmmhello's avatar
wmmhello 已提交
2622 2623 2624 2625
          break;
        }
      }
    }
X
Xiaoyu Wang 已提交
2626
    if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2627 2628 2629 2630
      info->pRequest->body.queryFp(info, req, code);
    }
  }
  tsem_wait(&params.sem);
2631

dengyihao's avatar
dengyihao 已提交
2632
end:
wmmhello's avatar
wmmhello 已提交
2633 2634
  taosThreadSpinDestroy(&params.lock);
  tsem_destroy(&params.sem);
2635
  //  ((STscObj *)taos)->schemalessType = 0;
2636
  pTscObj->schemalessType = 1;
2637
  uDebug("resultend:%s", request->msgBuf);
2638
  return (TAOS_RES *)request;
wmmhello's avatar
wmmhello 已提交
2639
}
wmmhello's avatar
wmmhello 已提交
2640 2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655 2656

/**
 * 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
dengyihao's avatar
dengyihao 已提交
2657
 * @return TAOS_RES
wmmhello's avatar
wmmhello 已提交
2658 2659
 */

2660 2661
TAOS_RES *taos_schemaless_insert_ttl_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol, int precision,
                                                int32_t ttl, int64_t reqid) {
wmmhello's avatar
wmmhello 已提交
2662 2663 2664 2665 2666
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

2667
  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
dengyihao's avatar
dengyihao 已提交
2668 2669 2670 2671 2672 2673 2674 2675 2676 2677 2678 2679
  if (!request) {
    uError("SML:taos_schemaless_insert error request is null");
    return NULL;
  }

  if (!lines) {
    SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
    request->code = TSDB_CODE_SML_INVALID_DATA;
    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
    return (TAOS_RES *)request;
  }

2680
  return taos_schemaless_insert_inner(request, lines, NULL, NULL, numLines, protocol, precision, ttl);
dengyihao's avatar
dengyihao 已提交
2681 2682
}

2683 2684 2685
TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, TSDB_DEFAULT_TABLE_TTL, 0);
}
wmmhello's avatar
wmmhello 已提交
2686

2687 2688 2689
TAOS_RES *taos_schemaless_insert_ttl(TAOS *taos, char *lines[], int numLines, int protocol, int precision, int32_t ttl) {
  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, ttl, 0);
}
wmmhello's avatar
wmmhello 已提交
2690

2691 2692
TAOS_RES *taos_schemaless_insert_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol, int precision, int64_t reqid) {
  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, TSDB_DEFAULT_TABLE_TTL, reqid);
wmmhello's avatar
wmmhello 已提交
2693 2694
}

2695 2696
TAOS_RES *taos_schemaless_insert_raw_ttl_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
                                                int precision, int32_t ttl, int64_t reqid) {
dengyihao's avatar
dengyihao 已提交
2697 2698 2699 2700 2701
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }

2702
  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
dengyihao's avatar
dengyihao 已提交
2703 2704 2705 2706 2707 2708 2709 2710 2711 2712 2713 2714 2715 2716 2717 2718 2719 2720 2721 2722 2723 2724 2725 2726
  if (!request) {
    uError("SML:taos_schemaless_insert error request is null");
    return NULL;
  }

  if (!lines || len <= 0) {
    SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
    request->code = TSDB_CODE_SML_INVALID_DATA;
    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
    return (TAOS_RES *)request;
  }

  int numLines = 0;
  *totalRows = 0;
  char *tmp = lines;
  for (int i = 0; i < len; i++) {
    if (lines[i] == '\n' || i == len - 1) {
      numLines++;
      if (tmp[0] != '#' || protocol != TSDB_SML_LINE_PROTOCOL) {  // ignore comment
        (*totalRows)++;
      }
      tmp = lines + i + 1;
    }
  }
2727
  return taos_schemaless_insert_inner(request, NULL, lines, lines + len, numLines, protocol, precision, ttl);
dengyihao's avatar
dengyihao 已提交
2728 2729
}

2730 2731 2732 2733 2734 2735
TAOS_RES *taos_schemaless_insert_raw_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol, int precision, int64_t reqid) {
  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision, TSDB_DEFAULT_TABLE_TTL, reqid);
}
TAOS_RES *taos_schemaless_insert_raw_ttl(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol, int precision, int32_t ttl) {
  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision, ttl, 0);
}
wmmhello's avatar
wmmhello 已提交
2736

2737 2738
TAOS_RES *taos_schemaless_insert_raw(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol, int precision) {
  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision, TSDB_DEFAULT_TABLE_TTL, 0);
wmmhello's avatar
wmmhello 已提交
2739
}