clientSml.c 102.3 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

23 24 25 26 27 28 29 30 31 32 33 34 35
#if (defined(__GNUC__) && (__GNUC__ >= 3)) || (defined(__INTEL_COMPILER) && (__INTEL_COMPILER >= 800)) || defined(__clang__)
#  define expect(expr,value)    (__builtin_expect ((expr),(value)) )
#else
#  define expect(expr,value)    (expr)
#endif

#ifndef likely
#define likely(expr)     expect((expr) != 0, 1)
#endif
#ifndef unlikely
#define unlikely(expr)   expect((expr) != 0, 0)
#endif

wmmhello's avatar
wmmhello 已提交
36
//=================================================================================================
wmmhello's avatar
wmmhello 已提交
37 38 39 40 41 42 43

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

dengyihao's avatar
dengyihao 已提交
44 45
#define JUMP_SPACE(sql, sqlEnd) \
  while (sql < sqlEnd) {        \
46
    if (unlikely(*sql == SPACE))          \
dengyihao's avatar
dengyihao 已提交
47 48 49
      sql++;                    \
    else                        \
      break;                    \
X
Xiaoyu Wang 已提交
50
  }
wmmhello's avatar
wmmhello 已提交
51
// comma ,
wmmhello's avatar
wmmhello 已提交
52
//#define IS_SLASH_COMMA(sql) (*(sql) == COMMA && *((sql)-1) == SLASH)
X
Xiaoyu Wang 已提交
53
#define IS_COMMA(sql)       (*(sql) == COMMA && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
54
// space
wmmhello's avatar
wmmhello 已提交
55
//#define IS_SLASH_SPACE(sql) (*(sql) == SPACE && *((sql)-1) == SLASH)
X
Xiaoyu Wang 已提交
56
#define IS_SPACE(sql)       (*(sql) == SPACE && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
57
// equal =
wmmhello's avatar
wmmhello 已提交
58
//#define IS_SLASH_EQUAL(sql) (*(sql) == EQUAL && *((sql)-1) == SLASH)
X
Xiaoyu Wang 已提交
59
#define IS_EQUAL(sql)       (*(sql) == EQUAL && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
60
// quote "
wmmhello's avatar
wmmhello 已提交
61
//#define IS_SLASH_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) == SLASH)
X
Xiaoyu Wang 已提交
62
#define IS_QUOTE(sql)       (*(sql) == QUOTE && *((sql)-1) != SLASH)
wmmhello's avatar
wmmhello 已提交
63
// SLASH
wmmhello's avatar
wmmhello 已提交
64
//#define IS_SLASH_SLASH(sql) (*(sql) == SLASH && *((sql)-1) == SLASH)
wmmhello's avatar
wmmhello 已提交
65

X
Xiaoyu Wang 已提交
66
#define IS_SLASH_LETTER(sql) \
wmmhello's avatar
wmmhello 已提交
67 68
  (*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == EQUAL || *(sql) == QUOTE || *(sql) == SLASH))                          \
//  (IS_SLASH_COMMA(sql) || IS_SLASH_SPACE(sql) || IS_SLASH_EQUAL(sql) || IS_SLASH_QUOTE(sql) || IS_SLASH_SLASH(sql))
wmmhello's avatar
wmmhello 已提交
69

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

X
Xiaoyu Wang 已提交
72 73 74 75 76 77 78 79
#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 已提交
80

81 82 83
#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)

84 85 86
#define OTD_JSON_SUB_FIELDS_NUM 2
#define OTD_JSON_FIELDS_NUM     4

X
Xiaoyu Wang 已提交
87 88 89 90
#define TS        "_ts"
#define TS_LEN    3
#define VALUE     "_value"
#define VALUE_LEN 6
91

92 93
#define JSON_METERS_NAME "__JM"

X
Xiaoyu Wang 已提交
94 95
#define BINARY_ADD_LEN 2  // "binary"   2 means " "
#define NCHAR_ADD_LEN  3  // L"nchar"   3 means L" "
wmmhello's avatar
wmmhello 已提交
96 97

#define MAX_RETRY_TIMES 5
wmmhello's avatar
wmmhello 已提交
98 99 100 101
//=================================================================================================
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;

typedef enum {
102
  SCHEMA_ACTION_NULL,
wmmhello's avatar
wmmhello 已提交
103 104 105 106 107
  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 已提交
108 109
} ESchemaAction;

wmmhello's avatar
wmmhello 已提交
110 111 112 113 114 115 116 117 118 119 120 121 122
/*********************** list start *********************************/
typedef struct {
  const void  *key;
  int32_t      keyLen;
  void        *value;
  bool         used;
}Node;

typedef struct NodeList{
  Node             data;
  struct NodeList* next;
}NodeList;

123 124 125
typedef int32_t (*_equal_fn_sml)(const void *, const void *);

static void* nodeListGet(NodeList* list, const void *key, int32_t len, _equal_fn_sml fn){
wmmhello's avatar
wmmhello 已提交
126 127
  NodeList *tmp = list;
  while(tmp){
128 129 130 131 132 133 134 135
    if(fn == NULL){
      if(tmp->data.used && tmp->data.keyLen == len && memcmp(tmp->data.key, key, len) == 0) {
        return tmp->data.value;
      }
    }else{
      if(tmp->data.used && fn(tmp->data.key, key) == 0) {
        return tmp->data.value;
      }
wmmhello's avatar
wmmhello 已提交
136
    }
137

wmmhello's avatar
wmmhello 已提交
138 139 140 141 142
    tmp = tmp->next;
  }
  return NULL;
}

143
static int nodeListSet(NodeList** list, const void *key, int32_t len, void* value, _equal_fn_sml fn){
wmmhello's avatar
wmmhello 已提交
144 145 146
  NodeList *tmp = *list;
  while (tmp){
    if(!tmp->data.used) break;
147 148 149 150 151 152 153 154
    if(fn == NULL){
      if(tmp->data.keyLen == len && memcmp(tmp->data.key, key, len) == 0) {
        return -1;
      }
    }else{
      if(tmp->data.keyLen == len && fn(tmp->data.key, key) == 0) {
        return -1;
      }
wmmhello's avatar
wmmhello 已提交
155
    }
156

wmmhello's avatar
wmmhello 已提交
157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189
    tmp = tmp->next;
  }
  if(tmp){
    tmp->data.key = key;
    tmp->data.keyLen = len;
    tmp->data.value = value;
    tmp->data.used = true;
  }else{
    NodeList *newNode = taosMemoryCalloc(1, sizeof(NodeList));
    if(newNode == NULL){
      return -1;
    }
    newNode->data.key = key;
    newNode->data.keyLen = len;
    newNode->data.value = value;
    newNode->data.used = true;
    newNode->next = *list;
    *list = newNode;
  }
  return 0;
}

static int nodeListSize(NodeList* list){
  int cnt = 0;
  while(list){
    if(list->data.used) cnt++;
    else break;
    list = list->next;
  }
  return cnt;
}
/*********************** list end *********************************/

wmmhello's avatar
wmmhello 已提交
190
typedef struct {
191 192 193 194
  char *measure;
  char *tags;
  char *cols;
  char *timestamp;
wmmhello's avatar
wmmhello 已提交
195 196 197 198 199 200

  int32_t measureLen;
  int32_t measureTagsLen;
  int32_t tagsLen;
  int32_t colsLen;
  int32_t timestampLen;
201 202

  SArray *colArray;
wmmhello's avatar
wmmhello 已提交
203 204 205
} SSmlLineInfo;

typedef struct {
X
Xiaoyu Wang 已提交
206 207 208 209
  const char *sTableName;  // super table name
  int32_t     sTableNameLen;
  char        childTableName[TSDB_TABLE_NAME_LEN];
  uint64_t    uid;
210
  void       *key;        // for openTsdb telnet
wmmhello's avatar
wmmhello 已提交
211

X
Xiaoyu Wang 已提交
212
  SArray *tags;
wmmhello's avatar
wmmhello 已提交
213

214
  // elements are SHashObj<cols key string, SSmlKv*> for find by key quickly
X
Xiaoyu Wang 已提交
215
  SArray *cols;
216
  STableDataCxt *tableDataCtx;
wmmhello's avatar
wmmhello 已提交
217 218 219
} SSmlTableInfo;

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

X
Xiaoyu Wang 已提交
223 224
  SArray   *cols;
  SHashObj *colHash;
225

wmmhello's avatar
wmmhello 已提交
226 227 228 229
  STableMeta *tableMeta;
} SSmlSTableMeta;

typedef struct {
X
Xiaoyu Wang 已提交
230 231
  int32_t len;
  char   *buf;
wmmhello's avatar
wmmhello 已提交
232 233
} SSmlMsgBuf;

234 235 236 237 238 239 240
typedef struct {
  int32_t code;
  int32_t lineNum;

  int32_t numOfSTables;
  int32_t numOfCTables;
  int32_t numOfCreateSTables;
241 242
  int32_t numOfAlterColSTables;
  int32_t numOfAlterTagSTables;
243 244 245 246 247 248 249 250

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

wmmhello's avatar
wmmhello 已提交
251
typedef struct {
X
Xiaoyu Wang 已提交
252 253 254 255
  int64_t id;

  SMLProtocolType protocol;
  int8_t          precision;
256
  bool            reRun;
257 258 259
  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 已提交
260

wmmhello's avatar
wmmhello 已提交
261 262
  NodeList *childTables;
  NodeList *superTables;
X
Xiaoyu Wang 已提交
263 264 265 266 267 268 269 270
  SHashObj *pVgHash;

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

  SSmlCostInfo cost;
wmmhello's avatar
wmmhello 已提交
271
  int32_t      lineNum;
X
Xiaoyu Wang 已提交
272
  SSmlMsgBuf   msgBuf;
wmmhello's avatar
wmmhello 已提交
273 274

  cJSON       *root;  // for parse json
wmmhello's avatar
wmmhello 已提交
275
  SSmlLineInfo      *lines; // element is SSmlLineInfo
276 277 278 279 280 281 282 283

  //
  SArray      *preLineTagKV;
  SArray      *preLineColKV;

  SSmlLineInfo preLine;
  STableMeta  *currSTableMeta;
  STableDataCxt *currTableDataCtx;
wmmhello's avatar
wmmhello 已提交
284
  bool         needModifySchema;
wmmhello's avatar
wmmhello 已提交
285
} SSmlHandle;
wmmhello's avatar
wmmhello 已提交
286 287
//=================================================================================================

wmmhello's avatar
wmmhello 已提交
288
//=================================================================================================
289
static volatile int64_t linesSmlHandleId = 0;
X
Xiaoyu Wang 已提交
290
static int64_t          smlGenId() {
291
  int64_t id;
wmmhello's avatar
wmmhello 已提交
292

293 294
  do {
    id = atomic_add_fetch_64(&linesSmlHandleId, 1);
wmmhello's avatar
wmmhello 已提交
295 296
  } while (id == 0);

297
  return id;
wmmhello's avatar
wmmhello 已提交
298 299
}

300
static inline bool smlDoubleToInt64OverFlow(double num) {
X
Xiaoyu Wang 已提交
301
  if (num >= (double)INT64_MAX || num <= (double)INT64_MIN) return true;
302 303 304 305 306 307 308 309 310 311 312 313
  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 已提交
314
static int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2) {
315
  if (pBuf->buf) {
316 317 318 319 320 321 322
    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 已提交
323
  }
wmmhello's avatar
wmmhello 已提交
324 325 326
  return TSDB_CODE_SML_INVALID_DATA;
}

X
Xiaoyu Wang 已提交
327
static int32_t smlGenerateSchemaAction(SSchema *colField, SHashObj *colHash, SSmlKv *kv, bool isTag,
328
                                       ESchemaAction *action, SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
329
  uint16_t *index = colHash ? (uint16_t *)taosHashGet(colHash, kv->key, kv->keyLen) : NULL;
330 331
  if (index) {
    if (colField[*index].type != kv->type) {
X
Xiaoyu Wang 已提交
332 333
      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);
334 335 336
      return TSDB_CODE_TSC_INVALID_VALUE;
    }

X
Xiaoyu Wang 已提交
337 338 339 340
    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))) {
341
      if (isTag) {
wmmhello's avatar
wmmhello 已提交
342
        *action = SCHEMA_ACTION_CHANGE_TAG_SIZE;
343
      } else {
wmmhello's avatar
wmmhello 已提交
344
        *action = SCHEMA_ACTION_CHANGE_COLUMN_SIZE;
345 346 347 348
      }
    }
  } else {
    if (isTag) {
wmmhello's avatar
wmmhello 已提交
349
      *action = SCHEMA_ACTION_ADD_TAG;
350
    } else {
wmmhello's avatar
wmmhello 已提交
351
      *action = SCHEMA_ACTION_ADD_COLUMN;
352 353
    }
  }
wmmhello's avatar
wmmhello 已提交
354 355 356
  return 0;
}

wmmhello's avatar
wmmhello 已提交
357
static int32_t smlFindNearestPowerOf2(int32_t length, uint8_t type) {
wmmhello's avatar
wmmhello 已提交
358
  int32_t result = 1;
X
Xiaoyu Wang 已提交
359
  while (result <= length) {
wmmhello's avatar
wmmhello 已提交
360 361
    result *= 2;
  }
362
  if (type == TSDB_DATA_TYPE_BINARY && result > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
wmmhello's avatar
wmmhello 已提交
363
    result = TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE;
364
  } else if (type == TSDB_DATA_TYPE_NCHAR && result > (TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
wmmhello's avatar
wmmhello 已提交
365 366
    result = (TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
  }
wmmhello's avatar
wmmhello 已提交
367

368
  if (type == TSDB_DATA_TYPE_NCHAR) {
369
    result = result * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
370
  } else if (type == TSDB_DATA_TYPE_BINARY) {
371
    result = result + VARSTR_HEADER_SIZE;
wmmhello's avatar
wmmhello 已提交
372
  }
373
  return result;
wmmhello's avatar
wmmhello 已提交
374 375
}

X
Xiaoyu Wang 已提交
376
static int32_t smlProcessSchemaAction(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
377
                                      ESchemaAction *action, bool isTag) {
378 379
  int32_t code = TSDB_CODE_SUCCESS;
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
380
    if (j == 0 && !isTag) continue;
381
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, j);
382
    code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, action, info);
X
Xiaoyu Wang 已提交
383
    if (code != TSDB_CODE_SUCCESS) {
384 385 386 387 388 389
      return code;
    }
  }
  return TSDB_CODE_SUCCESS;
}

390
static int32_t smlCheckMeta(SSchema *schema, int32_t length, SArray *cols, bool isTag) {
391
  SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
392 393
  int32_t   i = 0;
  for (; i < length; i++) {
394 395 396
    taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
  }

397
  if (isTag) {
398 399 400 401 402
    i = 0;
  } else {
    i = 1;
  }
  for (; i < taosArrayGetSize(cols); i++) {
403
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
X
Xiaoyu Wang 已提交
404
    if (taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL) {
wmmhello's avatar
wmmhello 已提交
405
      taosHashCleanup(hashTmp);
406 407 408
      return -1;
    }
  }
409
  taosHashCleanup(hashTmp);
410 411 412
  return 0;
}

413
static int32_t getBytes(uint8_t type, int32_t length) {
414 415 416 417 418 419 420
  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
    return smlFindNearestPowerOf2(length, type);
  } else {
    return tDataTypes[type].bytes;
  }
}

421 422
static int32_t smlBuildFieldsList(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
                                  SArray *results, int32_t numOfCols, bool isTag) {
wmmhello's avatar
wmmhello 已提交
423
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
424
    SSmlKv       *kv = (SSmlKv *)taosArrayGet(cols, j);
wmmhello's avatar
wmmhello 已提交
425 426
    ESchemaAction action = SCHEMA_ACTION_NULL;
    smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, &action, info);
427
    if (action == SCHEMA_ACTION_ADD_COLUMN || action == SCHEMA_ACTION_ADD_TAG) {
wmmhello's avatar
wmmhello 已提交
428 429 430 431 432
      SField field = {0};
      field.type = kv->type;
      field.bytes = getBytes(kv->type, kv->length);
      memcpy(field.name, kv->key, kv->keyLen);
      taosArrayPush(results, &field);
433
    } else if (action == SCHEMA_ACTION_CHANGE_COLUMN_SIZE || action == SCHEMA_ACTION_CHANGE_TAG_SIZE) {
wmmhello's avatar
wmmhello 已提交
434
      uint16_t *index = (uint16_t *)taosHashGet(schemaHash, kv->key, kv->keyLen);
435 436
      uint16_t  newIndex = *index;
      if (isTag) newIndex -= numOfCols;
wmmhello's avatar
wmmhello 已提交
437 438 439 440 441 442 443
      SField *field = (SField *)taosArrayGet(results, newIndex);
      field->bytes = getBytes(kv->type, kv->length);
    }
  }
  return TSDB_CODE_SUCCESS;
}

444 445 446 447 448
// 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;
449 450 451 452
  SMCreateStbReq pReq = {0};
  int32_t        code = TSDB_CODE_SUCCESS;
  SCmdMsgInfo    pCmdMsg = {0};

wmmhello's avatar
wmmhello 已提交
453 454 455 456 457 458
  // put front for free
  pReq.numOfColumns = taosArrayGetSize(pColumns);
  pReq.pColumns = pColumns;
  pReq.numOfTags = taosArrayGetSize(pTags);
  pReq.pTags = pTags;

dengyihao's avatar
dengyihao 已提交
459
  code = buildRequest(info->taos->id, "", 0, NULL, false, &pRequest, 0);
460 461 462 463
  if (code != TSDB_CODE_SUCCESS) {
    goto end;
  }

wmmhello's avatar
wmmhello 已提交
464
  pRequest->syncQuery = true;
465 466 467 468 469
  if (!pRequest->pDb) {
    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
    goto end;
  }

470
  if (action == SCHEMA_ACTION_CREATE_STABLE) {
wmmhello's avatar
wmmhello 已提交
471 472 473 474
    pReq.colVer = 1;
    pReq.tagVer = 1;
    pReq.suid = 0;
    pReq.source = TD_REQ_FROM_APP;
475
  } else if (action == SCHEMA_ACTION_ADD_TAG || action == SCHEMA_ACTION_CHANGE_TAG_SIZE) {
wmmhello's avatar
wmmhello 已提交
476 477 478 479
    pReq.colVer = pTableMeta->sversion;
    pReq.tagVer = pTableMeta->tversion + 1;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
480
  } else if (action == SCHEMA_ACTION_ADD_COLUMN || action == SCHEMA_ACTION_CHANGE_COLUMN_SIZE) {
wmmhello's avatar
wmmhello 已提交
481 482 483 484 485 486
    pReq.colVer = pTableMeta->sversion + 1;
    pReq.tagVer = pTableMeta->tversion;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
  }

487
  if (pReq.numOfTags == 0) {
wmmhello's avatar
wmmhello 已提交
488 489 490 491 492 493 494 495
    pReq.numOfTags = 1;
    SField field = {0};
    field.type = TSDB_DATA_TYPE_NCHAR;
    field.bytes = 1;
    strcpy(field.name, tsSmlTagName);
    taosArrayPush(pReq.pTags, &field);
  }

496 497 498 499 500 501 502 503 504 505 506 507 508 509
  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 已提交
510 511
  SQuery pQuery;
  memset(&pQuery, 0, sizeof(pQuery));
512 513 514 515 516 517 518
  pQuery.execMode = QUERY_EXEC_MODE_RPC;
  pQuery.pCmdMsg = &pCmdMsg;
  pQuery.msgType = pQuery.pCmdMsg->msgType;
  pQuery.stableQuery = true;

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

519
  if (pRequest->code == TSDB_CODE_SUCCESS) {
520 521 522 523 524
    catalogRemoveTableMeta(info->pCatalog, pName);
  }
  code = pRequest->code;
  taosMemoryFree(pCmdMsg.pMsg);

525
  end:
526 527 528 529 530
  destroyRequest(pRequest);
  tFreeSMCreateStbReq(&pReq);
  return code;
}

X
Xiaoyu Wang 已提交
531
static int32_t smlModifyDBSchemas(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
532 533 534
  if(info->dataFormat && !info->needModifySchema){
    return TSDB_CODE_SUCCESS;
  }
535 536 537
  int32_t     code = 0;
  SHashObj   *hashTmp = NULL;
  STableMeta *pTableMeta = NULL;
wmmhello's avatar
wmmhello 已提交
538

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

D
dapan1121 已提交
542 543 544 545 546
  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 已提交
547

wmmhello's avatar
wmmhello 已提交
548 549 550
  NodeList *tmp = info->superTables;
  while (tmp) {
    SSmlSTableMeta *sTableData = tmp->data.value;
X
Xiaoyu Wang 已提交
551
    bool            needCheckMeta = false;  // for multi thread
wmmhello's avatar
wmmhello 已提交
552

wmmhello's avatar
wmmhello 已提交
553 554
    size_t superTableLen = (size_t)tmp->data.keyLen;
    const void  *superTable = tmp->data.key;
555
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
wmmhello's avatar
wmmhello 已提交
556
    memcpy(pName.tname, superTable, superTableLen);
wmmhello's avatar
wmmhello 已提交
557

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

560
    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_STB_NOT_EXIST) {
561 562
      SArray *pColumns = taosArrayInit(taosArrayGetSize(sTableData->cols), sizeof(SField));
      SArray *pTags = taosArrayInit(taosArrayGetSize(sTableData->tags), sizeof(SField));
wmmhello's avatar
wmmhello 已提交
563 564 565 566
      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);
567
      if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
568
        uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, pName.tname);
569
        goto end;
wmmhello's avatar
wmmhello 已提交
570
      }
571
      info->cost.numOfCreateSTables++;
wmmhello's avatar
wmmhello 已提交
572 573 574 575 576 577 578
      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 已提交
579
    } else if (code == TSDB_CODE_SUCCESS) {
580 581
      hashTmp = taosHashInit(pTableMeta->tableInfo.numOfTags, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true,
                             HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
582 583
      for (uint16_t i = pTableMeta->tableInfo.numOfColumns;
           i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
584 585
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
wmmhello's avatar
wmmhello 已提交
586

587 588
      ESchemaAction action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &action, true);
589
      if (code != TSDB_CODE_SUCCESS) {
590
        goto end;
591
      }
592 593 594 595 596
      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 已提交
597 598 599 600 601 602

        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);
603
          if (i < pTableMeta->tableInfo.numOfColumns) {
wmmhello's avatar
wmmhello 已提交
604
            taosArrayPush(pColumns, &field);
605
          } else {
wmmhello's avatar
wmmhello 已提交
606 607 608
            taosArrayPush(pTags, &field);
          }
        }
609 610
        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->tags, pTags,
                           pTableMeta->tableInfo.numOfColumns, true);
wmmhello's avatar
wmmhello 已提交
611 612

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

618
        info->cost.numOfAlterTagSTables++;
wmmhello's avatar
wmmhello 已提交
619 620 621 622 623 624 625 626 627
        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 已提交
628
      }
629 630

      taosHashClear(hashTmp);
wmmhello's avatar
wmmhello 已提交
631
      for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns; i++) {
632 633
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
634 635
      action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->cols, &action, false);
636
      if (code != TSDB_CODE_SUCCESS) {
637
        goto end;
wmmhello's avatar
wmmhello 已提交
638
      }
639 640 641 642 643
      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 已提交
644 645 646 647 648 649

        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);
650
          if (i < pTableMeta->tableInfo.numOfColumns) {
wmmhello's avatar
wmmhello 已提交
651
            taosArrayPush(pColumns, &field);
652
          } else {
wmmhello's avatar
wmmhello 已提交
653 654 655 656
            taosArrayPush(pTags, &field);
          }
        }

657 658
        smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->cols, pColumns,
                           pTableMeta->tableInfo.numOfColumns, false);
wmmhello's avatar
wmmhello 已提交
659 660

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

666
        info->cost.numOfAlterColSTables++;
wmmhello's avatar
wmmhello 已提交
667
        taosMemoryFreeClear(pTableMeta);
wmmhello's avatar
wmmhello 已提交
668 669 670 671
        code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
        if (code != TSDB_CODE_SUCCESS) {
          goto end;
        }
672 673 674 675 676
        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 已提交
677
      }
wmmhello's avatar
wmmhello 已提交
678

679
      needCheckMeta = true;
wmmhello's avatar
wmmhello 已提交
680 681
      taosHashCleanup(hashTmp);
      hashTmp = NULL;
wmmhello's avatar
wmmhello 已提交
682
    } else {
X
Xiaoyu Wang 已提交
683
      uError("SML:0x%" PRIx64 " load table meta error: %s", info->id, tstrerror(code));
684
      goto end;
wmmhello's avatar
wmmhello 已提交
685
    }
686

X
Xiaoyu Wang 已提交
687 688
    if (needCheckMeta) {
      code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags,
689
                          sTableData->tags, true);
690
      if (code != TSDB_CODE_SUCCESS) {
691
        uError("SML:0x%" PRIx64 " check tag failed. super table name %s", info->id, pName.tname);
692 693
        goto end;
      }
694
      code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols, false);
695
      if (code != TSDB_CODE_SUCCESS) {
696
        uError("SML:0x%" PRIx64 " check cols failed. super table name %s", info->id, pName.tname);
697 698 699 700
        goto end;
      }
    }

701
    sTableData->tableMeta = pTableMeta;
wmmhello's avatar
wmmhello 已提交
702

wmmhello's avatar
wmmhello 已提交
703
    tmp = tmp->next;
wmmhello's avatar
wmmhello 已提交
704 705
  }
  return 0;
706

707
  end:
wmmhello's avatar
wmmhello 已提交
708 709
  taosHashCleanup(hashTmp);
  taosMemoryFreeClear(pTableMeta);
wmmhello's avatar
wmmhello 已提交
710
//  catalogRefreshTableMeta(info->pCatalog, &conn, &pName, 1);
711
  return code;
wmmhello's avatar
wmmhello 已提交
712 713
}

714
/******************************* parse basic type function **********************/
X
Xiaoyu Wang 已提交
715
static bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
716
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
717 718 719 720
  int32_t     len = kvVal->length;
  char       *endptr = NULL;
  double      result = taosStr2Double(pVal, &endptr);
  if (pVal == endptr) {
721
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
722 723 724
    return false;
  }

725
  int32_t left = len - (endptr - pVal);
X
Xiaoyu Wang 已提交
726
  if (left == 0 || (left == 3 && strncasecmp(endptr, "f64", left) == 0)) {
727 728
    kvVal->type = TSDB_DATA_TYPE_DOUBLE;
    kvVal->d = result;
X
Xiaoyu Wang 已提交
729 730
  } else if ((left == 3 && strncasecmp(endptr, "f32", left) == 0)) {
    if (!IS_VALID_FLOAT(result)) {
731 732
      smlBuildInvalidDataMsg(msg, "float out of range[-3.402823466e+38,3.402823466e+38]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
733
    }
734 735
    kvVal->type = TSDB_DATA_TYPE_FLOAT;
    kvVal->f = (float)result;
X
Xiaoyu Wang 已提交
736 737
  } else if ((left == 1 && *endptr == 'i') || (left == 3 && strncasecmp(endptr, "i64", left) == 0)) {
    if (smlDoubleToInt64OverFlow(result)) {
wmmhello's avatar
wmmhello 已提交
738 739
      errno = 0;
      int64_t tmp = taosStr2Int64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
740
      if (errno == ERANGE) {
wmmhello's avatar
wmmhello 已提交
741 742 743 744 745 746
        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 已提交
747
    }
748
    kvVal->type = TSDB_DATA_TYPE_BIGINT;
wmmhello's avatar
wmmhello 已提交
749
    kvVal->i = (int64_t)result;
wmmhello's avatar
wmmhello 已提交
750
  } else if ((left == 1 && *endptr == 'u') || (left == 3 && strncasecmp(endptr, "u64", left) == 0)) {
X
Xiaoyu Wang 已提交
751
    if (result >= (double)UINT64_MAX || result < 0) {
wmmhello's avatar
wmmhello 已提交
752 753
      errno = 0;
      uint64_t tmp = taosStr2UInt64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
754
      if (errno == ERANGE || result < 0) {
wmmhello's avatar
wmmhello 已提交
755 756 757 758 759 760
        smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", pVal);
        return false;
      }
      kvVal->type = TSDB_DATA_TYPE_UBIGINT;
      kvVal->u = tmp;
      return true;
761
    }
762
    kvVal->type = TSDB_DATA_TYPE_UBIGINT;
wmmhello's avatar
wmmhello 已提交
763
    kvVal->u = result;
X
Xiaoyu Wang 已提交
764 765
  } else if (left == 3 && strncasecmp(endptr, "i32", left) == 0) {
    if (!IS_VALID_INT(result)) {
766 767
      smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
768
    }
769 770
    kvVal->type = TSDB_DATA_TYPE_INT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
771 772
  } else if (left == 3 && strncasecmp(endptr, "u32", left) == 0) {
    if (!IS_VALID_UINT(result)) {
773 774
      smlBuildInvalidDataMsg(msg, "unsigned int out of range[0,4294967295]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
775
    }
776 777
    kvVal->type = TSDB_DATA_TYPE_UINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
778 779
  } else if (left == 3 && strncasecmp(endptr, "i16", left) == 0) {
    if (!IS_VALID_SMALLINT(result)) {
780 781
      smlBuildInvalidDataMsg(msg, "small int our of range[-32768,32767]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
782
    }
783 784
    kvVal->type = TSDB_DATA_TYPE_SMALLINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
785 786
  } else if (left == 3 && strncasecmp(endptr, "u16", left) == 0) {
    if (!IS_VALID_USMALLINT(result)) {
787 788
      smlBuildInvalidDataMsg(msg, "unsigned small int out of rang[0,65535]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
789
    }
790 791
    kvVal->type = TSDB_DATA_TYPE_USMALLINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
792 793
  } else if (left == 2 && strncasecmp(endptr, "i8", left) == 0) {
    if (!IS_VALID_TINYINT(result)) {
794 795
      smlBuildInvalidDataMsg(msg, "tiny int out of range[-128,127]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
796
    }
797 798
    kvVal->type = TSDB_DATA_TYPE_TINYINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
799 800
  } else if (left == 2 && strncasecmp(endptr, "u8", left) == 0) {
    if (!IS_VALID_UTINYINT(result)) {
801 802
      smlBuildInvalidDataMsg(msg, "unsigned tiny int out of range[0,255]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
803
    }
804 805
    kvVal->type = TSDB_DATA_TYPE_UTINYINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
806
  } else {
807
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
808 809
    return false;
  }
810
  return true;
wmmhello's avatar
wmmhello 已提交
811 812
}

wmmhello's avatar
wmmhello 已提交
813
static bool smlParseBool(SSmlKv *kvVal) {
wmmhello's avatar
wmmhello 已提交
814
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
815
  int32_t     len = kvVal->length;
816
  if ((len == 1) && (pVal[0] == 't' || pVal[0] == 'T')) {
817
    kvVal->i = TSDB_TRUE;
wmmhello's avatar
wmmhello 已提交
818 819 820
    return true;
  }

821
  if ((len == 1) && (pVal[0] == 'f' || pVal[0] == 'F')) {
822
    kvVal->i = TSDB_FALSE;
wmmhello's avatar
wmmhello 已提交
823 824 825
    return true;
  }

X
Xiaoyu Wang 已提交
826
  if ((len == 4) && !strncasecmp(pVal, "true", len)) {
827
    kvVal->i = TSDB_TRUE;
wmmhello's avatar
wmmhello 已提交
828 829
    return true;
  }
X
Xiaoyu Wang 已提交
830
  if ((len == 5) && !strncasecmp(pVal, "false", len)) {
831
    kvVal->i = TSDB_FALSE;
wmmhello's avatar
wmmhello 已提交
832 833 834 835 836
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
837
static bool smlIsBinary(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
838
  // binary: "abc"
wmmhello's avatar
wmmhello 已提交
839 840 841 842 843 844 845 846 847
  if (len < 2) {
    return false;
  }
  if (pVal[0] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
848
static bool smlIsNchar(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
849
  // nchar: L"abc"
wmmhello's avatar
wmmhello 已提交
850 851 852
  if (len < 3) {
    return false;
  }
wmmhello's avatar
wmmhello 已提交
853
  if (pVal[1] == '"' && pVal[len - 1] == '"' && (pVal[0] == 'l' || pVal[0] == 'L')) {
wmmhello's avatar
wmmhello 已提交
854 855 856 857
    return true;
  }
  return false;
}
858
/******************************* parse basic type function end **********************/
wmmhello's avatar
wmmhello 已提交
859

860
/******************************* time function **********************/
861
static uint8_t smlPrecisionConvert[7] = {TSDB_TIME_PRECISION_NANO, TSDB_TIME_PRECISION_HOURS, TSDB_TIME_PRECISION_MINUTES,
wmmhello's avatar
wmmhello 已提交
862 863
                                     TSDB_TIME_PRECISION_SECONDS, TSDB_TIME_PRECISION_MILLI, TSDB_TIME_PRECISION_MICRO,
                                     TSDB_TIME_PRECISION_NANO};
864 865
static int64_t smlFactorNS[3] = {NANOSECOND_PER_MSEC, NANOSECOND_PER_USEC, 1};
static int64_t smlFactorS[3] = {1000LL, 1000000LL, 1000000000LL};
866
static int64_t smlToMilli[3] = {3600000LL, 60000LL, 1000LL};
wmmhello's avatar
wmmhello 已提交
867

868
static int64_t smlGetTimeValue(const char *value, int32_t len, uint8_t fromPrecision, uint8_t toPrecision) {
X
Xiaoyu Wang 已提交
869
  char   *endPtr = NULL;
wafwerar's avatar
wafwerar 已提交
870
  int64_t tsInt64 = taosStr2Int64(value, &endPtr, 10);
871
  if (unlikely(value + len != endPtr)) {
872
    return -1;
wmmhello's avatar
wmmhello 已提交
873
  }
874 875 876 877 878 879 880

  if(unlikely(fromPrecision >= TSDB_TIME_PRECISION_HOURS)){
    int64_t unit = smlToMilli[fromPrecision - TSDB_TIME_PRECISION_HOURS];
    if(unit > INT64_MAX / tsInt64){
      return -1;
    }
    tsInt64 *= unit;
881
    fromPrecision = TSDB_TIME_PRECISION_MILLI;
wmmhello's avatar
wmmhello 已提交
882 883
  }

884
  return convertTimePrecision(tsInt64, fromPrecision, toPrecision);
885 886 887 888 889 890
}

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) {
891
    return TSDB_TIME_PRECISION_MILLI;
892 893
  } else {
    return -1;
wmmhello's avatar
wmmhello 已提交
894
  }
895 896
}

X
Xiaoyu Wang 已提交
897
static int64_t smlParseInfluxTime(SSmlHandle *info, const char *data, int32_t len) {
898
  uint8_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;
wmmhello's avatar
wmmhello 已提交
899

900 901
  if(unlikely(len == 0 || (len == 1 && data[0] == '0'))){
    return taosGetTimestampNs()/smlFactorNS[toPrecision];
wmmhello's avatar
wmmhello 已提交
902
  }
903

904 905 906 907
  uint8_t fromPrecision = smlPrecisionConvert[info->precision];

  int64_t ts = smlGetTimeValue(data, len, fromPrecision, toPrecision);
  if (unlikely(ts == -1)) {
908 909
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
wmmhello's avatar
wmmhello 已提交
910
  }
911 912 913
  return ts;
}

X
Xiaoyu Wang 已提交
914
static int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len) {
915 916 917
  uint8_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;

  if (unlikely(!data)) {
918 919
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
920
  }
921 922
  if (unlikely(len == 1 && data[0] == '0')) {
    return taosGetTimestampNs()/smlFactorNS[toPrecision];
923
  }
924 925
  uint8_t fromPrecision = smlGetTsTypeByLen(len);
  if (unlikely(fromPrecision == -1)) {
X
Xiaoyu Wang 已提交
926 927
    smlBuildInvalidDataMsg(&info->msgBuf,
                           "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
928 929
    return -1;
  }
930 931
  int64_t ts = smlGetTimeValue(data, len, fromPrecision, toPrecision);
  if (unlikely(ts == -1)) {
932 933 934 935 936 937
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

938
static int64_t smlParseTS(SSmlHandle *info, const char *data, int32_t len) {
939
  int64_t ts = 0;
X
Xiaoyu Wang 已提交
940
  if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
941
    //    uError("SML:data:%s,len:%d", data, len);
942
    ts = smlParseInfluxTime(info, data, len);
X
Xiaoyu Wang 已提交
943
  } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
944
    ts = smlParseOpenTsdbTime(info, data, len);
X
Xiaoyu Wang 已提交
945
  } else {
946
    ASSERT(0);
947
  }
wmmhello's avatar
wmmhello 已提交
948
  uDebug("SML:0x%" PRIx64 " smlParseTS:%" PRId64, info->id, ts);
949

950
  return ts;
951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021
}
/******************************* time function end **********************/

/******************************* Sml struct related function **********************/
static SSmlTableInfo *smlBuildTableInfo(int numRows, const char* measure, int32_t measureLen) {
  SSmlTableInfo *tag = (SSmlTableInfo *)taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
  if (!tag) {
    return NULL;
  }

  tag->sTableName = measure;
  tag->sTableNameLen = measureLen;

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

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

  cleanup:
  taosMemoryFree(tag);
  return NULL;
}

static int32_t smlCheckDupUnit(SHashObj *dumplicateKey, SArray *tags, SSmlMsgBuf *msg){
  for(int i = 0; i < taosArrayGetSize(tags); i++) {
    SSmlKv *tag = taosArrayGet(tags, i);
    if (smlCheckDuplicateKey(tag->key, tag->keyLen, dumplicateKey)) {
      smlBuildInvalidDataMsg(msg, "dumplicate key", tag->key);
      return TSDB_CODE_TSC_DUP_NAMES;
    }
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t smlJudgeDupColName(SArray *cols, SArray *tags, SSmlMsgBuf *msg) {
  SHashObj *dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  int ret = smlCheckDupUnit(dumplicateKey, cols, msg);
  if(ret != TSDB_CODE_SUCCESS){
    goto end;
  }
  ret = smlCheckDupUnit(dumplicateKey, tags, msg);
  if(ret != TSDB_CODE_SUCCESS){
    goto end;
  }

  end:
  taosHashCleanup(dumplicateKey);
  return ret;
}

static int32_t smlParseTableName(SArray *tags, char *childTableName) {
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
  if (childTableNameLen <= 0) return TSDB_CODE_SUCCESS;

  for(int i = 0; i < taosArrayGetSize(tags); i++){
    SSmlKv *tag = taosArrayGet(tags, i);
    // handle child table name
    if (childTableNameLen == tag->keyLen && strncmp(tag->key, tsSmlChildTableName, tag->keyLen) == 0) {
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, tag->value, (tag->length < TSDB_TABLE_NAME_LEN ? tag->length : TSDB_TABLE_NAME_LEN));
      break;
    }
  }
wmmhello's avatar
wmmhello 已提交
1022

1023 1024 1025
  return TSDB_CODE_SUCCESS;
}

1026 1027 1028 1029
static int32_t smlSetCTableName(SSmlTableInfo *oneTable){
  smlParseTableName(oneTable->tags, oneTable->childTableName);

  if (strlen(oneTable->childTableName) == 0) {
1030 1031
    SArray* dst = taosArrayDup(oneTable->tags, NULL);
    RandTableName rName = {dst, oneTable->sTableName, (uint8_t)oneTable->sTableNameLen,
1032 1033 1034
                           oneTable->childTableName, 0};

    buildChildTableName(&rName);
1035
    taosArrayDestroy(dst);
1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083
    oneTable->uid = rName.uid;
  } else {
    oneTable->uid = *(uint64_t *)(oneTable->childTableName);
  }
  return TSDB_CODE_SUCCESS;
}

static SSmlSTableMeta *smlBuildSTableMeta(bool isDataFormat) {
  SSmlSTableMeta *meta = (SSmlSTableMeta *)taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if (!meta) {
    return NULL;
  }

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

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

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

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

  cleanup:
  taosMemoryFree(meta);
  return NULL;
}

static void smlInsertMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols){
  for (int16_t i = 0; i < taosArrayGetSize(cols); ++i) {
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
1084
    taosArrayPush(metaArray, kv);
1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118
    if(unlikely(metaHash != NULL)) {
      taosHashPut(metaHash, kv->key, kv->keyLen, &i, SHORT_BYTES);
    }
  }
}

static STableMeta* smlGetMeta(SSmlHandle *info, const void* measure, int32_t measureLen){
  STableMeta *pTableMeta = NULL;

  SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
  tstrncpy(pName.dbname, info->pRequest->pDb, sizeof(pName.dbname));

  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);
  memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
  memcpy(pName.tname, measure, measureLen);

  catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
  return pTableMeta;
}

static void smlDestroySTableMeta(SSmlSTableMeta *meta) {
  taosHashCleanup(meta->tagHash);
  taosHashCleanup(meta->colHash);
  taosArrayDestroy(meta->tags);
  taosArrayDestroy(meta->cols);
  taosMemoryFree(meta->tableMeta);
  taosMemoryFree(meta);
}
/******************************* Sml struct related function end **********************/

wmmhello's avatar
wmmhello 已提交
1119
static int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
X
Xiaoyu Wang 已提交
1120
  // binary
wmmhello's avatar
wmmhello 已提交
1121
  if (smlIsBinary(pVal->value, pVal->length)) {
1122
    pVal->type = TSDB_DATA_TYPE_BINARY;
wmmhello's avatar
wmmhello 已提交
1123
    pVal->length -= BINARY_ADD_LEN;
1124
    if (pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
wmmhello's avatar
wmmhello 已提交
1125 1126
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
1127
    pVal->value += (BINARY_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
1128
    return TSDB_CODE_SUCCESS;
1129
  }
X
Xiaoyu Wang 已提交
1130
  // nchar
wmmhello's avatar
wmmhello 已提交
1131
  if (smlIsNchar(pVal->value, pVal->length)) {
1132
    pVal->type = TSDB_DATA_TYPE_NCHAR;
wmmhello's avatar
wmmhello 已提交
1133
    pVal->length -= NCHAR_ADD_LEN;
1134
    if (pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
wmmhello's avatar
wmmhello 已提交
1135 1136
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
1137
    pVal->value += (NCHAR_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
1138
    return TSDB_CODE_SUCCESS;
1139 1140
  }

X
Xiaoyu Wang 已提交
1141
  // bool
1142 1143 1144
  if (smlParseBool(pVal)) {
    pVal->type = TSDB_DATA_TYPE_BOOL;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
1145
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1146
  }
X
Xiaoyu Wang 已提交
1147
  // number
1148
  if (smlParseNumber(pVal, msg)) {
wmmhello's avatar
wmmhello 已提交
1149
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
1150
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1151 1152
  }

wmmhello's avatar
wmmhello 已提交
1153
  return TSDB_CODE_TSC_INVALID_VALUE;
wmmhello's avatar
wmmhello 已提交
1154 1155
}

1156 1157 1158
int32_t is_same_child_table_json(const void *a, const void *b){
  return (cJSON_Compare((const cJSON *)a, (const cJSON *)b, true)) ? 0 : 1;
}
wmmhello's avatar
wmmhello 已提交
1159

1160 1161 1162 1163 1164 1165 1166
int32_t is_same_child_table_telnet(const void *a, const void *b){
  SSmlLineInfo *t1 = (SSmlLineInfo *)a;
  SSmlLineInfo *t2 = (SSmlLineInfo *)b;
  return (((t1->measureLen == t2->measureLen) && memcmp(t1->measure, t2->measure, t1->measureLen) == 0)
      && ((t1->tagsLen == t2->tagsLen) && memcmp(t1->tags, t2->tags, t1->tagsLen) == 0)) ? 0 : 1;
}

1167 1168 1169 1170 1171
#define IS_SAME_CHILD_TABLE (elements->measureTagsLen == info->preLine.measureTagsLen \
&& memcmp(elements->measure, info->preLine.measure, elements->measureTagsLen) == 0)

#define IS_SAME_SUPER_TABLE (elements->measureLen == info->preLine.measureLen \
&& memcmp(elements->measure, info->preLine.measure, elements->measureLen) == 0)
1172

1173 1174 1175 1176 1177 1178 1179
#define IS_SAME_KEY (preKV->keyLen == kv.keyLen && memcmp(preKV->key, kv.key, kv.keyLen) == 0)

static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd,
                          SSmlLineInfo* currElement, bool isSameMeasure, bool isSameCTable){
  if(isSameCTable){
    return TSDB_CODE_SUCCESS;
  }
wmmhello's avatar
wmmhello 已提交
1180

1181 1182 1183 1184 1185
  int     cnt = 0;
  SArray *preLineKV = info->preLineTagKV;
  bool    isSuperKVInit = true;
  SArray *superKV = NULL;
  if(info->dataFormat){
1186
    if(!isSameMeasure){
1187
      SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, currElement->measure, currElement->measureLen, NULL);
wmmhello's avatar
wmmhello 已提交
1188

1189
      if(unlikely(sMeta == NULL)){
wmmhello's avatar
wmmhello 已提交
1190
        sMeta = smlBuildSTableMeta(info->dataFormat);
1191
        STableMeta * pTableMeta = smlGetMeta(info, currElement->measure, currElement->measureLen);
wmmhello's avatar
wmmhello 已提交
1192
        sMeta->tableMeta = pTableMeta;
1193 1194 1195 1196 1197
        if(pTableMeta == NULL){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
1198
        nodeListSet(&info->superTables, currElement->measure, currElement->measureLen, sMeta, NULL);
1199 1200
      }
      info->currSTableMeta = sMeta->tableMeta;
1201
      superKV = sMeta->tags;
1202 1203

      if(unlikely(taosArrayGetSize(superKV) == 0)){
1204
        isSuperKVInit = false;
1205
      }
1206 1207 1208 1209 1210 1211 1212 1213 1214 1215
      taosArraySetSize(preLineKV, 0);
    }
  }else{
    taosArraySetSize(preLineKV, 0);
  }


  while (*sql < sqlEnd) {
    if (unlikely(IS_SPACE(*sql))) {
      break;
1216 1217
    }

1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233
    bool hasSlash = false;
    // parse key
    const char *key = *sql;
    int32_t     keyLen = 0;
    while (*sql < sqlEnd) {
      if (unlikely(IS_COMMA(*sql))) {
        smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      if (unlikely(IS_EQUAL(*sql))) {
        keyLen = *sql - key;
        (*sql)++;
        break;
      }
      if(!hasSlash){
        hasSlash = (*(*sql) == SLASH);
1234
      }
1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256
      (*sql)++;
    }
    if(unlikely(hasSlash)) {
      PROCESS_SLASH(key, keyLen)
    }

    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
      smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
    }

    // parse value
    const char *value = *sql;
    int32_t     valueLen = 0;
    hasSlash = false;
    while (*sql < sqlEnd) {
      // parse value
      if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
        break;
      }else if (unlikely(IS_EQUAL(*sql))) {
        smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
        return TSDB_CODE_SML_INVALID_DATA;
1257
      }
1258

1259 1260
      if(!hasSlash){
        hasSlash = (*(*sql) == SLASH);
1261
      }
1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282

      (*sql)++;
    }
    valueLen = *sql - value;

    if (unlikely(valueLen == 0)) {
      smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }

    if(unlikely(hasSlash)) {
      PROCESS_SLASH(value, valueLen)
    }

    if (unlikely(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }

    SSmlKv kv = {.key = key, .type = TSDB_DATA_TYPE_NCHAR, .keyLen = keyLen, .value = value, .length = valueLen};
    if(info->dataFormat){
      if(unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)){
1283 1284 1285
        info->dataFormat = false;
        info->reRun      = true;
        return TSDB_CODE_SUCCESS;
1286
      }
1287

1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332
      if(isSameMeasure){
        if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
        SSmlKv *preKV = taosArrayGet(preLineKV, cnt);
        if(unlikely(kv.length > preKV->length)){
          preKV->length = kv.length;
          SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, currElement->measure, currElement->measureLen, NULL);
          ASSERT(tableMeta != NULL);

          SSmlKv *oldKV = taosArrayGet(tableMeta->tags, cnt);
          oldKV->length = kv.length;
          info->needModifySchema = true;
        }
        if(unlikely(!IS_SAME_KEY)){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
      }else{
        if(isSuperKVInit){
          if(unlikely(cnt >= taosArrayGetSize(superKV))) {
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }
          SSmlKv *preKV = taosArrayGet(superKV, cnt);
          if(unlikely(kv.length > preKV->length)) {
            preKV->length = kv.length;
          }else{
            kv.length = preKV->length;
          }
          info->needModifySchema = true;

          if(unlikely(!IS_SAME_KEY)){
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }
        }else{
          taosArrayPush(superKV, &kv);
        }
        taosArrayPush(preLineKV, &kv);
1333
      }
1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361 1362 1363 1364 1365 1366
    }else{
      taosArrayPush(preLineKV, &kv);
    }

    cnt++;
    if(IS_SPACE(*sql)){
      break;
    }
    (*sql)++;
  }

  void* oneTable = nodeListGet(info->childTables, currElement->measure, currElement->measureTagsLen, NULL);
  if ((oneTable != NULL)) {
    return TSDB_CODE_SUCCESS;
  }

  SSmlTableInfo *tinfo = smlBuildTableInfo(1, currElement->measure, currElement->measureLen);
  if (!tinfo) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  for(int i = 0; i < taosArrayGetSize(preLineKV); i++){
    taosArrayPush(tinfo->tags, taosArrayGet(preLineKV, i));
  }
  smlSetCTableName(tinfo);
  if(info->dataFormat) {
    info->currSTableMeta->uid = tinfo->uid;
    tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
    if(tinfo->tableDataCtx == NULL){
      smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
      return TSDB_CODE_SML_INVALID_DATA;
    }
  }

1367
  nodeListSet(&info->childTables, currElement->measure, currElement->measureTagsLen, tinfo, NULL);
1368 1369 1370

  return TSDB_CODE_SUCCESS;
}
1371

1372 1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383
static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd,
                          SSmlLineInfo* currElement, bool isSameMeasure, bool isSameCTable){
  int     cnt = 0;
  SArray *preLineKV = info->preLineColKV;
  bool    isSuperKVInit = true;
  SArray *superKV = NULL;
  if(info->dataFormat){
    if(unlikely(!isSameCTable)){
      SSmlTableInfo *oneTable = (SSmlTableInfo *)nodeListGet(info->childTables, currElement->measure, currElement->measureTagsLen, NULL);
      if (unlikely(oneTable == NULL)) {
        smlBuildInvalidDataMsg(&info->msgBuf, "child table should inside", currElement->measure);
        return TSDB_CODE_SML_INVALID_DATA;
1384
      }
1385
      info->currTableDataCtx = oneTable->tableDataCtx;
1386 1387
    }

1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399
    if(unlikely(!isSameMeasure)){
      SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, currElement->measure, currElement->measureLen, NULL);

      if(unlikely(sMeta == NULL)){
        sMeta = smlBuildSTableMeta(info->dataFormat);
        STableMeta * pTableMeta = smlGetMeta(info, currElement->measure, currElement->measureLen);
        sMeta->tableMeta = pTableMeta;
        if(pTableMeta == NULL){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
1400
        nodeListSet(&info->superTables, currElement->measure, currElement->measureLen, sMeta, NULL);
1401 1402 1403 1404 1405 1406
      }
      info->currSTableMeta = sMeta->tableMeta;
      superKV = sMeta->cols;
      if(unlikely(taosArrayGetSize(superKV) == 0)){
        isSuperKVInit = false;
      }
1407 1408 1409 1410 1411
      taosArraySetSize(preLineKV, 0);
    }
  }

  while (*sql < sqlEnd) {
1412
    if (unlikely(IS_SPACE(*sql))) {
1413 1414 1415
      break;
    }

1416
    bool hasSlash = false;
1417 1418 1419 1420
    // parse key
    const char *key = *sql;
    int32_t     keyLen = 0;
    while (*sql < sqlEnd) {
1421
      if (unlikely(IS_COMMA(*sql))) {
1422 1423 1424
        smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
1425
      if (unlikely(IS_EQUAL(*sql))) {
1426 1427 1428 1429
        keyLen = *sql - key;
        (*sql)++;
        break;
      }
1430 1431 1432
      if(!hasSlash){
        hasSlash = (*(*sql) == SLASH);
      }
1433 1434
      (*sql)++;
    }
1435 1436 1437
    if(unlikely(hasSlash)) {
      PROCESS_SLASH(key, keyLen)
    }
1438

1439
    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
1440 1441 1442 1443 1444 1445 1446
      smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
    }

    // parse value
    const char *value = *sql;
    int32_t     valueLen = 0;
1447
    hasSlash              = false;
1448 1449 1450
    bool        isInQuote = false;
    while (*sql < sqlEnd) {
      // parse value
1451
      if (IS_QUOTE(*sql)) {
1452 1453 1454 1455
        isInQuote = !isInQuote;
        (*sql)++;
        continue;
      }
wmmhello's avatar
wmmhello 已提交
1456
      if (!isInQuote){
1457
        if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
wmmhello's avatar
wmmhello 已提交
1458
          break;
1459
        } else if (unlikely(IS_EQUAL(*sql))) {
wmmhello's avatar
wmmhello 已提交
1460 1461 1462
          smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
          return TSDB_CODE_SML_INVALID_DATA;
        }
1463
      }
1464 1465 1466
      if(!hasSlash){
        hasSlash = (*(*sql) == SLASH);
      }
wmmhello's avatar
wmmhello 已提交
1467

1468 1469 1470 1471
      (*sql)++;
    }
    valueLen = *sql - value;

1472
    if (unlikely(isInQuote)) {
1473 1474 1475
      smlBuildInvalidDataMsg(&info->msgBuf, "only one quote", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
1476
    if (unlikely(valueLen == 0)) {
1477 1478 1479
      smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
1480 1481 1482
    if(unlikely(hasSlash)) {
      PROCESS_SLASH(value, valueLen)
    }
1483 1484

    SSmlKv kv = {.key = key, .keyLen = keyLen, .value = value, .length = valueLen};
1485 1486 1487
    int32_t ret = smlParseValue(&kv, &info->msgBuf);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
1488 1489 1490
    }

    if(info->dataFormat){
1491 1492
      //cnt begin 0, add ts so + 2
      if(unlikely(cnt + 2 > info->currSTableMeta->tableInfo.numOfColumns)){
1493 1494 1495
        info->dataFormat = false;
        info->reRun      = true;
        return TSDB_CODE_SUCCESS;
1496 1497
      }
      // bind data
1498 1499
      ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, cnt + 1);
      if (unlikely(ret != TSDB_CODE_SUCCESS)) {
1500 1501 1502 1503
        uError("smlBuildCol error, retry");
        info->dataFormat = false;
        info->reRun      = true;
        return TSDB_CODE_SUCCESS;
1504 1505
      }

1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535
      if(isSameMeasure){
        if(cnt >= taosArrayGetSize(preLineKV)) {
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
        SSmlKv *preKV = taosArrayGet(preLineKV, cnt);
        if(kv.type != preKV->type){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }

        if(unlikely(IS_VAR_DATA_TYPE(kv.type) && kv.length > preKV->length)){
          preKV->length = kv.length;
          SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, currElement->measure, currElement->measureLen, NULL);
          ASSERT(tableMeta != NULL);

          SSmlKv *oldKV = taosArrayGet(tableMeta->cols, cnt);
          oldKV->length = kv.length;
          info->needModifySchema = true;
        }
        if(unlikely(!IS_SAME_KEY)){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
      }else{
        if(isSuperKVInit){
          if(unlikely(cnt >= taosArrayGetSize(superKV))) {
1536 1537 1538 1539
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }
1540 1541
          SSmlKv *preKV = taosArrayGet(superKV, cnt);
          if(unlikely(kv.type != preKV->type)){
1542 1543 1544 1545 1546
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }

1547 1548 1549
          if(IS_VAR_DATA_TYPE(kv.type)){
            if(kv.length > preKV->length) {
              preKV->length = kv.length;
1550
            }else{
1551
              kv.length = preKV->length;
1552
            }
wmmhello's avatar
wmmhello 已提交
1553
            info->needModifySchema = true;
1554
          }
1555 1556 1557 1558
          if(unlikely(!IS_SAME_KEY)){
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
1559
          }
1560 1561
        }else{
          taosArrayPush(superKV, &kv);
1562
        }
1563
        taosArrayPush(preLineKV, &kv);
1564
      }
1565
    }else{
1566 1567 1568 1569
      if(currElement->colArray == NULL){
        currElement->colArray = taosArrayInit(16, sizeof(SSmlKv));
        taosArraySetSize(currElement->colArray, 1);
      }
1570 1571
      taosArrayPush(currElement->colArray, &kv);   //reserve for timestamp
    }
1572

1573 1574 1575 1576 1577
    cnt++;
    if(IS_SPACE(*sql)){
      break;
    }
    (*sql)++;
1578 1579
  }

1580 1581
  return TSDB_CODE_SUCCESS;
}
1582

1583 1584 1585 1586 1587
static int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements) {
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
  JUMP_SPACE(sql, sqlEnd)
  if (unlikely(*sql == COMMA)) return TSDB_CODE_SML_INVALID_DATA;
  elements->measure = sql;
wmmhello's avatar
wmmhello 已提交
1588

wmmhello's avatar
wmmhello 已提交
1589
  // parse measure
wmmhello's avatar
wmmhello 已提交
1590
  while (sql < sqlEnd) {
X
Xiaoyu Wang 已提交
1591
    if ((sql != elements->measure) && IS_SLASH_LETTER(sql)) {
wmmhello's avatar
wmmhello 已提交
1592 1593
      MOVE_FORWARD_ONE(sql, sqlEnd - sql);
      sqlEnd--;
wmmhello's avatar
wmmhello 已提交
1594 1595
      continue;
    }
X
Xiaoyu Wang 已提交
1596
    if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1597 1598 1599
      break;
    }

X
Xiaoyu Wang 已提交
1600
    if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
1601 1602
      break;
    }
wmmhello's avatar
wmmhello 已提交
1603 1604
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
1605
  elements->measureLen = sql - elements->measure;
1606
  if (unlikely(IS_INVALID_TABLE_LEN(elements->measureLen))) {
1607
    smlBuildInvalidDataMsg(&info->msgBuf, "measure is empty or too large than 192", NULL);
wmmhello's avatar
wmmhello 已提交
1608
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
wmmhello's avatar
wmmhello 已提交
1609
  }
wmmhello's avatar
wmmhello 已提交
1610

1611 1612 1613 1614 1615 1616 1617 1618 1619 1620
  // to get measureTagsLen before
  const char* tmp = sql;
  while (tmp < sqlEnd){
    if (IS_SPACE(tmp)) {
      break;
    }
    tmp++;
  }
  elements->measureTagsLen = tmp - elements->measure;

1621 1622 1623 1624 1625 1626 1627 1628
  bool isSameCTable = false;
  bool isSameMeasure = false;
  if(IS_SAME_CHILD_TABLE){
    isSameCTable = true;
    isSameMeasure = true;
  }else if(info->dataFormat) {
    isSameMeasure = IS_SAME_SUPER_TABLE;
  }
wmmhello's avatar
wmmhello 已提交
1629
  // parse tag
1630 1631
  if (*sql == COMMA) sql++;
  elements->tags = sql;
1632

1633 1634 1635 1636 1637 1638 1639 1640
  // tinfo != NULL means child table has never occur before
  int ret = smlParseTagKv(info, &sql, sqlEnd, elements, isSameMeasure, isSameCTable);
  if(unlikely(ret != TSDB_CODE_SUCCESS)){
    return ret;
  }
  if(unlikely(info->reRun)){
    return TSDB_CODE_SUCCESS;
  }
1641

1642
  sql = elements->measure + elements->measureTagsLen;
1643

1644
  elements->tagsLen = sql - elements->tags;
wmmhello's avatar
wmmhello 已提交
1645

wmmhello's avatar
wmmhello 已提交
1646
  // parse cols
wmmhello's avatar
wmmhello 已提交
1647
  JUMP_SPACE(sql, sqlEnd)
wmmhello's avatar
wmmhello 已提交
1648
  elements->cols = sql;
1649

1650
  ret = smlParseColKv(info, &sql, sqlEnd, elements, isSameMeasure, isSameCTable);
1651
  if(unlikely(ret != TSDB_CODE_SUCCESS)){
1652
    return ret;
wmmhello's avatar
wmmhello 已提交
1653
  }
1654

1655
  if(unlikely(info->reRun)){
1656
    return TSDB_CODE_SUCCESS;
1657
  }
1658

wmmhello's avatar
wmmhello 已提交
1659
  elements->colsLen = sql - elements->cols;
1660
  if (unlikely(elements->colsLen == 0)) {
1661
    smlBuildInvalidDataMsg(&info->msgBuf, "cols is empty", NULL);
wmmhello's avatar
wmmhello 已提交
1662 1663
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
1664

wmmhello's avatar
wmmhello 已提交
1665
  // parse timestamp
wmmhello's avatar
wmmhello 已提交
1666
  JUMP_SPACE(sql, sqlEnd)
wmmhello's avatar
wmmhello 已提交
1667
  elements->timestamp = sql;
wmmhello's avatar
wmmhello 已提交
1668 1669
  while (sql < sqlEnd) {
    if (isspace(*sql)) {
wmmhello's avatar
wmmhello 已提交
1670 1671 1672 1673
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
1674
  elements->timestampLen = sql - elements->timestamp;
wmmhello's avatar
wmmhello 已提交
1675

1676 1677 1678 1679
  int64_t ts = smlParseTS(info, elements->timestamp, elements->timestampLen);
  if (ts <= 0) {
    uError("SML:0x%" PRIx64 " smlParseTS error:%" PRId64, info->id, ts);
    return TSDB_CODE_INVALID_TIMESTAMP;
1680
  }
1681 1682
  // add ts to
  SSmlKv kv = { .key = TS, .keyLen = TS_LEN, .i = ts, .type = TSDB_DATA_TYPE_TIMESTAMP, .length = (int16_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
1683
  if(info->dataFormat){
1684
    smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 0);
1685
    smlBuildRow(info->currTableDataCtx);
1686 1687
  }else{
    taosArraySet(elements->colArray, 0, &kv);
1688
  }
1689
  info->preLine = *elements;
1690

1691
  return ret;
wmmhello's avatar
wmmhello 已提交
1692 1693
}

1694
static void smlParseTelnetElement(char **sql, char *sqlEnd, char **data, int32_t *len) {
wmmhello's avatar
wmmhello 已提交
1695
  while (*sql < sqlEnd) {
1696
    if (unlikely((**sql != SPACE && !(*data)))) {
1697
      *data = *sql;
1698
    } else if (unlikely(**sql == SPACE && *data)) {
1699 1700 1701 1702 1703 1704 1705
      *len = *sql - *data;
      break;
    }
    (*sql)++;
  }
}

1706
static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
1707
  if(is_same_child_table_telnet(elements, &info->preLine) == 0){
1708 1709 1710 1711 1712 1713 1714 1715 1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729
    return TSDB_CODE_SUCCESS;
  }

  bool isSameMeasure = IS_SAME_SUPER_TABLE;

  int     cnt = 0;
  SArray *preLineKV = info->preLineTagKV;
  bool    isSuperKVInit = true;
  SArray *superKV = NULL;
  if(info->dataFormat){
    if(!isSameMeasure){
      SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);

      if(unlikely(sMeta == NULL)){
        sMeta = smlBuildSTableMeta(info->dataFormat);
        STableMeta * pTableMeta = smlGetMeta(info, elements->measure, elements->measureLen);
        sMeta->tableMeta = pTableMeta;
        if(pTableMeta == NULL){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
1730
        nodeListSet(&info->superTables, elements->measure, elements->measureLen, sMeta, NULL);
1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741 1742 1743
      }
      info->currSTableMeta = sMeta->tableMeta;
      superKV = sMeta->tags;

      if(unlikely(taosArrayGetSize(superKV) == 0)){
        isSuperKVInit = false;
      }
      taosArraySetSize(preLineKV, 0);
    }
  }else{
    taosArraySetSize(preLineKV, 0);
  }

wmmhello's avatar
wmmhello 已提交
1744
  const char *sql = data;
wmmhello's avatar
wmmhello 已提交
1745 1746
  while (sql < sqlEnd) {
    JUMP_SPACE(sql, sqlEnd)
1747
    if (unlikely(*sql == '\0')) break;
wmmhello's avatar
wmmhello 已提交
1748

wmmhello's avatar
wmmhello 已提交
1749
    const char *key = sql;
X
Xiaoyu Wang 已提交
1750
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
1751 1752

    // parse key
wmmhello's avatar
wmmhello 已提交
1753
    while (sql < sqlEnd) {
1754
      if (unlikely(*sql == SPACE)) {
wmmhello's avatar
wmmhello 已提交
1755 1756 1757
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
1758
      if (unlikely(*sql == EQUAL)) {
wmmhello's avatar
wmmhello 已提交
1759 1760
        keyLen = sql - key;
        sql++;
1761 1762
        break;
      }
wmmhello's avatar
wmmhello 已提交
1763
      sql++;
1764
    }
wmmhello's avatar
wmmhello 已提交
1765

1766
    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
1767
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1768
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
1769
    }
1770 1771 1772 1773
//    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
//      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
//      return TSDB_CODE_TSC_DUP_NAMES;
//    }
1774 1775

    // parse value
wmmhello's avatar
wmmhello 已提交
1776
    const char *value = sql;
X
Xiaoyu Wang 已提交
1777
    int32_t     valueLen = 0;
wmmhello's avatar
wmmhello 已提交
1778
    while (sql < sqlEnd) {
wmmhello's avatar
wmmhello 已提交
1779
      // parse value
1780
      if (unlikely(*sql == SPACE)) {
1781 1782
        break;
      }
1783
      if (unlikely(*sql == EQUAL)) {
wmmhello's avatar
wmmhello 已提交
1784 1785 1786 1787
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
1788
    }
wmmhello's avatar
wmmhello 已提交
1789
    valueLen = sql - value;
wmmhello's avatar
wmmhello 已提交
1790

1791
    if (unlikely(valueLen == 0)) {
1792
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1793
      return TSDB_CODE_TSC_INVALID_VALUE;
1794
    }
wmmhello's avatar
wmmhello 已提交
1795

1796
    if (unlikely(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
wmmhello's avatar
wmmhello 已提交
1797 1798 1799
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }

1800 1801 1802 1803
    SSmlKv kv = {.key = key, .keyLen = keyLen, .value = value, .length = valueLen, .type = TSDB_DATA_TYPE_NCHAR};

    if(info->dataFormat){
      if(unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)){
1804 1805 1806
        info->dataFormat = false;
        info->reRun      = true;
        return TSDB_CODE_SUCCESS;
1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843
      }

      if(isSameMeasure){
        if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
        SSmlKv *preKV = taosArrayGet(preLineKV, cnt);
        if(unlikely(kv.length > preKV->length)){
          preKV->length = kv.length;
          SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
          ASSERT(tableMeta != NULL);

          SSmlKv *oldKV = taosArrayGet(tableMeta->tags, cnt);
          oldKV->length = kv.length;
          info->needModifySchema = true;
        }
        if(unlikely(!IS_SAME_KEY)){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
      }else{
        if(isSuperKVInit){
          if(unlikely(cnt >= taosArrayGetSize(superKV))) {
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }
          SSmlKv *preKV = taosArrayGet(superKV, cnt);
          if(unlikely(kv.length > preKV->length)) {
            preKV->length = kv.length;
          }else{
            kv.length = preKV->length;
          }
          info->needModifySchema = true;
1844

1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857 1858
          if(unlikely(!IS_SAME_KEY)){
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }
        }else{
          taosArrayPush(superKV, &kv);
        }
        taosArrayPush(preLineKV, &kv);
      }
    }else{
      taosArrayPush(preLineKV, &kv);
    }
    cnt++;
1859
  }
1860
  SSmlTableInfo *tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements, POINTER_BYTES, is_same_child_table_telnet);
1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877
  if (unlikely(tinfo == NULL)) {
    tinfo = smlBuildTableInfo(1, elements->measure, elements->measureLen);
    if (!tinfo) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
    for(int i = 0; i < taosArrayGetSize(preLineKV); i++){
      taosArrayPush(tinfo->tags, taosArrayGet(preLineKV, i));
    }
    smlSetCTableName(tinfo);
    if (info->dataFormat) {
      info->currSTableMeta->uid = tinfo->uid;
      tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
      if (tinfo->tableDataCtx == NULL) {
        smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
        return TSDB_CODE_SML_INVALID_DATA;
      }
    }
1878

1879 1880 1881
    SSmlLineInfo *key = taosMemoryMalloc(sizeof(SSmlLineInfo));
    *key = *elements;
    tinfo->key = key;
1882
    nodeListSet(&info->childTables, key, POINTER_BYTES, tinfo, is_same_child_table_telnet);
1883
  }
1884 1885
  info->currTableDataCtx = tinfo->tableDataCtx;

1886 1887
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
1888

1889
// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
1890
static int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements) {
X
Xiaoyu Wang 已提交
1891
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
1892 1893

  // parse metric
1894 1895
  smlParseTelnetElement(&sql, sqlEnd, &elements->measure, &elements->measureLen);
  if (unlikely((!(elements->measure) || IS_INVALID_TABLE_LEN(elements->measureLen)))) {
1896
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1897
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
1898 1899 1900
  }

  // parse timestamp
1901 1902
  smlParseTelnetElement(&sql, sqlEnd, &elements->timestamp, &elements->timestampLen);
  if (unlikely(!elements->timestamp || elements->timestampLen == 0)) {
1903 1904 1905 1906
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }

1907 1908 1909 1910 1911 1912
  bool needConverTime = false;  // get TS before parse tag(get meta), so need conver time
  if(info->dataFormat && info->currSTableMeta == NULL){
    needConverTime = true;
  }
  int64_t ts = smlParseTS(info, elements->timestamp, elements->timestampLen);
  if (unlikely(ts < 0)) {
1913
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
1914
    return TSDB_CODE_INVALID_TIMESTAMP;
1915
  }
1916
  SSmlKv kvTs = { .key = TS, .keyLen = TS_LEN, .i = ts, .type = TSDB_DATA_TYPE_TIMESTAMP, .length = (int16_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
1917 1918

  // parse value
1919 1920
  smlParseTelnetElement(&sql, sqlEnd, &elements->cols, &elements->colsLen);
  if (unlikely(!elements->cols || elements->colsLen == 0)) {
1921 1922 1923
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
    return TSDB_CODE_TSC_INVALID_VALUE;
  }
1924

1925 1926 1927 1928 1929
  SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN, .value = elements->cols, .length = elements->colsLen};
  if (smlParseNumber(&kv, &info->msgBuf)) {
    kv.length = (int16_t)tDataTypes[kv.type].bytes;
  }else{
    return TSDB_CODE_TSC_INVALID_VALUE;
1930
  }
1931
  JUMP_SPACE(sql, sqlEnd)
1932

1933 1934
  elements->tags = sql;
  elements->tagsLen = sqlEnd - sql;
1935 1936 1937

  int ret = smlParseTelnetTags(info, sql, sqlEnd, elements, &info->msgBuf);
  if (unlikely(ret != TSDB_CODE_SUCCESS)) {
1938
    return ret;
wmmhello's avatar
wmmhello 已提交
1939
  }
wmmhello's avatar
wmmhello 已提交
1940

1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960
  if(unlikely(info->reRun)){
    return TSDB_CODE_SUCCESS;
  }

  if(info->dataFormat){
    if(needConverTime) {
      kvTs.i = convertTimePrecision(kvTs.i, TSDB_TIME_PRECISION_NANO, info->currSTableMeta->tableInfo.precision);
    }
    ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kvTs, 0);
    if(ret == TSDB_CODE_SUCCESS){
      ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 1);
    }
    if(ret == TSDB_CODE_SUCCESS){
      ret = smlBuildRow(info->currTableDataCtx);
    }
    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
      smlBuildInvalidDataMsg(&info->msgBuf, "smlBuildCol error", NULL);
      return ret;
    }
  }else{
1961 1962 1963
    if(elements->colArray == NULL){
      elements->colArray = taosArrayInit(16, sizeof(SSmlKv));
    }
1964 1965 1966 1967 1968
    taosArrayPush(elements->colArray, &kvTs);
    taosArrayPush(elements->colArray, &kv);
  }
  info->preLine = *elements;

wmmhello's avatar
wmmhello 已提交
1969 1970 1971
  return TSDB_CODE_SUCCESS;
}

1972
static int32_t smlUpdateMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols, bool isTag, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
1973
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
1974
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
wmmhello's avatar
wmmhello 已提交
1975

wmmhello's avatar
wmmhello 已提交
1976
    int16_t *index = (int16_t *)taosHashGet(metaHash, kv->key, kv->keyLen);
X
Xiaoyu Wang 已提交
1977
    if (index) {
1978 1979 1980 1981 1982 1983 1984 1985
      SSmlKv *value = (SSmlKv *)taosArrayGet(metaArray, *index);
      if (isTag){
        if (kv->length > value->length) {
          value->length = kv->length;
        }
        continue;
      }
      if (kv->type != value->type) {
wmmhello's avatar
wmmhello 已提交
1986
        smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
wmmhello's avatar
wmmhello 已提交
1987
        return TSDB_CODE_SML_NOT_SAME_TYPE;
1988 1989 1990 1991
      }

      if (IS_VAR_DATA_TYPE(kv->type) && (kv->length > value->length)) {  // update string len, if bigger
        value->length = kv->length;
1992
      }
X
Xiaoyu Wang 已提交
1993
    } else {
wmmhello's avatar
wmmhello 已提交
1994 1995 1996
      size_t tmp = taosArrayGetSize(metaArray);
      ASSERT(tmp <= INT16_MAX);
      int16_t size = tmp;
1997
      taosArrayPush(metaArray, kv);
wmmhello's avatar
wmmhello 已提交
1998
      taosHashPut(metaHash, kv->key, kv->keyLen, &size, SHORT_BYTES);
1999
    }
wmmhello's avatar
wmmhello 已提交
2000
  }
wmmhello's avatar
wmmhello 已提交
2001

wmmhello's avatar
wmmhello 已提交
2002
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2003 2004
}

wmmhello's avatar
wmmhello 已提交
2005
static void smlDestroyTableInfo(SSmlTableInfo *tag) {
2006 2007 2008
  for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
    SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
    taosHashCleanup(kvHash);
wmmhello's avatar
wmmhello 已提交
2009
  }
2010

2011
  taosMemoryFree(tag->key);
2012 2013 2014 2015 2016
  taosArrayDestroy(tag->cols);
  taosArrayDestroy(tag->tags);
  taosMemoryFree(tag);
}

2017
static int32_t smlPushCols(SArray *colsArray, SArray *cols) {
2018
  SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
2019
  if (!kvHash) {
2020
    uError("SML:smlDealCols failed to allocate memory");
2021
    return TSDB_CODE_OUT_OF_MEMORY;
2022
  }
X
Xiaoyu Wang 已提交
2023
  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
2024
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
wmmhello's avatar
wmmhello 已提交
2025
    taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
2026 2027
  }

2028
  taosArrayPush(colsArray, &kvHash);
2029 2030 2031
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
2032 2033
void smlDestroyInfo(SSmlHandle *info) {
  if (!info) return;
2034 2035 2036
  qDestroyQuery(info->pQuery);

  // destroy info->childTables
wmmhello's avatar
wmmhello 已提交
2037 2038 2039 2040 2041 2042 2043 2044
  NodeList* tmp = info->childTables;
  while (tmp) {
    if(tmp->data.used) {
      smlDestroyTableInfo(tmp->data.value);
    }
    NodeList* t = tmp->next;
    taosMemoryFree(tmp);
    tmp = t;
2045 2046 2047
  }

  // destroy info->superTables
wmmhello's avatar
wmmhello 已提交
2048 2049 2050 2051 2052 2053 2054 2055
  tmp = info->superTables;
  while (tmp) {
    if(tmp->data.used) {
      smlDestroySTableMeta(tmp->data.value);
    }
    NodeList* t = tmp->next;
    taosMemoryFree(tmp);
    tmp = t;
2056 2057 2058 2059
  }

  // destroy info->pVgHash
  taosHashCleanup(info->pVgHash);
wmmhello's avatar
wmmhello 已提交
2060

2061 2062 2063
  taosArrayDestroy(info->preLineTagKV);
  taosArrayDestroy(info->preLineColKV);

wmmhello's avatar
wmmhello 已提交
2064 2065 2066 2067
  if(!info->dataFormat){
    for(int i = 0; i < info->lineNum; i++){
      taosArrayDestroy(info->lines[i].colArray);
    }
wmmhello's avatar
wmmhello 已提交
2068
    taosMemoryFree(info->lines);
wmmhello's avatar
wmmhello 已提交
2069
  }
wmmhello's avatar
wmmhello 已提交
2070

wmmhello's avatar
wmmhello 已提交
2071
  cJSON_Delete(info->root);
wmmhello's avatar
wmmhello 已提交
2072
  taosMemoryFreeClear(info);
wmmhello's avatar
wmmhello 已提交
2073
}
2074

wmmhello's avatar
wmmhello 已提交
2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086
static SSmlHandle *smlBuildSmlInfo(TAOS *taos) {
  int32_t     code = TSDB_CODE_SUCCESS;
  SSmlHandle *info = (SSmlHandle *)taosMemoryCalloc(1, sizeof(SSmlHandle));
  if (NULL == info) {
    return NULL;
  }
  info->taos = acquireTscObj(*(int64_t *)taos);
  code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
  if (code != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " get catalog error %d", info->id, code);
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
2087
  info->pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
wmmhello's avatar
wmmhello 已提交
2088 2089 2090
  info->id = smlGenId();
  info->pQuery = smlInitHandle();
  info->dataFormat = true;
2091

2092 2093 2094
  info->preLineTagKV = taosArrayInit(8, sizeof(SSmlKv));
  info->preLineColKV = taosArrayInit(8, sizeof(SSmlKv));

wmmhello's avatar
wmmhello 已提交
2095 2096
  if (NULL == info->pVgHash) {
    uError("create SSmlHandle failed");
2097 2098 2099 2100
    goto cleanup;
  }

  return info;
wmmhello's avatar
wmmhello 已提交
2101 2102

cleanup:
2103 2104 2105 2106 2107
  smlDestroyInfo(info);
  return NULL;
}

/************* TSDB_SML_JSON_PROTOCOL function start **************/
2108
static int32_t smlParseMetricFromJSON(SSmlHandle *info, cJSON *root, SSmlLineInfo *elements) {
2109 2110
  cJSON *metric = cJSON_GetObjectItem(root, "metric");
  if (!cJSON_IsString(metric)) {
X
Xiaoyu Wang 已提交
2111
    return TSDB_CODE_TSC_INVALID_JSON;
2112 2113
  }

2114 2115
  elements->measureLen = strlen(metric->valuestring);
  if (IS_INVALID_TABLE_LEN(elements->measureLen)) {
X
Xiaoyu Wang 已提交
2116
    uError("OTD:0x%" PRIx64 " Metric lenght is 0 or large than 192", info->id);
2117 2118 2119
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
  }

2120
  elements->measure = metric->valuestring;
wmmhello's avatar
wmmhello 已提交
2121
  return TSDB_CODE_SUCCESS;
2122 2123
}

2124
static int64_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int32_t toPrecision) {
2125
  int32_t size = cJSON_GetArraySize(root);
2126 2127 2128
  if (unlikely(size != OTD_JSON_SUB_FIELDS_NUM)) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);
    return -1;
2129 2130 2131
  }

  cJSON *value = cJSON_GetObjectItem(root, "value");
2132 2133 2134
  if (unlikely(!cJSON_IsNumber(value))) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);
    return -1;
2135 2136 2137
  }

  cJSON *type = cJSON_GetObjectItem(root, "type");
2138 2139 2140
  if (unlikely(!cJSON_IsString(type))) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate json", NULL);
    return -1;
2141 2142 2143
  }

  double timeDouble = value->valuedouble;
2144
  if (unlikely(smlDoubleToInt64OverFlow(timeDouble))) {
2145
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
2146
    return -1;
2147
  }
wmmhello's avatar
wmmhello 已提交
2148 2149

  if (timeDouble == 0) {
2150
    return taosGetTimestampNs()/smlFactorNS[toPrecision];
wmmhello's avatar
wmmhello 已提交
2151 2152 2153
  }

  if (timeDouble < 0) {
2154
    return timeDouble;
2155 2156
  }

2157
  int64_t tsInt64 = timeDouble;
2158
  size_t typeLen = strlen(type->valuestring);
wmmhello's avatar
wmmhello 已提交
2159
  if (typeLen == 1 && (type->valuestring[0] == 's' || type->valuestring[0] == 'S')) {
X
Xiaoyu Wang 已提交
2160
    // seconds
2161 2162 2163
    int8_t fromPrecision = TSDB_TIME_PRECISION_SECONDS;
    if(smlFactorS[toPrecision] < INT64_MAX / tsInt64){
      return tsInt64 * smlFactorS[toPrecision];
2164
    }
2165
    return -1;
wmmhello's avatar
wmmhello 已提交
2166
  } else if (typeLen == 2 && (type->valuestring[1] == 's' || type->valuestring[1] == 'S')) {
2167 2168
    switch (type->valuestring[0]) {
      case 'm':
wmmhello's avatar
wmmhello 已提交
2169
      case 'M':
X
Xiaoyu Wang 已提交
2170
        // milliseconds
2171
        return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_MILLI, toPrecision);
2172 2173
        break;
      case 'u':
wmmhello's avatar
wmmhello 已提交
2174
      case 'U':
X
Xiaoyu Wang 已提交
2175
        // microseconds
2176
        return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_MICRO, toPrecision);
2177 2178
        break;
      case 'n':
wmmhello's avatar
wmmhello 已提交
2179
      case 'N':
2180
        return convertTimePrecision(tsInt64, TSDB_TIME_PRECISION_NANO, toPrecision);
2181 2182
        break;
      default:
2183
        return -1;
2184 2185
    }
  } else {
2186
    return -1;
2187 2188 2189 2190 2191 2192 2193 2194 2195 2196 2197 2198
  }
}

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

2199
static int64_t smlParseTSFromJSON(SSmlHandle *info, cJSON *root) {
X
Xiaoyu Wang 已提交
2200
  // Timestamp must be the first KV to parse
2201
  int32_t toPrecision = info->currSTableMeta ? info->currSTableMeta->tableInfo.precision : TSDB_TIME_PRECISION_NANO;
2202 2203
  cJSON *timestamp = cJSON_GetObjectItem(root, "timestamp");
  if (cJSON_IsNumber(timestamp)) {
X
Xiaoyu Wang 已提交
2204
    // timestamp value 0 indicates current system time
2205
    double timeDouble = timestamp->valuedouble;
2206
    if (unlikely(smlDoubleToInt64OverFlow(timeDouble))) {
2207
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
2208
      return -1;
2209
    }
wmmhello's avatar
wmmhello 已提交
2210

2211 2212 2213 2214 2215 2216
    if (unlikely(timeDouble < 0)) {
      smlBuildInvalidDataMsg(&info->msgBuf,
                             "timestamp is negative", NULL);
      return timeDouble;
    }else if (unlikely(timeDouble == 0)) {
      return taosGetTimestampNs()/smlFactorNS[toPrecision];
2217
    }
2218

2219
    uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);
2220 2221 2222 2223 2224
    int8_t fromPrecision = smlGetTsTypeByLen(tsLen);
    if (unlikely(fromPrecision == -1)) {
      smlBuildInvalidDataMsg(&info->msgBuf,
                             "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", NULL);
      return -1;
2225
    }
2226 2227 2228 2229 2230 2231 2232 2233 2234
    int64_t tsInt64 = timeDouble;
    if(fromPrecision == TSDB_TIME_PRECISION_SECONDS){
      if(smlFactorS[toPrecision] < INT64_MAX / tsInt64){
        return tsInt64 * smlFactorS[toPrecision];
      }
      return -1;
    }else{
      return convertTimePrecision(timeDouble, fromPrecision, toPrecision);
    }
2235
  } else if (cJSON_IsObject(timestamp)) {
2236
    return smlParseTSFromJSONObj(info, timestamp, toPrecision);
2237
  } else {
2238 2239 2240
    smlBuildInvalidDataMsg(&info->msgBuf,
                           "invalidate json", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
2241
  }
wmmhello's avatar
wmmhello 已提交
2242 2243
}

X
Xiaoyu Wang 已提交
2244
static int32_t smlConvertJSONBool(SSmlKv *pVal, char *typeStr, cJSON *value) {
2245 2246 2247 2248 2249 2250 2251
  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 已提交
2252

2253 2254
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
2255

X
Xiaoyu Wang 已提交
2256 2257 2258
static int32_t smlConvertJSONNumber(SSmlKv *pVal, char *typeStr, cJSON *value) {
  // tinyint
  if (strcasecmp(typeStr, "i8") == 0 || strcasecmp(typeStr, "tinyint") == 0) {
2259 2260 2261 2262 2263 2264 2265 2266 2267
    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 已提交
2268 2269
  // smallint
  if (strcasecmp(typeStr, "i16") == 0 || strcasecmp(typeStr, "smallint") == 0) {
2270 2271 2272 2273 2274 2275 2276 2277 2278
    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 已提交
2279 2280
  // int
  if (strcasecmp(typeStr, "i32") == 0 || strcasecmp(typeStr, "int") == 0) {
2281 2282 2283 2284 2285 2286 2287 2288 2289
    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 已提交
2290 2291
  // bigint
  if (strcasecmp(typeStr, "i64") == 0 || strcasecmp(typeStr, "bigint") == 0) {
2292 2293
    pVal->type = TSDB_DATA_TYPE_BIGINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
X
Xiaoyu Wang 已提交
2294
    if (value->valuedouble >= (double)INT64_MAX) {
2295
      pVal->i = INT64_MAX;
X
Xiaoyu Wang 已提交
2296
    } else if (value->valuedouble <= (double)INT64_MIN) {
2297
      pVal->i = INT64_MIN;
X
Xiaoyu Wang 已提交
2298
    } else {
2299
      pVal->i = value->valuedouble;
2300 2301 2302
    }
    return TSDB_CODE_SUCCESS;
  }
X
Xiaoyu Wang 已提交
2303 2304
  // float
  if (strcasecmp(typeStr, "f32") == 0 || strcasecmp(typeStr, "float") == 0) {
2305 2306 2307
    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 已提交
2308
    }
2309 2310 2311 2312 2313
    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 已提交
2314 2315
  // double
  if (strcasecmp(typeStr, "f64") == 0 || strcasecmp(typeStr, "double") == 0) {
2316 2317 2318 2319
    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 已提交
2320 2321
  }

X
Xiaoyu Wang 已提交
2322
  // if reach here means type is unsupported
2323 2324 2325
  uError("OTD:invalid type(%s) for JSON Number", typeStr);
  return TSDB_CODE_TSC_INVALID_JSON_TYPE;
}
2326

X
Xiaoyu Wang 已提交
2327
static int32_t smlConvertJSONString(SSmlKv *pVal, char *typeStr, cJSON *value) {
2328 2329 2330 2331 2332 2333 2334 2335 2336
  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 已提交
2337

2338
  if (pVal->type == TSDB_DATA_TYPE_BINARY && pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
wmmhello's avatar
wmmhello 已提交
2339 2340
    return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
  }
2341 2342
  if (pVal->type == TSDB_DATA_TYPE_NCHAR &&
      pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
wmmhello's avatar
wmmhello 已提交
2343 2344 2345
    return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
  }

wmmhello's avatar
wmmhello 已提交
2346 2347
  pVal->value = value->valuestring;
  return TSDB_CODE_SUCCESS;
2348 2349 2350 2351 2352 2353 2354 2355 2356 2357 2358 2359 2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373
}

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 已提交
2374
      }
2375
      break;
wmmhello's avatar
wmmhello 已提交
2376
    }
2377 2378 2379 2380 2381 2382 2383 2384 2385 2386 2387 2388 2389 2390 2391 2392
    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 已提交
2393
  }
2394 2395

  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2396 2397
}

2398 2399 2400 2401 2402 2403 2404 2405
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 已提交
2406
    }
2407 2408 2409 2410 2411 2412 2413 2414 2415 2416
    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 已提交
2417

X
Xiaoyu Wang 已提交
2418
      char *tsDefaultJSONStrType = "nchar";  // todo
2419 2420
      smlConvertJSONString(kv, tsDefaultJSONStrType, root);
      break;
wmmhello's avatar
wmmhello 已提交
2421
    }
2422 2423 2424 2425 2426 2427 2428 2429 2430 2431
    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 已提交
2432
  }
2433 2434 2435 2436

  return TSDB_CODE_SUCCESS;
}

2437
static int32_t smlParseColsFromJSON(cJSON *root, SSmlKv *kv) {
2438 2439 2440 2441 2442
  cJSON *metricVal = cJSON_GetObjectItem(root, "value");
  if (metricVal == NULL) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

2443
  int32_t ret = smlParseValueFromJSON(metricVal, kv);
2444 2445 2446
  if (ret != TSDB_CODE_SUCCESS) {
    return ret;
  }
2447

2448
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2449 2450
}

2451
static int32_t smlParseTagsFromJSON(SSmlHandle *info, cJSON *root, SSmlLineInfo *elements) {
2452
  int32_t ret = TSDB_CODE_SUCCESS;
2453

2454
  cJSON *tags = cJSON_GetObjectItem(root, "tags");
2455
  if (unlikely(tags == NULL || tags->type != cJSON_Object)) {
2456 2457
    return TSDB_CODE_TSC_INVALID_JSON;
  }
wmmhello's avatar
wmmhello 已提交
2458

2459 2460 2461 2462 2463
  // add measure to tags to identify one child table
  cJSON *cMeasure = cJSON_AddStringToObject(tags, JSON_METERS_NAME, elements->measure);
  if(unlikely(cMeasure == NULL)){
    return TSDB_CODE_TSC_INVALID_JSON;
  }
2464
  elements->tags = (char*)tags;
2465
  if(is_same_child_table_json(elements->tags, info->preLine.tags) == 0){
2466
    cJSON_DeleteItemFromObjectCaseSensitive(tags, JSON_METERS_NAME);
2467 2468
    return TSDB_CODE_SUCCESS;
  }
2469
  cJSON_DeleteItemFromObjectCaseSensitive(tags, JSON_METERS_NAME);
2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485 2486 2487 2488 2489

  bool isSameMeasure = IS_SAME_SUPER_TABLE;

  int     cnt = 0;
  SArray *preLineKV = info->preLineTagKV;
  bool    isSuperKVInit = true;
  SArray *superKV = NULL;
  if(info->dataFormat){
    if(unlikely(!isSameMeasure)){
      SSmlSTableMeta *sMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);

      if(unlikely(sMeta == NULL)){
        sMeta = smlBuildSTableMeta(info->dataFormat);
        STableMeta * pTableMeta = smlGetMeta(info, elements->measure, elements->measureLen);
        sMeta->tableMeta = pTableMeta;
        if(pTableMeta == NULL){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
2490
        nodeListSet(&info->superTables, elements->measure, elements->measureLen, sMeta, NULL);
2491 2492 2493 2494 2495 2496 2497 2498 2499 2500 2501 2502 2503
      }
      info->currSTableMeta = sMeta->tableMeta;
      superKV = sMeta->tags;

      if(unlikely(taosArrayGetSize(superKV) == 0)){
        isSuperKVInit = false;
      }
      taosArraySetSize(preLineKV, 0);
    }
  }else{
    taosArraySetSize(preLineKV, 0);
  }

2504 2505 2506
  int32_t tagNum = cJSON_GetArraySize(tags);
  for (int32_t i = 0; i < tagNum; ++i) {
    cJSON *tag = cJSON_GetArrayItem(tags, i);
2507
    if (unlikely(tag == NULL)) {
2508
      return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
2509
    }
2510
    if(unlikely(tag == cMeasure)) continue;
2511
    size_t keyLen = strlen(tag->string);
2512
    if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
2513 2514 2515 2516
      uError("OTD:Tag key length is 0 or too large than 64");
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
    }

2517
    // add kv to SSmlKv
2518
    SSmlKv kv ={.key = tag->string, .keyLen = keyLen};
X
Xiaoyu Wang 已提交
2519
    // value
2520
    ret = smlParseValueFromJSON(tag, &kv);
2521
    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
2522 2523
      return ret;
    }
2524 2525 2526

    if(info->dataFormat){
      if(unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)){
2527 2528 2529
        info->dataFormat = false;
        info->reRun      = true;
        return TSDB_CODE_SUCCESS;
2530 2531 2532 2533 2534 2535 2536 2537 2538 2539 2540 2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552 2553 2554 2555 2556 2557 2558 2559 2560 2561 2562 2563 2564 2565 2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580 2581 2582 2583
      }

      if(isSameMeasure){
        if(unlikely(cnt >= taosArrayGetSize(preLineKV))) {
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
        SSmlKv *preKV = taosArrayGet(preLineKV, cnt);
        if(unlikely(kv.length > preKV->length)){
          preKV->length = kv.length;
          SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
          ASSERT(tableMeta != NULL);

          SSmlKv *oldKV = taosArrayGet(tableMeta->tags, cnt);
          oldKV->length = kv.length;
          info->needModifySchema = true;
        }
        if(unlikely(!IS_SAME_KEY)){
          info->dataFormat = false;
          info->reRun      = true;
          return TSDB_CODE_SUCCESS;
        }
      }else{
        if(isSuperKVInit){
          if(unlikely(cnt >= taosArrayGetSize(superKV))) {
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }
          SSmlKv *preKV = taosArrayGet(superKV, cnt);
          if(unlikely(kv.length > preKV->length)) {
            preKV->length = kv.length;
          }else{
            kv.length = preKV->length;
          }
          info->needModifySchema = true;

          if(unlikely(!IS_SAME_KEY)){
            info->dataFormat = false;
            info->reRun      = true;
            return TSDB_CODE_SUCCESS;
          }
        }else{
          taosArrayPush(superKV, &kv);
        }
        taosArrayPush(preLineKV, &kv);
      }
    }else{
      taosArrayPush(preLineKV, &kv);
    }
    cnt++;
  }

2584 2585 2586 2587 2588 2589 2590 2591 2592 2593 2594 2595 2596 2597 2598 2599 2600
  SSmlTableInfo *tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements->tags, POINTER_BYTES, is_same_child_table_json);
  if (unlikely(tinfo == NULL)) {
    tinfo = smlBuildTableInfo(1, elements->measure, elements->measureLen);
    if (unlikely(!tinfo)) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
    for (int i = 0; i < taosArrayGetSize(preLineKV); i++) {
      taosArrayPush(tinfo->tags, taosArrayGet(preLineKV, i));
    }
    smlSetCTableName(tinfo);
    if (info->dataFormat) {
      info->currSTableMeta->uid = tinfo->uid;
      tinfo->tableDataCtx = smlInitTableDataCtx(info->pQuery, info->currSTableMeta);
      if (tinfo->tableDataCtx == NULL) {
        smlBuildInvalidDataMsg(&info->msgBuf, "smlInitTableDataCtx error", NULL);
        return TSDB_CODE_SML_INVALID_DATA;
      }
2601
    }
wmmhello's avatar
wmmhello 已提交
2602

2603
    nodeListSet(&info->childTables, tags, POINTER_BYTES, tinfo, is_same_child_table_json);
2604 2605
  }
  info->currTableDataCtx = tinfo->tableDataCtx;
2606

2607
  return ret;
wmmhello's avatar
wmmhello 已提交
2608 2609
}

2610
static int32_t smlParseJSONString(SSmlHandle *info, cJSON *root, SSmlLineInfo *elements) {
2611
  int32_t ret = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2612

2613
  int32_t size = cJSON_GetArraySize(root);
X
Xiaoyu Wang 已提交
2614
  // outmost json fields has to be exactly 4
2615
  if (unlikely(size != OTD_JSON_FIELDS_NUM)) {
X
Xiaoyu Wang 已提交
2616
    uError("OTD:0x%" PRIx64 " Invalid number of JSON fields in data point %d", info->id, size);
2617
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
2618
  }
2619

X
Xiaoyu Wang 已提交
2620
  // Parse metric
2621 2622
  ret = smlParseMetricFromJSON(info, root, elements);
  if (unlikely(ret != TSDB_CODE_SUCCESS)) {
X
Xiaoyu Wang 已提交
2623
    uError("OTD:0x%" PRIx64 " Unable to parse metric from JSON payload", info->id);
2624
    return ret;
wmmhello's avatar
wmmhello 已提交
2625
  }
X
Xiaoyu Wang 已提交
2626
  uDebug("OTD:0x%" PRIx64 " Parse metric from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2627

X
Xiaoyu Wang 已提交
2628
  // Parse metric value
2629 2630 2631
  SSmlKv kv = {.key = VALUE, .keyLen = VALUE_LEN};
  ret = smlParseColsFromJSON(root, &kv);
  if (unlikely(ret)) {
X
Xiaoyu Wang 已提交
2632
    uError("OTD:0x%" PRIx64 " Unable to parse metric value from JSON payload", info->id);
2633
    return ret;
2634
  }
X
Xiaoyu Wang 已提交
2635
  uDebug("OTD:0x%" PRIx64 " Parse metric value from JSON payload finished", info->id);
2636

2637 2638 2639
  // Parse tags
  ret = smlParseTagsFromJSON(info, root, elements);
  if (unlikely(ret)) {
X
Xiaoyu Wang 已提交
2640
    uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
2641
    return ret;
2642
  }
X
Xiaoyu Wang 已提交
2643
  uDebug("OTD:0x%" PRIx64 " Parse tags from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2644

2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655 2656 2657 2658 2659 2660 2661 2662 2663 2664 2665 2666 2667 2668 2669 2670 2671
  if(unlikely(info->reRun)){
    return TSDB_CODE_SUCCESS;
  }

  // Parse timestamp
  // notice!!! put ts back to tag to ensure get meta->precision
  int64_t ts = smlParseTSFromJSON(info, root);
  if (unlikely(ts < 0)) {
    uError("OTD:0x%" PRIx64 " Unable to parse timestamp from JSON payload", info->id);
    return TSDB_CODE_INVALID_TIMESTAMP;
  }
  uDebug("OTD:0x%" PRIx64 " Parse timestamp from JSON payload finished", info->id);
  SSmlKv kvTs = { .key = TS, .keyLen = TS_LEN, .i = ts, .type = TSDB_DATA_TYPE_TIMESTAMP, .length = (int16_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};

  if(info->dataFormat){
    ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kvTs, 0);
    if(ret == TSDB_CODE_SUCCESS){
      ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 1);
    }
    if(ret == TSDB_CODE_SUCCESS){
      ret = smlBuildRow(info->currTableDataCtx);
    }
    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
      smlBuildInvalidDataMsg(&info->msgBuf, "smlBuildCol error", NULL);
      return ret;
    }
  }else{
2672 2673 2674
    if(elements->colArray == NULL){
      elements->colArray = taosArrayInit(16, sizeof(SSmlKv));
    }
2675 2676 2677 2678 2679
    taosArrayPush(elements->colArray, &kvTs);
    taosArrayPush(elements->colArray, &kv);
  }
  info->preLine = *elements;

2680
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2681
}
2682
/************* TSDB_SML_JSON_PROTOCOL function end **************/
2683 2684 2685
static int32_t smlParseLineBottom(SSmlHandle *info) {
  if(info->dataFormat) return TSDB_CODE_SUCCESS;

wmmhello's avatar
wmmhello 已提交
2686 2687
  for(int32_t i = 0; i < info->lineNum; i ++){
    SSmlLineInfo* elements = info->lines + i;
2688
    SSmlTableInfo *tinfo = NULL;
2689
    if(info->protocol == TSDB_SML_LINE_PROTOCOL){
2690
      tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements->measure, elements->measureTagsLen, NULL);
2691 2692
    }else if(info->protocol == TSDB_SML_TELNET_PROTOCOL){
      tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements, POINTER_BYTES, is_same_child_table_telnet);
2693 2694 2695 2696
    }else{
      tinfo = (SSmlTableInfo *)nodeListGet(info->childTables, elements->tags, POINTER_BYTES, is_same_child_table_json);
    }

wmmhello's avatar
wmmhello 已提交
2697
    if(tinfo == NULL){
2698 2699 2700
      uError("SML:0x%" PRIx64 "get oneTable failed, line num:%d", info->id, i);
      smlBuildInvalidDataMsg(&info->msgBuf, "get oneTable failed", elements->measure);
      return TSDB_CODE_SML_INVALID_DATA;
2701
    }
wmmhello's avatar
wmmhello 已提交
2702

2703 2704 2705 2706 2707
    if (taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS) {
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
      return TSDB_CODE_PAR_INVALID_TAGS_NUM;
    }

2708 2709 2710
    if (taosArrayGetSize(elements->colArray) + taosArrayGetSize(tinfo->tags) > TSDB_MAX_COLUMNS) {
      smlBuildInvalidDataMsg(&info->msgBuf, "too many columns than 4096", NULL);
      return TSDB_CODE_PAR_TOO_MANY_COLUMNS;
wmmhello's avatar
wmmhello 已提交
2711
    }
wmmhello's avatar
wmmhello 已提交
2712

2713
    int ret = smlPushCols(tinfo->cols, elements->colArray);
X
Xiaoyu Wang 已提交
2714
    if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2715 2716 2717
      return ret;
    }

2718
    SSmlSTableMeta *tableMeta = (SSmlSTableMeta *)nodeListGet(info->superTables, elements->measure, elements->measureLen, NULL);
2719
    if (tableMeta) {  // update meta
wmmhello's avatar
wmmhello 已提交
2720
      ret = smlUpdateMeta(tableMeta->colHash, tableMeta->cols, elements->colArray, false, &info->msgBuf);
2721
      if (ret == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2722
        ret = smlUpdateMeta(tableMeta->tagHash, tableMeta->tags, tinfo->tags, true, &info->msgBuf);
2723 2724 2725 2726 2727
      }
      if (ret != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
        return ret;
      }
X
Xiaoyu Wang 已提交
2728
    } else {
2729 2730 2731 2732 2733
      ret = smlJudgeDupColName(elements->colArray, tinfo->tags, &info->msgBuf);
      if (ret != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
        return ret;
      }
wmmhello's avatar
wmmhello 已提交
2734

2735 2736 2737
      SSmlSTableMeta *meta = smlBuildSTableMeta(info->dataFormat);
      smlInsertMeta(meta->tagHash, meta->tags, tinfo->tags);
      smlInsertMeta(meta->colHash, meta->cols, elements->colArray);
2738
      nodeListSet(&info->superTables, elements->measure, elements->measureLen, meta, NULL);
wmmhello's avatar
wmmhello 已提交
2739
    }
wmmhello's avatar
wmmhello 已提交
2740
  }
2741

wmmhello's avatar
wmmhello 已提交
2742 2743 2744
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2745
static int32_t smlParseJSON(SSmlHandle *info, char *payload) {
2746 2747
  int32_t payloadNum = 0;
  int32_t ret = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2748

2749
  if (unlikely(payload == NULL)) {
X
Xiaoyu Wang 已提交
2750
    uError("SML:0x%" PRIx64 " empty JSON Payload", info->id);
2751
    return TSDB_CODE_TSC_INVALID_JSON;
2752
  }
2753

wmmhello's avatar
wmmhello 已提交
2754
  info->root = cJSON_Parse(payload);
2755
  if (unlikely(info->root == NULL)) {
X
Xiaoyu Wang 已提交
2756
    uError("SML:0x%" PRIx64 " parse json failed:%s", info->id, payload);
2757 2758
    return TSDB_CODE_TSC_INVALID_JSON;
  }
X
Xiaoyu Wang 已提交
2759
  // multiple data points must be sent in JSON array
2760
  if (cJSON_IsArray(info->root)) {
wmmhello's avatar
wmmhello 已提交
2761
    payloadNum = cJSON_GetArraySize(info->root);
2762 2763
  } else if (cJSON_IsObject(info->root)) {
    payloadNum = 1;
2764
  } else {
X
Xiaoyu Wang 已提交
2765
    uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2766
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
2767
  }
wmmhello's avatar
wmmhello 已提交
2768

2769 2770
  int32_t i = 0;
  while (i < payloadNum) {
wmmhello's avatar
wmmhello 已提交
2771
    cJSON *dataPoint = (payloadNum == 1 && cJSON_IsObject(info->root)) ? info->root : cJSON_GetArrayItem(info->root, i);
2772 2773 2774 2775 2776 2777 2778
    if(info->dataFormat) {
      SSmlLineInfo element = {0};
      ret = smlParseJSONString(info, dataPoint, &element);
    }else{
      ret = smlParseJSONString(info, dataPoint, info->lines + i);
    }
    if (unlikely(ret != TSDB_CODE_SUCCESS)) {
X
Xiaoyu Wang 已提交
2779
      uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2780 2781 2782 2783 2784 2785 2786 2787 2788 2789 2790 2791 2792 2793 2794 2795 2796 2797 2798 2799 2800 2801 2802 2803 2804 2805 2806 2807 2808 2809 2810 2811
      return ret;
    }

    if(unlikely(info->reRun)){
      i = 0;
      info->reRun = false;
      // clear info->childTables
      NodeList* pList = info->childTables;
      while (pList) {
        if(pList->data.used) {
          smlDestroyTableInfo(pList->data.value);
          pList->data.used = false;
        }
        pList = pList->next;
      }

      // clear info->superTables
      pList = info->superTables;
      while (pList) {
        if(pList->data.used) {
          smlDestroySTableMeta(pList->data.value);
          pList->data.used = false;
        }
        pList = pList->next;
      }

      if(unlikely(info->lines != NULL)){
        uError("SML:0x%" PRIx64 " info->lines != NULL", info->id);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      info->lineNum = payloadNum;
      info->lines = taosMemoryCalloc(info->lineNum, sizeof(SSmlLineInfo));
2812

2813
      memset(&info->preLine, 0, sizeof(SSmlLineInfo));
2814 2815 2816
      info->currSTableMeta = NULL;
      info->currTableDataCtx = NULL;

2817 2818 2819 2820
      SVnodeModifOpStmt* stmt= (SVnodeModifOpStmt*)(info->pQuery->pRoot);
      stmt->freeHashFunc(stmt->pTableBlockHashObj);
      stmt->pTableBlockHashObj = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
      continue;
2821
    }
2822
    i++;
2823 2824
  }

2825
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2826
}
2827

X
Xiaoyu Wang 已提交
2828
static int32_t smlInsertData(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
2829 2830
  int32_t code = TSDB_CODE_SUCCESS;

wmmhello's avatar
wmmhello 已提交
2831 2832 2833
  NodeList* tmp = info->childTables;
  while (tmp) {
    SSmlTableInfo *tableData = (SSmlTableInfo *)tmp->data.value;
wmmhello's avatar
wmmhello 已提交
2834 2835

    SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
wmmhello's avatar
wmmhello 已提交
2836
    tstrncpy(pName.dbname, info->pRequest->pDb, sizeof(pName.dbname));
wmmhello's avatar
wmmhello 已提交
2837
    memcpy(pName.tname, tableData->childTableName, strlen(tableData->childTableName));
D
dapan1121 已提交
2838 2839 2840 2841 2842 2843

    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 已提交
2844

wmmhello's avatar
wmmhello 已提交
2845
    SVgroupInfo vg;
D
dapan1121 已提交
2846
    code = catalogGetTableHashVgroup(info->pCatalog, &conn, &pName, &vg);
2847
    if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2848
      uError("SML:0x%" PRIx64 " catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName);
wmmhello's avatar
wmmhello 已提交
2849 2850
      return code;
    }
X
Xiaoyu Wang 已提交
2851
    taosHashPut(info->pVgHash, (const char *)&vg.vgId, sizeof(vg.vgId), (char *)&vg, sizeof(vg));
wmmhello's avatar
wmmhello 已提交
2852

wmmhello's avatar
wmmhello 已提交
2853
    SSmlSTableMeta *pMeta =
2854
        (SSmlSTableMeta *)nodeListGet(info->superTables, tableData->sTableName, tableData->sTableNameLen, NULL);
wmmhello's avatar
wmmhello 已提交
2855
    ASSERT(NULL != pMeta);
wmmhello's avatar
wmmhello 已提交
2856

2857
    // use tablemeta of stable to save vgid and uid of child table
wmmhello's avatar
wmmhello 已提交
2858 2859
    pMeta->tableMeta->vgId = vg.vgId;
    pMeta->tableMeta->uid = tableData->uid;  // one table merge data block together according uid
wmmhello's avatar
wmmhello 已提交
2860

wmmhello's avatar
wmmhello 已提交
2861 2862
    code = smlBindData(info->pQuery, info->dataFormat, tableData->tags, pMeta->cols, tableData->cols,
                       pMeta->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen,
2863
                       info->ttl, info->msgBuf.buf, info->msgBuf.len);
X
Xiaoyu Wang 已提交
2864 2865
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
wmmhello's avatar
wmmhello 已提交
2866 2867
      return code;
    }
wmmhello's avatar
wmmhello 已提交
2868
    tmp = tmp->next;
wmmhello's avatar
wmmhello 已提交
2869
  }
wmmhello's avatar
wmmhello 已提交
2870

wmmhello's avatar
wmmhello 已提交
2871
  code = smlBuildOutput(info->pQuery, info->pVgHash);
2872
  if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2873
    uError("SML:0x%" PRIx64 " smlBuildOutput failed", info->id);
2874 2875
    return code;
  }
2876 2877
  info->cost.insertRpcTime = taosGetTimestampUs();

2878 2879 2880
  SAppClusterSummary *pActivity = &info->taos->pAppInfo->summary;
  atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);

wmmhello's avatar
wmmhello 已提交
2881 2882
  launchQueryImpl(info->pRequest, info->pQuery, true, NULL);
  return info->pRequest->code;
wmmhello's avatar
wmmhello 已提交
2883 2884
}

X
Xiaoyu Wang 已提交
2885 2886
static void smlPrintStatisticInfo(SSmlHandle *info) {
  uError("SML:0x%" PRIx64
2887
             " 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 已提交
2888
        parse cost:%" PRId64 ",schema cost:%" PRId64 ",bind cost:%" PRId64 ",rpc cost:%" PRId64 ",total cost:%" PRId64
2889
             "",
X
Xiaoyu Wang 已提交
2890
         info->id, info->cost.code, info->cost.lineNum, info->cost.numOfSTables, info->cost.numOfCTables,
2891 2892
         info->cost.numOfCreateSTables, info->cost.numOfAlterTagSTables, info->cost.numOfAlterColSTables,
         info->cost.schemaTime - info->cost.parseTime,
X
Xiaoyu Wang 已提交
2893 2894
         info->cost.insertBindTime - info->cost.schemaTime, info->cost.insertRpcTime - info->cost.insertBindTime,
         info->cost.endTime - info->cost.insertRpcTime, info->cost.endTime - info->cost.parseTime);
2895 2896
}

dengyihao's avatar
dengyihao 已提交
2897
static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char *rawLineEnd, int numLines) {
wmmhello's avatar
wmmhello 已提交
2898
  int32_t code = TSDB_CODE_SUCCESS;
2899
  if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
dengyihao's avatar
dengyihao 已提交
2900
    if (lines) {
wmmhello's avatar
wmmhello 已提交
2901
      code = smlParseJSON(info, *lines);
dengyihao's avatar
dengyihao 已提交
2902
    } else if (rawLine) {
wmmhello's avatar
wmmhello 已提交
2903 2904
      code = smlParseJSON(info, rawLine);
    }
2905
    if (code != TSDB_CODE_SUCCESS) {
dengyihao's avatar
dengyihao 已提交
2906
      uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, lines ? *lines : rawLine);
2907 2908
      return code;
    }
wmmhello's avatar
wmmhello 已提交
2909
    return code;
wmmhello's avatar
wmmhello 已提交
2910
  }
wmmhello's avatar
wmmhello 已提交
2911

2912
  char *oldRaw = rawLine;
2913 2914
  int32_t i = 0;
  while (i < numLines) {
wmmhello's avatar
wmmhello 已提交
2915
    char *tmp = NULL;
dengyihao's avatar
dengyihao 已提交
2916 2917
    int   len = 0;
    if (lines) {
wmmhello's avatar
wmmhello 已提交
2918 2919
      tmp = lines[i];
      len = strlen(tmp);
dengyihao's avatar
dengyihao 已提交
2920
    } else if (rawLine) {
wmmhello's avatar
wmmhello 已提交
2921
      tmp = rawLine;
dengyihao's avatar
dengyihao 已提交
2922 2923
      while (rawLine < rawLineEnd) {
        if (*(rawLine++) == '\n') {
wmmhello's avatar
wmmhello 已提交
2924 2925 2926 2927
          break;
        }
        len++;
      }
dengyihao's avatar
dengyihao 已提交
2928
      if (info->protocol == TSDB_SML_LINE_PROTOCOL && tmp[0] == '#') {  // this line is comment
wmmhello's avatar
wmmhello 已提交
2929 2930
        continue;
      }
wmmhello's avatar
wmmhello 已提交
2931 2932
    }

2933 2934
    uDebug("SML:0x%" PRIx64 " smlParseLine israw:%d, len:%d, sql:%s", info->id, info->isRawLine, len, (info->isRawLine ? "rawdata" : tmp));

X
Xiaoyu Wang 已提交
2935
    if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
wmmhello's avatar
wmmhello 已提交
2936 2937 2938 2939 2940 2941
      if(info->dataFormat){
        SSmlLineInfo element = {0};
        code = smlParseInfluxString(info, tmp, tmp + len, &element);
      }else{
        code = smlParseInfluxString(info, tmp, tmp + len, info->lines + i);
      }
X
Xiaoyu Wang 已提交
2942
    } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
2943 2944 2945 2946 2947 2948 2949
      if(info->dataFormat) {
        SSmlLineInfo element = {0};
        code = smlParseTelnetString(info, (char *)tmp, (char *)tmp + len, &element);
      }else{
        code = smlParseTelnetString(info, (char *)tmp, (char *)tmp + len, info->lines + i);
      }

X
Xiaoyu Wang 已提交
2950
    } else {
2951 2952
      ASSERT(0);
    }
wmmhello's avatar
wmmhello 已提交
2953
    if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2954
      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, tmp);
2955
      return code;
wmmhello's avatar
wmmhello 已提交
2956
    }
2957 2958
    if(info->reRun){
      i = 0;
2959
      rawLine = oldRaw;
2960 2961
      info->reRun = false;
      // clear info->childTables
wmmhello's avatar
wmmhello 已提交
2962 2963 2964 2965 2966 2967 2968
      NodeList* pList = info->childTables;
      while (pList) {
        if(pList->data.used) {
          smlDestroyTableInfo(pList->data.value);
          pList->data.used = false;
        }
        pList = pList->next;
2969 2970 2971
      }

      // clear info->superTables
wmmhello's avatar
wmmhello 已提交
2972 2973 2974 2975 2976 2977 2978 2979 2980 2981 2982 2983
      pList = info->superTables;
      while (pList) {
        if(pList->data.used) {
          smlDestroySTableMeta(pList->data.value);
          pList->data.used = false;
        }
        pList = pList->next;
      }

      if(info->lines != NULL){
        uError("SML:0x%" PRIx64 " info->lines != NULL", info->id);
        return TSDB_CODE_SML_INVALID_DATA;
2984
      }
wmmhello's avatar
wmmhello 已提交
2985 2986
      info->lines = taosMemoryCalloc(info->lineNum, sizeof(SSmlLineInfo));

2987
      memset(&info->preLine, 0, sizeof(SSmlLineInfo));
2988 2989 2990
      info->currSTableMeta = NULL;
      info->currTableDataCtx = NULL;

2991 2992 2993
      SVnodeModifOpStmt* stmt= (SVnodeModifOpStmt*)(info->pQuery->pRoot);
      stmt->freeHashFunc(stmt->pTableBlockHashObj);
      stmt->pTableBlockHashObj = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
2994
      continue;
2995
    }
2996
    i++;
wmmhello's avatar
wmmhello 已提交
2997
  }
2998

2999 3000 3001
  return code;
}

dengyihao's avatar
dengyihao 已提交
3002
static int smlProcess(SSmlHandle *info, char *lines[], char *rawLine, char *rawLineEnd, int numLines) {
3003
  int32_t code = TSDB_CODE_SUCCESS;
3004 3005
  int32_t retryNum = 0;

3006 3007
  info->cost.parseTime = taosGetTimestampUs();

wmmhello's avatar
wmmhello 已提交
3008
  code = smlParseLine(info, lines, rawLine, rawLineEnd, numLines);
3009
  if (code != 0) {
X
Xiaoyu Wang 已提交
3010
    uError("SML:0x%" PRIx64 " smlParseLine error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
3011
    return code;
3012
  }
wmmhello's avatar
wmmhello 已提交
3013

3014 3015 3016 3017 3018 3019
  code = smlParseLineBottom(info);
  if (code != 0) {
    uError("SML:0x%" PRIx64 " smlParseLineBottom error : %s", info->id, tstrerror(code));
    return code;
  }

3020
  info->cost.lineNum = numLines;
wmmhello's avatar
wmmhello 已提交
3021 3022
  info->cost.numOfSTables = nodeListSize(info->superTables);
  info->cost.numOfCTables = nodeListSize(info->childTables);
3023 3024

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

X
Xiaoyu Wang 已提交
3026
  do {
3027 3028
    code = smlModifyDBSchemas(info);
    if (code == 0) break;
wmmhello's avatar
wmmhello 已提交
3029
  } while (retryNum++ < nodeListSize(info->superTables) * MAX_RETRY_TIMES);
3030

wmmhello's avatar
wmmhello 已提交
3031
  if (code != 0) {
X
Xiaoyu Wang 已提交
3032
    uError("SML:0x%" PRIx64 " smlModifyDBSchemas error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
3033
    return code;
wmmhello's avatar
wmmhello 已提交
3034
  }
wmmhello's avatar
wmmhello 已提交
3035

3036
  info->cost.insertBindTime = taosGetTimestampUs();
wmmhello's avatar
wmmhello 已提交
3037 3038
  code = smlInsertData(info);
  if (code != 0) {
X
Xiaoyu Wang 已提交
3039
    uError("SML:0x%" PRIx64 " smlInsertData error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
3040
    return code;
wmmhello's avatar
wmmhello 已提交
3041 3042 3043 3044 3045
  }

  return code;
}

wmmhello's avatar
wmmhello 已提交
3046 3047 3048 3049 3050
TAOS_RES *taos_schemaless_insert_inner(TAOS *taos, char *lines[], char *rawLine, char *rawLineEnd,
                                       int numLines, int protocol, int precision, int32_t ttl, int64_t reqid) {
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
3051
  }
3052

wmmhello's avatar
wmmhello 已提交
3053 3054 3055 3056
  SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
  if (request == NULL) {
    uError("SML:taos_schemaless_insert error request is null");
    return NULL;
wmmhello's avatar
wmmhello 已提交
3057
  }
3058

wmmhello's avatar
wmmhello 已提交
3059 3060 3061 3062
  SSmlHandle *info = smlBuildSmlInfo(taos);
  if (info == NULL) {
    request->code = TSDB_CODE_OUT_OF_MEMORY;
    uError("SML:taos_schemaless_insert error SSmlHandle is null");
wmmhello's avatar
wmmhello 已提交
3063 3064
    goto end;
  }
wmmhello's avatar
wmmhello 已提交
3065 3066 3067 3068 3069 3070 3071 3072
  info->pRequest = request;
  info->isRawLine = rawLine != NULL;
  info->ttl       = ttl;
  info->precision = precision;
  info->protocol = protocol;
  info->msgBuf.buf = info->pRequest->msgBuf;
  info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
  info->lineNum = numLines;
wmmhello's avatar
wmmhello 已提交
3073

wmmhello's avatar
wmmhello 已提交
3074
  SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
wmmhello's avatar
wmmhello 已提交
3075 3076 3077
  if (request->pDb == NULL) {
    request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
    smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
wmmhello's avatar
wmmhello 已提交
3078 3079 3080
    goto end;
  }

X
Xiaoyu Wang 已提交
3081
  if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
3082
    request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
wmmhello's avatar
wmmhello 已提交
3083
    smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
3084
    goto end;
wmmhello's avatar
wmmhello 已提交
3085 3086
  }

X
Xiaoyu Wang 已提交
3087 3088
  if (protocol == TSDB_SML_LINE_PROTOCOL &&
      (precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)) {
3089
    request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
wmmhello's avatar
wmmhello 已提交
3090
    smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
3091 3092 3093
    goto end;
  }

3094
  if (protocol == TSDB_SML_JSON_PROTOCOL) {
wmmhello's avatar
wmmhello 已提交
3095
    numLines = 1;
3096
  } else if (numLines <= 0) {
wmmhello's avatar
wmmhello 已提交
3097 3098 3099 3100 3101
    request->code = TSDB_CODE_SML_INVALID_DATA;
    smlBuildInvalidDataMsg(&msg, "line num is invalid", NULL);
    goto end;
  }

wmmhello's avatar
wmmhello 已提交
3102 3103 3104 3105 3106
  int32_t code = smlProcess(info, lines, rawLine, rawLineEnd, numLines);
  request->code = code;
  info->cost.endTime = taosGetTimestampUs();
  info->cost.code = code;
  smlPrintStatisticInfo(info);
3107

wmmhello's avatar
wmmhello 已提交
3108
end:
3109
  uDebug("resultend:%s", request->msgBuf);
wmmhello's avatar
wmmhello 已提交
3110
  smlDestroyInfo(info);
3111
  return (TAOS_RES *)request;
wmmhello's avatar
wmmhello 已提交
3112
}
wmmhello's avatar
wmmhello 已提交
3113 3114 3115 3116 3117 3118 3119 3120 3121 3122 3123 3124 3125 3126 3127 3128 3129

/**
 * 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 已提交
3130
 * @return TAOS_RES
wmmhello's avatar
wmmhello 已提交
3131 3132
 */

3133 3134
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 已提交
3135
  return taos_schemaless_insert_inner(taos, lines, NULL, NULL, numLines, protocol, precision, ttl, reqid);
dengyihao's avatar
dengyihao 已提交
3136 3137
}

3138 3139 3140
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 已提交
3141

3142 3143 3144
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 已提交
3145

3146 3147
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 已提交
3148 3149
}

3150
TAOS_RES *taos_schemaless_insert_raw_ttl_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
3151
                                                    int precision, int32_t ttl, int64_t reqid) {
dengyihao's avatar
dengyihao 已提交
3152 3153 3154 3155 3156 3157 3158 3159 3160 3161 3162 3163
  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;
    }
  }
wmmhello's avatar
wmmhello 已提交
3164
  return taos_schemaless_insert_inner(taos, NULL, lines, lines + len, *totalRows, protocol, precision, ttl, reqid);
dengyihao's avatar
dengyihao 已提交
3165 3166
}

3167 3168 3169 3170 3171 3172
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 已提交
3173

3174 3175
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 已提交
3176
}