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

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

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

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

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

X
Xiaoyu Wang 已提交
53 54
#define IS_SLASH_LETTER(sql) \
  (IS_SLASH_COMMA(sql) || IS_SLASH_SPACE(sql) || IS_SLASH_EQUAL(sql) || IS_SLASH_QUOTE(sql) || IS_SLASH_SLASH(sql))
wmmhello's avatar
wmmhello 已提交
55

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

X
Xiaoyu Wang 已提交
58 59 60 61 62 63 64 65
#define PROCESS_SLASH(key, keyLen)           \
  for (int i = 1; i < keyLen; ++i) {         \
    if (IS_SLASH_LETTER(key + i)) {          \
      MOVE_FORWARD_ONE(key + i, keyLen - i); \
      i--;                                   \
      keyLen--;                              \
    }                                        \
  }
wmmhello's avatar
wmmhello 已提交
66

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

70 71 72
#define OTD_JSON_SUB_FIELDS_NUM 2
#define OTD_JSON_FIELDS_NUM     4

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

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

#define MAX_RETRY_TIMES 5
X
Xiaoyu Wang 已提交
82
#define LINE_BATCH      20000
wmmhello's avatar
wmmhello 已提交
83 84 85 86
//=================================================================================================
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;

typedef enum {
87 88 89
  SCHEMA_ACTION_NULL,
  SCHEMA_ACTION_COLUMN,
  SCHEMA_ACTION_TAG
wmmhello's avatar
wmmhello 已提交
90 91 92
} ESchemaAction;

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

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

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

X
Xiaoyu Wang 已提交
111
  SArray *tags;
wmmhello's avatar
wmmhello 已提交
112

113 114
  // if info->formatData is true, elements are SArray<SSmlKv*>.
  // if info->formatData is false, elements are SHashObj<cols key string, SSmlKv*> for find by key quickly
X
Xiaoyu Wang 已提交
115
  SArray *cols;
wmmhello's avatar
wmmhello 已提交
116 117 118
} SSmlTableInfo;

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

X
Xiaoyu Wang 已提交
122 123
  SArray   *cols;
  SHashObj *colHash;
124

wmmhello's avatar
wmmhello 已提交
125 126 127 128
  STableMeta *tableMeta;
} SSmlSTableMeta;

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

133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
typedef struct {
  int32_t code;
  int32_t lineNum;

  int32_t numOfSTables;
  int32_t numOfCTables;
  int32_t numOfCreateSTables;

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

X
Xiaoyu Wang 已提交
148 149 150
typedef struct {
  SRequestObj     *request;
  tsem_t           sem;
wmmhello's avatar
wmmhello 已提交
151 152 153
  TdThreadSpinlock lock;
} Params;

wmmhello's avatar
wmmhello 已提交
154
typedef struct {
X
Xiaoyu Wang 已提交
155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177
  int64_t id;
  Params *params;
  bool    isLast;

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

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

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

  SSmlCostInfo cost;
  int32_t      affectedRows;
  SSmlMsgBuf   msgBuf;
  SHashObj    *dumplicateKey;  // for dumplicate key
  SArray      *colsContainer;  // for cols parse, if dataFormat == false
wmmhello's avatar
wmmhello 已提交
178
} SSmlHandle;
wmmhello's avatar
wmmhello 已提交
179 180
//=================================================================================================

wmmhello's avatar
wmmhello 已提交
181
//=================================================================================================
182
static volatile int64_t linesSmlHandleId = 0;
X
Xiaoyu Wang 已提交
183 184
static int64_t          smlGenId() {
           int64_t id;
wmmhello's avatar
wmmhello 已提交
185

X
Xiaoyu Wang 已提交
186 187
           do {
             id = atomic_add_fetch_64(&linesSmlHandleId, 1);
wmmhello's avatar
wmmhello 已提交
188 189
  } while (id == 0);

X
Xiaoyu Wang 已提交
190
           return id;
wmmhello's avatar
wmmhello 已提交
191 192
}

193
static inline bool smlDoubleToInt64OverFlow(double num) {
X
Xiaoyu Wang 已提交
194
  if (num >= (double)INT64_MAX || num <= (double)INT64_MIN) return true;
195 196 197 198 199 200 201 202 203 204 205 206
  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 已提交
207
static int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2) {
208 209 210 211 212 213 214 215
  if(pBuf->buf){
    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 已提交
216
  }
wmmhello's avatar
wmmhello 已提交
217 218 219
  return TSDB_CODE_SML_INVALID_DATA;
}

X
Xiaoyu Wang 已提交
220
static int32_t smlGenerateSchemaAction(SSchema *colField, SHashObj *colHash, SSmlKv *kv, bool isTag,
221
                                       ESchemaAction *action, SSmlHandle *info) {
222
  uint16_t *index = (uint16_t *)taosHashGet(colHash, kv->key, kv->keyLen);
223 224
  if (index) {
    if (colField[*index].type != kv->type) {
X
Xiaoyu Wang 已提交
225 226
      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);
227 228 229
      return TSDB_CODE_TSC_INVALID_VALUE;
    }

X
Xiaoyu Wang 已提交
230 231 232 233
    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))) {
234
      if (isTag) {
235
        *action = SCHEMA_ACTION_TAG;
236
      } else {
237
        *action = SCHEMA_ACTION_COLUMN;
238 239 240 241
      }
    }
  } else {
    if (isTag) {
242
      *action = SCHEMA_ACTION_TAG;
243
    } else {
244
      *action = SCHEMA_ACTION_COLUMN;
245 246
    }
  }
wmmhello's avatar
wmmhello 已提交
247 248 249
  return 0;
}

wmmhello's avatar
wmmhello 已提交
250
static int32_t smlFindNearestPowerOf2(int32_t length, uint8_t type) {
wmmhello's avatar
wmmhello 已提交
251
  int32_t result = 1;
X
Xiaoyu Wang 已提交
252
  while (result <= length) {
wmmhello's avatar
wmmhello 已提交
253 254
    result *= 2;
  }
wmmhello's avatar
wmmhello 已提交
255 256 257 258 259
  if (type == TSDB_DATA_TYPE_BINARY && result > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE){
    result = TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE;
  } else if (type == TSDB_DATA_TYPE_NCHAR && result > (TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
    result = (TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
  }
wmmhello's avatar
wmmhello 已提交
260

261 262 263 264
  if (type == TSDB_DATA_TYPE_NCHAR){
    result = result * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
  }else if (type == TSDB_DATA_TYPE_BINARY){
    result = result + VARSTR_HEADER_SIZE;
wmmhello's avatar
wmmhello 已提交
265
  }
266
  return result;
wmmhello's avatar
wmmhello 已提交
267 268
}

X
Xiaoyu Wang 已提交
269
static int32_t smlProcessSchemaAction(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
270
                                      ESchemaAction *action, bool isTag) {
271 272
  int32_t code = TSDB_CODE_SUCCESS;
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
273
    if(j == 0 && !isTag) continue;
X
Xiaoyu Wang 已提交
274
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, j);
275
    code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, action, info);
X
Xiaoyu Wang 已提交
276
    if (code != TSDB_CODE_SUCCESS) {
277 278 279 280 281 282
      return code;
    }
  }
  return TSDB_CODE_SUCCESS;
}

283
static int32_t smlCheckMeta(SSchema *schema, int32_t length, SArray *cols, bool isTag) {
284
  SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
285 286
  int32_t i = 0;
  for ( ;i < length; i++) {
287 288 289
    taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
  }

290 291 292 293 294 295
  if (isTag){
    i = 0;
  } else {
    i = 1;
  }
  for (; i < taosArrayGetSize(cols); i++) {
X
Xiaoyu Wang 已提交
296 297
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
    if (taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL) {
298 299 300
      return -1;
    }
  }
301
  taosHashCleanup(hashTmp);
302 303 304
  return 0;
}

305 306 307 308 309 310 311 312
static int32_t getBytes(uint8_t type, int32_t length){
  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
    return smlFindNearestPowerOf2(length, type);
  } else {
    return tDataTypes[type].bytes;
  }
}

wmmhello's avatar
wmmhello 已提交
313 314 315 316 317
//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, SSmlSTableMeta *sTableData,
                               STableMeta *pTableMeta, ESchemaAction action){

318 319 320 321 322 323 324 325 326 327 328 329 330 331 332
  SRequestObj*   pRequest = NULL;
  SMCreateStbReq pReq = {0};
  int32_t        code = TSDB_CODE_SUCCESS;
  SCmdMsgInfo    pCmdMsg = {0};

  code = buildRequest(info->taos->id, "", 0, NULL, false, &pRequest);
  if (code != TSDB_CODE_SUCCESS) {
    goto end;
  }

  if (!pRequest->pDb) {
    code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
    goto end;
  }

wmmhello's avatar
wmmhello 已提交
333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349
  if (action == SCHEMA_ACTION_NULL){
    pReq.colVer = 1;
    pReq.tagVer = 1;
    pReq.suid = 0;
    pReq.source = TD_REQ_FROM_APP;
  } else if (action == SCHEMA_ACTION_TAG){
    pReq.colVer = pTableMeta->sversion;
    pReq.tagVer = pTableMeta->tversion + 1;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
  } else if (action == SCHEMA_ACTION_COLUMN){
    pReq.colVer = pTableMeta->sversion + 1;
    pReq.tagVer = pTableMeta->tversion;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
  }

350 351 352 353
  pReq.commentLen = -1;
  pReq.igExists = true;
  tNameExtractFullName(pName, pReq.name);

wmmhello's avatar
wmmhello 已提交
354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375
  if(action == SCHEMA_ACTION_NULL || action == SCHEMA_ACTION_COLUMN){
    pReq.numOfColumns = taosArrayGetSize(sTableData->cols);
    pReq.pColumns = taosArrayInit(pReq.numOfColumns, sizeof(SField));
    for (int i = 0; i < pReq.numOfColumns; i++) {
      SSmlKv *kv = (SSmlKv *)taosArrayGetP(sTableData->cols, i);
      SField field = {0};
      field.type = kv->type;
      field.bytes = getBytes(kv->type, kv->length);
      memcpy(field.name, kv->key, kv->keyLen);
      taosArrayPush(pReq.pColumns, &field);
    }
  }else if (action == SCHEMA_ACTION_TAG){
    pReq.numOfColumns = pTableMeta->tableInfo.numOfColumns;
    pReq.pColumns = taosArrayInit(pReq.numOfColumns, sizeof(SField));
    for (int i = 0; i < pReq.numOfColumns; i++) {
      SSchema *s = &pTableMeta->schema[i];
      SField field = {0};
      field.type = s->type;
      field.bytes = s->bytes;
      strcpy(field.name, s->name);
      taosArrayPush(pReq.pColumns, &field);
    }
376 377
  }

wmmhello's avatar
wmmhello 已提交
378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400
  if(action == SCHEMA_ACTION_NULL || action == SCHEMA_ACTION_TAG){
    pReq.numOfTags = taosArrayGetSize(sTableData->tags);
    if (pReq.numOfTags == 0){
      pReq.numOfTags = 1;
      pReq.pTags = taosArrayInit(pReq.numOfTags, sizeof(SField));
      SField field = {0};
      field.type = TSDB_DATA_TYPE_NCHAR;
      field.bytes = 1;
      strcpy(field.name, tsSmlTagName);
      taosArrayPush(pReq.pTags, &field);
    }else{
      pReq.pTags = taosArrayInit(pReq.numOfTags, sizeof(SField));
      for (int i = 0; i < pReq.numOfTags; i++) {
        SSmlKv *kv = (SSmlKv *)taosArrayGetP(sTableData->tags, i);
        SField field = {0};
        field.type = kv->type;
        field.bytes = getBytes(kv->type, kv->length);
        memcpy(field.name, kv->key, kv->keyLen);
        taosArrayPush(pReq.pTags, &field);
      }
    }
  }else if (action == SCHEMA_ACTION_COLUMN){
    pReq.numOfTags = pTableMeta->tableInfo.numOfTags;
401 402
    pReq.pTags = taosArrayInit(pReq.numOfTags, sizeof(SField));
    for (int i = 0; i < pReq.numOfTags; i++) {
wmmhello's avatar
wmmhello 已提交
403
      SSchema *s = &pTableMeta->schema[i + pTableMeta->tableInfo.numOfColumns];
404
      SField field = {0};
wmmhello's avatar
wmmhello 已提交
405 406 407
      field.type = s->type;
      field.bytes = s->bytes;
      strcpy(field.name, s->name);
408 409 410 411 412 413 414 415 416 417 418 419 420 421 422
      taosArrayPush(pReq.pTags, &field);
    }
  }

  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) {
    tFreeSMCreateStbReq(&pReq);
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto end;
  }
  tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq);

D
dapan1121 已提交
423
  SQuery pQuery = {0};
424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442
  pQuery.execMode = QUERY_EXEC_MODE_RPC;
  pQuery.pCmdMsg = &pCmdMsg;
  pQuery.msgType = pQuery.pCmdMsg->msgType;
  pQuery.stableQuery = true;

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

  if(pRequest->code == TSDB_CODE_SUCCESS){
    catalogRemoveTableMeta(info->pCatalog, pName);
  }
  code = pRequest->code;
  taosMemoryFree(pCmdMsg.pMsg);

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

X
Xiaoyu Wang 已提交
443
static int32_t smlModifyDBSchemas(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
444
  int32_t code = 0;
X
Xiaoyu Wang 已提交
445
  SName   pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
446
  strcpy(pName.dbname, info->pRequest->pDb);
wmmhello's avatar
wmmhello 已提交
447

D
dapan1121 已提交
448 449 450 451 452
  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 已提交
453 454

  SSmlSTableMeta **tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, NULL);
wmmhello's avatar
wmmhello 已提交
455
  while (tableMetaSml) {
X
Xiaoyu Wang 已提交
456 457 458
    SSmlSTableMeta *sTableData = *tableMetaSml;
    STableMeta     *pTableMeta = NULL;
    bool            needCheckMeta = false;  // for multi thread
wmmhello's avatar
wmmhello 已提交
459

wmmhello's avatar
wmmhello 已提交
460
    size_t superTableLen = 0;
X
Xiaoyu Wang 已提交
461
    void  *superTable = taosHashGetKey(tableMetaSml, &superTableLen);
462
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
wmmhello's avatar
wmmhello 已提交
463
    memcpy(pName.tname, superTable, superTableLen);
wmmhello's avatar
wmmhello 已提交
464

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

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

482 483
      ESchemaAction action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &action, true);
484 485
      if (code != TSDB_CODE_SUCCESS) {
        taosHashCleanup(hashTmp);
486
        goto end;
487
      }
488
      if (action == SCHEMA_ACTION_TAG){
wmmhello's avatar
wmmhello 已提交
489
        code = smlSendMetaMsg(info, &pName, sTableData, pTableMeta, action);
490 491 492 493 494 495 496 497 498 499
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, superTable);
          goto end;
        }
      }

      code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
      if (code != TSDB_CODE_SUCCESS) {
        goto end;
      }
500 501

      taosHashClear(hashTmp);
502
      for (uint16_t i = 1; i < pTableMeta->tableInfo.numOfColumns; i++) {
503 504
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
505 506
      action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->cols, &action, false);
507 508
      taosHashCleanup(hashTmp);
      if (code != TSDB_CODE_SUCCESS) {
509
        goto end;
wmmhello's avatar
wmmhello 已提交
510
      }
511
      if (action == SCHEMA_ACTION_COLUMN){
wmmhello's avatar
wmmhello 已提交
512
        code = smlSendMetaMsg(info, &pName, sTableData, pTableMeta, action);
513 514 515 516 517
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, superTable);
          goto end;
        }
      }
wmmhello's avatar
wmmhello 已提交
518

D
dapan1121 已提交
519
      code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
wmmhello's avatar
wmmhello 已提交
520
      if (code != TSDB_CODE_SUCCESS) {
521
        goto end;
wmmhello's avatar
wmmhello 已提交
522
      }
523
      needCheckMeta = true;
wmmhello's avatar
wmmhello 已提交
524
    } else {
X
Xiaoyu Wang 已提交
525
      uError("SML:0x%" PRIx64 " load table meta error: %s", info->id, tstrerror(code));
526
      goto end;
wmmhello's avatar
wmmhello 已提交
527
    }
X
Xiaoyu Wang 已提交
528
    if (pTableMeta) taosMemoryFree(pTableMeta);
529

D
dapan1121 已提交
530
    code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
531
    if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
532
      uError("SML:0x%" PRIx64 " catalogGetSTableMeta failed. super table name %s", info->id, (char *)superTable);
533
      goto end;
534
    }
535

X
Xiaoyu Wang 已提交
536 537
    if (needCheckMeta) {
      code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags,
538
                          sTableData->tags, true);
539
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
540
        uError("SML:0x%" PRIx64 " check tag failed. super table name %s", info->id, (char *)superTable);
541 542
        goto end;
      }
543
      code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols, false);
544
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
545
        uError("SML:0x%" PRIx64 " check cols failed. super table name %s", info->id, (char *)superTable);
546 547 548 549
        goto end;
      }
    }

550
    sTableData->tableMeta = pTableMeta;
wmmhello's avatar
wmmhello 已提交
551

X
Xiaoyu Wang 已提交
552
    tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, tableMetaSml);
wmmhello's avatar
wmmhello 已提交
553 554
  }
  return 0;
555 556

end:
D
dapan1121 已提交
557
  catalogRefreshTableMeta(info->pCatalog, &conn, &pName, 1);
558
  return code;
wmmhello's avatar
wmmhello 已提交
559 560
}

X
Xiaoyu Wang 已提交
561
static bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
562
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
563 564 565 566
  int32_t     len = kvVal->length;
  char       *endptr = NULL;
  double      result = taosStr2Double(pVal, &endptr);
  if (pVal == endptr) {
567
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
568 569 570
    return false;
  }

571
  int32_t left = len - (endptr - pVal);
X
Xiaoyu Wang 已提交
572
  if (left == 0 || (left == 3 && strncasecmp(endptr, "f64", left) == 0)) {
573 574
    kvVal->type = TSDB_DATA_TYPE_DOUBLE;
    kvVal->d = result;
X
Xiaoyu Wang 已提交
575 576
  } else if ((left == 3 && strncasecmp(endptr, "f32", left) == 0)) {
    if (!IS_VALID_FLOAT(result)) {
577 578
      smlBuildInvalidDataMsg(msg, "float out of range[-3.402823466e+38,3.402823466e+38]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
579
    }
580 581
    kvVal->type = TSDB_DATA_TYPE_FLOAT;
    kvVal->f = (float)result;
X
Xiaoyu Wang 已提交
582 583
  } else if ((left == 1 && *endptr == 'i') || (left == 3 && strncasecmp(endptr, "i64", left) == 0)) {
    if (smlDoubleToInt64OverFlow(result)) {
wmmhello's avatar
wmmhello 已提交
584 585
      errno = 0;
      int64_t tmp = taosStr2Int64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
586
      if (errno == ERANGE) {
wmmhello's avatar
wmmhello 已提交
587 588 589 590 591 592
        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 已提交
593
    }
594
    kvVal->type = TSDB_DATA_TYPE_BIGINT;
wmmhello's avatar
wmmhello 已提交
595
    kvVal->i = (int64_t)result;
wmmhello's avatar
wmmhello 已提交
596
  } else if ((left == 1 && *endptr == 'u') || (left == 3 && strncasecmp(endptr, "u64", left) == 0)) {
X
Xiaoyu Wang 已提交
597
    if (result >= (double)UINT64_MAX || result < 0) {
wmmhello's avatar
wmmhello 已提交
598 599
      errno = 0;
      uint64_t tmp = taosStr2UInt64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
600
      if (errno == ERANGE || result < 0) {
wmmhello's avatar
wmmhello 已提交
601 602 603 604 605 606
        smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", pVal);
        return false;
      }
      kvVal->type = TSDB_DATA_TYPE_UBIGINT;
      kvVal->u = tmp;
      return true;
607
    }
608
    kvVal->type = TSDB_DATA_TYPE_UBIGINT;
wmmhello's avatar
wmmhello 已提交
609
    kvVal->u = result;
X
Xiaoyu Wang 已提交
610 611
  } else if (left == 3 && strncasecmp(endptr, "i32", left) == 0) {
    if (!IS_VALID_INT(result)) {
612 613
      smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
614
    }
615 616
    kvVal->type = TSDB_DATA_TYPE_INT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
617 618
  } else if (left == 3 && strncasecmp(endptr, "u32", left) == 0) {
    if (!IS_VALID_UINT(result)) {
619 620
      smlBuildInvalidDataMsg(msg, "unsigned int out of range[0,4294967295]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
621
    }
622 623
    kvVal->type = TSDB_DATA_TYPE_UINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
624 625
  } else if (left == 3 && strncasecmp(endptr, "i16", left) == 0) {
    if (!IS_VALID_SMALLINT(result)) {
626 627
      smlBuildInvalidDataMsg(msg, "small int our of range[-32768,32767]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
628
    }
629 630
    kvVal->type = TSDB_DATA_TYPE_SMALLINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
631 632
  } else if (left == 3 && strncasecmp(endptr, "u16", left) == 0) {
    if (!IS_VALID_USMALLINT(result)) {
633 634
      smlBuildInvalidDataMsg(msg, "unsigned small int out of rang[0,65535]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
635
    }
636 637
    kvVal->type = TSDB_DATA_TYPE_USMALLINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
638 639
  } else if (left == 2 && strncasecmp(endptr, "i8", left) == 0) {
    if (!IS_VALID_TINYINT(result)) {
640 641
      smlBuildInvalidDataMsg(msg, "tiny int out of range[-128,127]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
642
    }
643 644
    kvVal->type = TSDB_DATA_TYPE_TINYINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
645 646
  } else if (left == 2 && strncasecmp(endptr, "u8", left) == 0) {
    if (!IS_VALID_UTINYINT(result)) {
647 648
      smlBuildInvalidDataMsg(msg, "unsigned tiny int out of range[0,255]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
649
    }
650 651
    kvVal->type = TSDB_DATA_TYPE_UTINYINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
652
  } else {
653
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
654 655
    return false;
  }
656
  return true;
wmmhello's avatar
wmmhello 已提交
657 658
}

wmmhello's avatar
wmmhello 已提交
659
static bool smlParseBool(SSmlKv *kvVal) {
wmmhello's avatar
wmmhello 已提交
660
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
661
  int32_t     len = kvVal->length;
662
  if ((len == 1) && (pVal[0] == 't' || pVal[0] == 'T')) {
wmmhello's avatar
wmmhello 已提交
663
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
664 665 666
    return true;
  }

667
  if ((len == 1) && (pVal[0] == 'f' || pVal[0] == 'F')) {
wmmhello's avatar
wmmhello 已提交
668
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
669 670 671
    return true;
  }

X
Xiaoyu Wang 已提交
672
  if ((len == 4) && !strncasecmp(pVal, "true", len)) {
wmmhello's avatar
wmmhello 已提交
673
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
674 675
    return true;
  }
X
Xiaoyu Wang 已提交
676
  if ((len == 5) && !strncasecmp(pVal, "false", len)) {
wmmhello's avatar
wmmhello 已提交
677
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
678 679 680 681 682
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
683
static bool smlIsBinary(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
684
  // binary: "abc"
wmmhello's avatar
wmmhello 已提交
685 686 687 688 689 690 691 692 693
  if (len < 2) {
    return false;
  }
  if (pVal[0] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
694
static bool smlIsNchar(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
695
  // nchar: L"abc"
wmmhello's avatar
wmmhello 已提交
696 697 698
  if (len < 3) {
    return false;
  }
X
Xiaoyu Wang 已提交
699
  if ((pVal[0] == 'l' || pVal[0] == 'L') && pVal[1] == '"' && pVal[len - 1] == '"') {
wmmhello's avatar
wmmhello 已提交
700 701 702 703 704
    return true;
  }
  return false;
}

705
static int64_t smlGetTimeValue(const char *value, int32_t len, int8_t type) {
X
Xiaoyu Wang 已提交
706
  char   *endPtr = NULL;
wafwerar's avatar
wafwerar 已提交
707
  int64_t tsInt64 = taosStr2Int64(value, &endPtr, 10);
X
Xiaoyu Wang 已提交
708
  if (value + len != endPtr) {
709
    return -1;
wmmhello's avatar
wmmhello 已提交
710
  }
711
  double ts = tsInt64;
712 713
  switch (type) {
    case TSDB_TIME_PRECISION_HOURS:
wmmhello's avatar
wmmhello 已提交
714 715
      ts *= NANOSECOND_PER_HOUR;
      tsInt64 *= NANOSECOND_PER_HOUR;
716 717
      break;
    case TSDB_TIME_PRECISION_MINUTES:
wmmhello's avatar
wmmhello 已提交
718 719
      ts *= NANOSECOND_PER_MINUTE;
      tsInt64 *= NANOSECOND_PER_MINUTE;
720 721
      break;
    case TSDB_TIME_PRECISION_SECONDS:
wmmhello's avatar
wmmhello 已提交
722 723
      ts *= NANOSECOND_PER_SEC;
      tsInt64 *= NANOSECOND_PER_SEC;
724 725
      break;
    case TSDB_TIME_PRECISION_MILLI:
wmmhello's avatar
wmmhello 已提交
726 727
      ts *= NANOSECOND_PER_MSEC;
      tsInt64 *= NANOSECOND_PER_MSEC;
728 729
      break;
    case TSDB_TIME_PRECISION_MICRO:
wmmhello's avatar
wmmhello 已提交
730 731
      ts *= NANOSECOND_PER_USEC;
      tsInt64 *= NANOSECOND_PER_USEC;
732 733 734 735 736
      break;
    case TSDB_TIME_PRECISION_NANO:
      break;
    default:
      ASSERT(0);
wmmhello's avatar
wmmhello 已提交
737
  }
X
Xiaoyu Wang 已提交
738
  if (ts >= (double)INT64_MAX || ts < 0) {
739
    return -1;
wmmhello's avatar
wmmhello 已提交
740 741
  }

742
  return tsInt64;
743 744 745 746 747 748
}

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) {
749
    return TSDB_TIME_PRECISION_MILLI;
750 751
  } else {
    return -1;
wmmhello's avatar
wmmhello 已提交
752
  }
753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771
}

static int8_t smlGetTsTypeByPrecision(int8_t precision) {
  switch (precision) {
    case TSDB_SML_TIMESTAMP_HOURS:
      return TSDB_TIME_PRECISION_HOURS;
    case TSDB_SML_TIMESTAMP_MILLI_SECONDS:
      return TSDB_TIME_PRECISION_MILLI;
    case TSDB_SML_TIMESTAMP_NANO_SECONDS:
    case TSDB_SML_TIMESTAMP_NOT_CONFIGURED:
      return TSDB_TIME_PRECISION_NANO;
    case TSDB_SML_TIMESTAMP_MICRO_SECONDS:
      return TSDB_TIME_PRECISION_MICRO;
    case TSDB_SML_TIMESTAMP_SECONDS:
      return TSDB_TIME_PRECISION_SECONDS;
    case TSDB_SML_TIMESTAMP_MINUTES:
      return TSDB_TIME_PRECISION_MINUTES;
    default:
      return -1;
wmmhello's avatar
wmmhello 已提交
772
  }
773 774
}

X
Xiaoyu Wang 已提交
775 776
static int64_t smlParseInfluxTime(SSmlHandle *info, const char *data, int32_t len) {
  if (len == 0 || (len == 1 && data[0] == '0')) {
777
    return taosGetTimestampNs();
wmmhello's avatar
wmmhello 已提交
778 779
  }

780 781 782 783
  int8_t tsType = smlGetTsTypeByPrecision(info->precision);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp precision", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
784
  }
785 786

  int64_t ts = smlGetTimeValue(data, len, tsType);
X
Xiaoyu Wang 已提交
787
  if (ts == -1) {
788 789
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
wmmhello's avatar
wmmhello 已提交
790
  }
791 792 793
  return ts;
}

X
Xiaoyu Wang 已提交
794 795
static int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len) {
  if (!data) {
796 797
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
798
  }
X
Xiaoyu Wang 已提交
799
  if (len == 1 && data[0] == '0') {
800 801
    return taosGetTimestampNs();
  }
802 803
  int8_t tsType = smlGetTsTypeByLen(len);
  if (tsType == -1) {
X
Xiaoyu Wang 已提交
804 805
    smlBuildInvalidDataMsg(&info->msgBuf,
                           "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
806 807 808
    return -1;
  }
  int64_t ts = smlGetTimeValue(data, len, tsType);
X
Xiaoyu Wang 已提交
809
  if (ts == -1) {
810 811 812 813 814 815
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

X
Xiaoyu Wang 已提交
816
static int32_t smlParseTS(SSmlHandle *info, const char *data, int32_t len, SArray *cols) {
817
  int64_t ts = 0;
X
Xiaoyu Wang 已提交
818
  if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
819
//    uError("SML:data:%s,len:%d", data, len);
820
    ts = smlParseInfluxTime(info, data, len);
X
Xiaoyu Wang 已提交
821
  } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
822
    ts = smlParseOpenTsdbTime(info, data, len);
X
Xiaoyu Wang 已提交
823
  } else {
824
    ASSERT(0);
825
  }
826

wmmhello's avatar
wmmhello 已提交
827
  if (ts == -1) return TSDB_CODE_INVALID_TIMESTAMP;
828 829 830

  // add ts to
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
831
  if (!kv) {
832 833 834 835 836 837 838 839
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  kv->key = TS;
  kv->keyLen = TS_LEN;
  kv->i = ts;
  kv->type = TSDB_DATA_TYPE_TIMESTAMP;
  kv->length = (int16_t)tDataTypes[kv->type].bytes;
X
Xiaoyu Wang 已提交
840
  if (cols) taosArrayPush(cols, &kv);
841 842 843
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
844
static int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
X
Xiaoyu Wang 已提交
845
  // binary
wmmhello's avatar
wmmhello 已提交
846
  if (smlIsBinary(pVal->value, pVal->length)) {
847
    pVal->type = TSDB_DATA_TYPE_BINARY;
wmmhello's avatar
wmmhello 已提交
848
    pVal->length -= BINARY_ADD_LEN;
wmmhello's avatar
wmmhello 已提交
849 850 851
    if (pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
852
    pVal->value += (BINARY_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
853
    return TSDB_CODE_SUCCESS;
854
  }
X
Xiaoyu Wang 已提交
855
  // nchar
wmmhello's avatar
wmmhello 已提交
856
  if (smlIsNchar(pVal->value, pVal->length)) {
857
    pVal->type = TSDB_DATA_TYPE_NCHAR;
wmmhello's avatar
wmmhello 已提交
858
    pVal->length -= NCHAR_ADD_LEN;
wmmhello's avatar
wmmhello 已提交
859 860 861
    if(pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
862
    pVal->value += (NCHAR_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
863
    return TSDB_CODE_SUCCESS;
864 865
  }

X
Xiaoyu Wang 已提交
866
  // bool
867 868 869
  if (smlParseBool(pVal)) {
    pVal->type = TSDB_DATA_TYPE_BOOL;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
870
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
871
  }
X
Xiaoyu Wang 已提交
872
  // number
873
  if (smlParseNumber(pVal, msg)) {
wmmhello's avatar
wmmhello 已提交
874
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
875
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
876 877
  }

wmmhello's avatar
wmmhello 已提交
878
  return TSDB_CODE_TSC_INVALID_VALUE;
wmmhello's avatar
wmmhello 已提交
879 880
}

X
Xiaoyu Wang 已提交
881 882
static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
883
  JUMP_SPACE(sql)
X
Xiaoyu Wang 已提交
884
  if (*sql == COMMA) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
885
  elements->measure = sql;
wmmhello's avatar
wmmhello 已提交
886

wmmhello's avatar
wmmhello 已提交
887
  // parse measure
wmmhello's avatar
wmmhello 已提交
888
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
889 890
    if ((sql != elements->measure) && IS_SLASH_LETTER(sql)) {
      MOVE_FORWARD_ONE(sql, strlen(sql) + 1);
wmmhello's avatar
wmmhello 已提交
891 892
      continue;
    }
X
Xiaoyu Wang 已提交
893
    if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
894 895 896
      break;
    }

X
Xiaoyu Wang 已提交
897
    if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
898 899
      break;
    }
wmmhello's avatar
wmmhello 已提交
900 901
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
902
  elements->measureLen = sql - elements->measure;
X
Xiaoyu Wang 已提交
903
  if (IS_INVALID_TABLE_LEN(elements->measureLen)) {
904
    smlBuildInvalidDataMsg(msg, "measure is empty or too large than 192", NULL);
wmmhello's avatar
wmmhello 已提交
905
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
wmmhello's avatar
wmmhello 已提交
906
  }
wmmhello's avatar
wmmhello 已提交
907

wmmhello's avatar
wmmhello 已提交
908
  // parse tag
X
Xiaoyu Wang 已提交
909
  if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
910
    elements->tagsLen = 0;
X
Xiaoyu Wang 已提交
911 912
  } else {
    if (*sql == COMMA) sql++;
wmmhello's avatar
wmmhello 已提交
913 914
    elements->tags = sql;
    while (*sql != '\0') {
X
Xiaoyu Wang 已提交
915
      if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
916 917 918
        break;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
919
    }
wmmhello's avatar
wmmhello 已提交
920
    elements->tagsLen = sql - elements->tags;
921
  }
wmmhello's avatar
wmmhello 已提交
922
  elements->measureTagsLen = sql - elements->measure;
wmmhello's avatar
wmmhello 已提交
923

wmmhello's avatar
wmmhello 已提交
924 925 926
  // parse cols
  JUMP_SPACE(sql)
  elements->cols = sql;
wmmhello's avatar
wmmhello 已提交
927
  bool isInQuote = false;
wmmhello's avatar
wmmhello 已提交
928
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
929
    if (IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
930 931
      isInQuote = !isInQuote;
    }
X
Xiaoyu Wang 已提交
932
    if (!isInQuote && IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
933 934 935 936
      break;
    }
    sql++;
  }
X
Xiaoyu Wang 已提交
937
  if (isInQuote) {
938 939 940
    smlBuildInvalidDataMsg(msg, "only one quote", elements->cols);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
941
  elements->colsLen = sql - elements->cols;
X
Xiaoyu Wang 已提交
942
  if (elements->colsLen == 0) {
wmmhello's avatar
wmmhello 已提交
943 944 945
    smlBuildInvalidDataMsg(msg, "cols is empty", NULL);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
946

wmmhello's avatar
wmmhello 已提交
947 948 949
  // parse timestamp
  JUMP_SPACE(sql)
  elements->timestamp = sql;
wmmhello's avatar
wmmhello 已提交
950
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
951
    if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
952 953 954 955
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
956
  elements->timestampLen = sql - elements->timestamp;
wmmhello's avatar
wmmhello 已提交
957 958 959 960

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
961
static void smlParseTelnetElement(const char **sql, const char **data, int32_t *len) {
962
  while (**sql != '\0') {
X
Xiaoyu Wang 已提交
963
    if (**sql != SPACE && !(*data)) {
964
      *data = *sql;
X
Xiaoyu Wang 已提交
965
    } else if (**sql == SPACE && *data) {
966 967 968 969 970 971 972
      *len = *sql - *data;
      break;
    }
    (*sql)++;
  }
}

X
Xiaoyu Wang 已提交
973 974
static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTableName, SHashObj *dumplicateKey,
                                  SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
975
  const char *sql = data;
X
Xiaoyu Wang 已提交
976 977
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
  while (*sql != '\0') {
wmmhello's avatar
wmmhello 已提交
978
    JUMP_SPACE(sql)
X
Xiaoyu Wang 已提交
979
    if (*sql == '\0') break;
wmmhello's avatar
wmmhello 已提交
980

wmmhello's avatar
wmmhello 已提交
981
    const char *key = sql;
X
Xiaoyu Wang 已提交
982
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
983 984

    // parse key
X
Xiaoyu Wang 已提交
985 986
    while (*sql != '\0') {
      if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
987 988 989
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
990
      if (*sql == EQUAL) {
wmmhello's avatar
wmmhello 已提交
991 992
        keyLen = sql - key;
        sql++;
993 994
        break;
      }
wmmhello's avatar
wmmhello 已提交
995
      sql++;
996
    }
wmmhello's avatar
wmmhello 已提交
997

X
Xiaoyu Wang 已提交
998
    if (IS_INVALID_COL_LEN(keyLen)) {
999
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1000
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
1001
    }
X
Xiaoyu Wang 已提交
1002
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1003
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1004
      return TSDB_CODE_TSC_DUP_NAMES;
1005 1006 1007
    }

    // parse value
wmmhello's avatar
wmmhello 已提交
1008
    const char *value = sql;
X
Xiaoyu Wang 已提交
1009 1010
    int32_t     valueLen = 0;
    while (*sql != '\0') {
wmmhello's avatar
wmmhello 已提交
1011 1012
      // parse value
      if (*sql == SPACE) {
1013 1014
        break;
      }
wmmhello's avatar
wmmhello 已提交
1015 1016 1017 1018 1019
      if (*sql == EQUAL) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
1020
    }
wmmhello's avatar
wmmhello 已提交
1021
    valueLen = sql - value;
wmmhello's avatar
wmmhello 已提交
1022

X
Xiaoyu Wang 已提交
1023
    if (valueLen == 0) {
1024
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1025
      return TSDB_CODE_TSC_INVALID_VALUE;
1026
    }
wmmhello's avatar
wmmhello 已提交
1027

X
Xiaoyu Wang 已提交
1028 1029
    // handle child table name
    if (childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1030 1031 1032 1033 1034
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, value, (valueLen < TSDB_TABLE_NAME_LEN ? valueLen : TSDB_TABLE_NAME_LEN));
      continue;
    }

wmmhello's avatar
wmmhello 已提交
1035 1036 1037 1038
    if(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }

1039 1040
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1041
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1042 1043 1044
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1045
    kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1046
    kv->type = TSDB_DATA_TYPE_NCHAR;
1047

X
Xiaoyu Wang 已提交
1048
    if (cols) taosArrayPush(cols, &kv);
1049 1050 1051 1052
  }

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

1054
// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
X
Xiaoyu Wang 已提交
1055 1056
static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTableInfo *tinfo, SArray *cols) {
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
1057 1058 1059

  // parse metric
  smlParseTelnetElement(&sql, &tinfo->sTableName, &tinfo->sTableNameLen);
1060
  if (!(tinfo->sTableName) || IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
1061
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1062
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
1063 1064 1065 1066
  }

  // parse timestamp
  const char *timestamp = NULL;
X
Xiaoyu Wang 已提交
1067
  int32_t     tLen = 0;
1068 1069 1070 1071 1072 1073 1074 1075 1076
  smlParseTelnetElement(&sql, &timestamp, &tLen);
  if (!timestamp || tLen == 0) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
    return TSDB_CODE_SML_INVALID_DATA;
  }

  int32_t ret = smlParseTS(info, timestamp, tLen, cols);
  if (ret != TSDB_CODE_SUCCESS) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql);
wmmhello's avatar
wmmhello 已提交
1077
    return ret;
1078 1079 1080 1081
  }

  // parse value
  const char *value = NULL;
X
Xiaoyu Wang 已提交
1082
  int32_t     valueLen = 0;
1083 1084 1085
  smlParseTelnetElement(&sql, &value, &valueLen);
  if (!value || valueLen == 0) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
wmmhello's avatar
wmmhello 已提交
1086
    return TSDB_CODE_TSC_INVALID_VALUE;
1087 1088 1089
  }

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1090
  if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1091 1092 1093 1094
  taosArrayPush(cols, &kv);
  kv->key = VALUE;
  kv->keyLen = VALUE_LEN;
  kv->value = value;
wmmhello's avatar
wmmhello 已提交
1095
  kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1096 1097
  if ((ret = smlParseValue(kv, &info->msgBuf)) != TSDB_CODE_SUCCESS) {
    return ret;
1098 1099 1100
  }

  // parse tags
1101
  ret = smlParseTelnetTags(sql, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
1102 1103
  if (ret != TSDB_CODE_SUCCESS) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1104
    return ret;
1105 1106 1107 1108 1109
  }

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1110 1111 1112
static int32_t smlParseCols(const char *data, int32_t len, SArray *cols, char *childTableName, bool isTag,
                            SHashObj *dumplicateKey, SSmlMsgBuf *msg) {
  if (len == 0) {
wmmhello's avatar
wmmhello 已提交
1113
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1114 1115
  }

X
Xiaoyu Wang 已提交
1116
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
wmmhello's avatar
wmmhello 已提交
1117
  const char *sql = data;
X
Xiaoyu Wang 已提交
1118
  while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1119
    const char *key = sql;
X
Xiaoyu Wang 已提交
1120
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
1121

X
Xiaoyu Wang 已提交
1122
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1123
      // parse key
X
Xiaoyu Wang 已提交
1124
      if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1125 1126 1127
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
1128
      if (IS_EQUAL(sql)) {
wmmhello's avatar
wmmhello 已提交
1129 1130
        keyLen = sql - key;
        sql++;
wmmhello's avatar
wmmhello 已提交
1131 1132
        break;
      }
wmmhello's avatar
wmmhello 已提交
1133
      sql++;
wmmhello's avatar
wmmhello 已提交
1134
    }
wmmhello's avatar
wmmhello 已提交
1135

X
Xiaoyu Wang 已提交
1136
    if (IS_INVALID_COL_LEN(keyLen)) {
wmmhello's avatar
wmmhello 已提交
1137
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1138
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
wmmhello's avatar
wmmhello 已提交
1139
    }
X
Xiaoyu Wang 已提交
1140
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1141
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1142
      return TSDB_CODE_TSC_DUP_NAMES;
1143 1144
    }

wmmhello's avatar
wmmhello 已提交
1145
    // parse value
wmmhello's avatar
wmmhello 已提交
1146
    const char *value = sql;
X
Xiaoyu Wang 已提交
1147 1148 1149
    int32_t     valueLen = 0;
    bool        isInQuote = false;
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1150
      // parse value
X
Xiaoyu Wang 已提交
1151
      if (!isTag && IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
1152
        isInQuote = !isInQuote;
wmmhello's avatar
wmmhello 已提交
1153 1154
        sql++;
        continue;
wmmhello's avatar
wmmhello 已提交
1155
      }
wmmhello's avatar
wmmhello 已提交
1156
      if (!isInQuote && IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1157 1158
        break;
      }
wmmhello's avatar
wmmhello 已提交
1159 1160 1161 1162 1163
      if (!isInQuote && IS_EQUAL(sql)) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
1164
    }
wmmhello's avatar
wmmhello 已提交
1165 1166 1167
    valueLen = sql - value;
    sql++;

X
Xiaoyu Wang 已提交
1168
    if (isInQuote) {
wmmhello's avatar
wmmhello 已提交
1169 1170 1171
      smlBuildInvalidDataMsg(msg, "only one quote", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
X
Xiaoyu Wang 已提交
1172
    if (valueLen == 0) {
wmmhello's avatar
wmmhello 已提交
1173
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1174 1175
      return TSDB_CODE_SML_INVALID_DATA;
    }
wmmhello's avatar
wmmhello 已提交
1176 1177
    PROCESS_SLASH(key, keyLen)
    PROCESS_SLASH(value, valueLen)
wmmhello's avatar
wmmhello 已提交
1178

X
Xiaoyu Wang 已提交
1179 1180
    // handle child table name
    if (childTableName && childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1181 1182 1183 1184 1185
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, value, (valueLen < TSDB_TABLE_NAME_LEN ? valueLen : TSDB_TABLE_NAME_LEN));
      continue;
    }

wmmhello's avatar
wmmhello 已提交
1186
    // add kv to SSmlKv
wmmhello's avatar
wmmhello 已提交
1187
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1188 1189
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if (cols) taosArrayPush(cols, &kv);
1190

wmmhello's avatar
wmmhello 已提交
1191 1192 1193
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1194
    kv->length = valueLen;
X
Xiaoyu Wang 已提交
1195
    if (isTag) {
1196 1197 1198
      if(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
        return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
      }
wmmhello's avatar
wmmhello 已提交
1199
      kv->type = TSDB_DATA_TYPE_NCHAR;
X
Xiaoyu Wang 已提交
1200
    } else {
wmmhello's avatar
wmmhello 已提交
1201 1202 1203
      int32_t ret = smlParseValue(kv, msg);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
wmmhello's avatar
wmmhello 已提交
1204
      }
wmmhello's avatar
wmmhello 已提交
1205 1206
    }
  }
wmmhello's avatar
wmmhello 已提交
1207

wmmhello's avatar
wmmhello 已提交
1208 1209 1210
  return TSDB_CODE_SUCCESS;
}

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

wmmhello's avatar
wmmhello 已提交
1215
    int16_t *index = (int16_t *)taosHashGet(metaHash, kv->key, kv->keyLen);
X
Xiaoyu Wang 已提交
1216
    if (index) {
wmmhello's avatar
wmmhello 已提交
1217
      SSmlKv **value = (SSmlKv **)taosArrayGet(metaArray, *index);
X
Xiaoyu Wang 已提交
1218
      if (kv->type != (*value)->type) {
wmmhello's avatar
wmmhello 已提交
1219
        smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
wmmhello's avatar
wmmhello 已提交
1220
        return TSDB_CODE_SML_NOT_SAME_TYPE;
X
Xiaoyu Wang 已提交
1221 1222 1223
      } else {
        if (IS_VAR_DATA_TYPE(kv->type)) {  // update string len, if bigger
          if (kv->length > (*value)->length) {
wmmhello's avatar
wmmhello 已提交
1224
            *value = kv;
1225 1226 1227
          }
        }
      }
X
Xiaoyu Wang 已提交
1228
    } else {
wmmhello's avatar
wmmhello 已提交
1229 1230 1231 1232 1233
      size_t tmp = taosArrayGetSize(metaArray);
      ASSERT(tmp <= INT16_MAX);
      int16_t size = tmp;
      taosArrayPush(metaArray, &kv);
      taosHashPut(metaHash, kv->key, kv->keyLen, &size, SHORT_BYTES);
1234
    }
wmmhello's avatar
wmmhello 已提交
1235
  }
wmmhello's avatar
wmmhello 已提交
1236

wmmhello's avatar
wmmhello 已提交
1237
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1238 1239
}

X
Xiaoyu Wang 已提交
1240
static void smlInsertMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols) {
wmmhello's avatar
wmmhello 已提交
1241 1242 1243 1244
  for (int16_t i = 0; i < taosArrayGetSize(cols); ++i) {
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
    taosArrayPush(metaArray, &kv);
    taosHashPut(metaHash, kv->key, kv->keyLen, &i, SHORT_BYTES);
wmmhello's avatar
wmmhello 已提交
1245 1246 1247
  }
}

X
Xiaoyu Wang 已提交
1248
static SSmlTableInfo *smlBuildTableInfo() {
1249
  SSmlTableInfo *tag = (SSmlTableInfo *)taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
X
Xiaoyu Wang 已提交
1250
  if (!tag) {
1251
    return NULL;
wmmhello's avatar
wmmhello 已提交
1252
  }
1253 1254 1255 1256 1257

  tag->cols = taosArrayInit(16, POINTER_BYTES);
  if (tag->cols == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1258
  }
1259 1260 1261 1262 1263

  tag->tags = taosArrayInit(16, POINTER_BYTES);
  if (tag->tags == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1264
  }
1265 1266 1267 1268 1269
  return tag;

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

X
Xiaoyu Wang 已提交
1272 1273 1274
static void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag) {
  if (info->dataFormat) {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1275 1276
      SArray *kvArray = (SArray *)taosArrayGetP(tag->cols, i);
      for (int j = 0; j < taosArrayGetSize(kvArray); ++j) {
wmmhello's avatar
wmmhello 已提交
1277
        SSmlKv *p = (SSmlKv *)taosArrayGetP(kvArray, j);
X
Xiaoyu Wang 已提交
1278 1279 1280
        if (info->protocol == TSDB_SML_JSON_PROTOCOL &&
            (p->type == TSDB_DATA_TYPE_NCHAR || p->type == TSDB_DATA_TYPE_BINARY)) {
          taosMemoryFree((void *)p->value);
wmmhello's avatar
wmmhello 已提交
1281
        }
1282 1283 1284 1285
        taosMemoryFree(p);
      }
      taosArrayDestroy(kvArray);
    }
X
Xiaoyu Wang 已提交
1286 1287
  } else {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1288
      SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
X
Xiaoyu Wang 已提交
1289
      void    **p1 = (void **)taosHashIterate(kvHash, NULL);
1290 1291
      while (p1) {
        taosMemoryFree(*p1);
X
Xiaoyu Wang 已提交
1292
        p1 = (void **)taosHashIterate(kvHash, p1);
1293 1294 1295 1296
      }
      taosHashCleanup(kvHash);
    }
  }
X
Xiaoyu Wang 已提交
1297
  for (size_t i = 0; i < taosArrayGetSize(tag->tags); i++) {
wmmhello's avatar
wmmhello 已提交
1298
    SSmlKv *p = (SSmlKv *)taosArrayGetP(tag->tags, i);
X
Xiaoyu Wang 已提交
1299 1300 1301 1302
    if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
      taosMemoryFree((void *)p->key);
      if (p->type == TSDB_DATA_TYPE_NCHAR || p->type == TSDB_DATA_TYPE_BINARY) {
        taosMemoryFree((void *)p->value);
wmmhello's avatar
wmmhello 已提交
1303 1304 1305 1306
      }
    }
    taosMemoryFree(p);
  }
X
Xiaoyu Wang 已提交
1307 1308
  if (info->protocol == TSDB_SML_JSON_PROTOCOL && tag->sTableName) {
    taosMemoryFree((void *)tag->sTableName);
wmmhello's avatar
wmmhello 已提交
1309
  }
1310 1311 1312 1313 1314
  taosArrayDestroy(tag->cols);
  taosArrayDestroy(tag->tags);
  taosMemoryFree(tag);
}

X
Xiaoyu Wang 已提交
1315
static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) {
1316 1317
  SArray *s1 = *(SArray **)key1;
  SArray *s2 = *(SArray **)key2;
1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330
  SSmlKv *kv1 = (SSmlKv *)taosArrayGetP(s1, 0);
  SSmlKv *kv2 = (SSmlKv *)taosArrayGetP(s2, 0);
  ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP);
  ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP);
  if (kv1->i < kv2->i) {
    return -1;
  } else if (kv1->i > kv2->i) {
    return 1;
  } else {
    return 0;
  }
}

X
Xiaoyu Wang 已提交
1331
static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
1332 1333
  SHashObj *s1 = *(SHashObj **)key1;
  SHashObj *s2 = *(SHashObj **)key2;
1334 1335
  SSmlKv *kv1 = *(SSmlKv **)taosHashGet(s1, TS, TS_LEN);
  SSmlKv *kv2 = *(SSmlKv **)taosHashGet(s2, TS, TS_LEN);
1336 1337 1338 1339 1340 1341 1342 1343 1344 1345 1346
  ASSERT(kv1->type == TSDB_DATA_TYPE_TIMESTAMP);
  ASSERT(kv2->type == TSDB_DATA_TYPE_TIMESTAMP);
  if (kv1->i < kv2->i) {
    return -1;
  } else if (kv1->i > kv2->i) {
    return 1;
  } else {
    return 0;
  }
}

1347 1348
static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
  if(dataFormat){
1349
    void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GT);
1350 1351
    if(p == NULL){
      taosArrayPush(oneTable->cols, &cols);
1352 1353
    }else{
      taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
1354
    }
1355 1356 1357 1358
    return TSDB_CODE_SUCCESS;
  }

  SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1359
  if (!kvHash) {
1360 1361 1362
    uError("SML:smlDealCols failed to allocate memory");
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
1363
  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
1364
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
wmmhello's avatar
wmmhello 已提交
1365
    taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
1366 1367
  }

1368
  void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GT);
1369 1370
  if(p == NULL){
    taosArrayPush(oneTable->cols, &kvHash);
1371 1372
  }else{
    taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &kvHash);
1373
  }
1374 1375 1376
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1377 1378 1379
static SSmlSTableMeta *smlBuildSTableMeta() {
  SSmlSTableMeta *meta = (SSmlSTableMeta *)taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if (!meta) {
1380 1381 1382 1383 1384 1385 1386 1387
    return NULL;
  }
  meta->tagHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (meta->tagHash == NULL) {
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }

wmmhello's avatar
wmmhello 已提交
1388 1389
  meta->colHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (meta->colHash == NULL) {
1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411
    uError("SML:smlBuildSTableMeta failed to allocate memory");
    goto cleanup;
  }

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

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

cleanup:
  taosMemoryFree(meta);
  return NULL;
}

X
Xiaoyu Wang 已提交
1412
static void smlDestroySTableMeta(SSmlSTableMeta *meta) {
1413
  taosHashCleanup(meta->tagHash);
wmmhello's avatar
wmmhello 已提交
1414
  taosHashCleanup(meta->colHash);
1415 1416 1417
  taosArrayDestroy(meta->tags);
  taosArrayDestroy(meta->cols);
  taosMemoryFree(meta->tableMeta);
wmmhello's avatar
wmmhello 已提交
1418
  taosMemoryFree(meta);
1419 1420 1421 1422 1423
}

static void smlDestroyCols(SArray *cols) {
  if (!cols) return;
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
wmmhello's avatar
wmmhello 已提交
1424
    void *kv = taosArrayGetP(cols, i);
1425 1426 1427 1428
    taosMemoryFree(kv);
  }
}

X
Xiaoyu Wang 已提交
1429 1430
static void smlDestroyInfo(SSmlHandle *info) {
  if (!info) return;
1431 1432 1433 1434
  qDestroyQuery(info->pQuery);
  smlDestroyHandle(info->exec);

  // destroy info->childTables
X
Xiaoyu Wang 已提交
1435
  void **p1 = (void **)taosHashIterate(info->childTables, NULL);
1436
  while (p1) {
X
Xiaoyu Wang 已提交
1437 1438
    smlDestroyTableInfo(info, (SSmlTableInfo *)(*p1));
    p1 = (void **)taosHashIterate(info->childTables, p1);
1439 1440 1441 1442
  }
  taosHashCleanup(info->childTables);

  // destroy info->superTables
X
Xiaoyu Wang 已提交
1443
  p1 = (void **)taosHashIterate(info->superTables, NULL);
1444
  while (p1) {
X
Xiaoyu Wang 已提交
1445 1446
    smlDestroySTableMeta((SSmlSTableMeta *)(*p1));
    p1 = (void **)taosHashIterate(info->superTables, p1);
1447 1448 1449 1450 1451 1452
  }
  taosHashCleanup(info->superTables);

  // destroy info->pVgHash
  taosHashCleanup(info->pVgHash);
  taosHashCleanup(info->dumplicateKey);
X
Xiaoyu Wang 已提交
1453
  if (!info->dataFormat) {
wmmhello's avatar
wmmhello 已提交
1454 1455
    taosArrayDestroy(info->colsContainer);
  }
wmmhello's avatar
wmmhello 已提交
1456
  destroyRequest(info->pRequest);
1457 1458 1459
  taosMemoryFreeClear(info);
}

1460
static SSmlHandle* smlBuildSmlInfo(STscObj* pTscObj, SRequestObj* request, SMLProtocolType protocol, int8_t precision){
1461 1462 1463 1464 1465
  int32_t code = TSDB_CODE_SUCCESS;
  SSmlHandle* info = (SSmlHandle*)taosMemoryCalloc(1, sizeof(SSmlHandle));
  if (NULL == info) {
    return NULL;
  }
X
Xiaoyu Wang 已提交
1466
  info->id = smlGenId();
1467

1468
  info->pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY);
1469
  if (NULL == info->pQuery) {
X
Xiaoyu Wang 已提交
1470
    uError("SML:0x%" PRIx64 " create info->pQuery error", info->id);
1471 1472
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
1473
  info->pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
1474
  info->pQuery->haveResultSet = false;
X
Xiaoyu Wang 已提交
1475 1476 1477 1478
  info->pQuery->msgType = TDMT_VND_SUBMIT;
  info->pQuery->pRoot = (SNode *)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT);
  if (NULL == info->pQuery->pRoot) {
    uError("SML:0x%" PRIx64 " create info->pQuery->pRoot error", info->id);
1479 1480
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
1481
  ((SVnodeModifOpStmt *)(info->pQuery->pRoot))->payloadType = PAYLOAD_TYPE_KV;
1482

1483 1484 1485 1486 1487 1488 1489
  if (pTscObj){
    info->taos        = pTscObj;
    code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " get catalog error %d", info->id, code);
      goto cleanup;
    }
wmmhello's avatar
wmmhello 已提交
1490
  }
1491

X
Xiaoyu Wang 已提交
1492 1493 1494
  info->precision = precision;
  info->protocol = protocol;
  if (protocol == TSDB_SML_LINE_PROTOCOL) {
1495
    info->dataFormat = tsSmlDataFormat;
X
Xiaoyu Wang 已提交
1496
  } else {
1497 1498
    info->dataFormat = true;
  }
1499 1500 1501 1502 1503 1504

  if(request){
    info->pRequest = request;
    info->msgBuf.buf = info->pRequest->msgBuf;
    info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
  }
1505

X
Xiaoyu Wang 已提交
1506
  info->exec = smlInitHandle(info->pQuery);
1507 1508
  info->childTables = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  info->superTables = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1509
  info->pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
1510 1511

  info->dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1512
  if (!info->dataFormat) {
1513
    info->colsContainer = taosArrayInit(32, POINTER_BYTES);
X
Xiaoyu Wang 已提交
1514 1515
    if (NULL == info->colsContainer) {
      uError("SML:0x%" PRIx64 " create info failed", info->id);
1516 1517 1518
      goto cleanup;
    }
  }
X
Xiaoyu Wang 已提交
1519 1520 1521
  if (NULL == info->exec || NULL == info->childTables || NULL == info->superTables || NULL == info->pVgHash ||
      NULL == info->dumplicateKey) {
    uError("SML:0x%" PRIx64 " create info failed", info->id);
1522 1523 1524 1525 1526 1527 1528 1529 1530 1531
    goto cleanup;
  }

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

/************* TSDB_SML_JSON_PROTOCOL function start **************/
X
Xiaoyu Wang 已提交
1532
static int32_t smlJsonCreateSring(const char **output, char *input, int32_t inputLen) {
1533
  *output = (const char *)taosMemoryMalloc(inputLen);
X
Xiaoyu Wang 已提交
1534
  if (*output == NULL) {
1535 1536 1537
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

X
Xiaoyu Wang 已提交
1538
  memcpy((void *)(*output), input, inputLen);
1539 1540 1541 1542 1543 1544
  return TSDB_CODE_SUCCESS;
}

static int32_t smlParseMetricFromJSON(SSmlHandle *info, cJSON *root, SSmlTableInfo *tinfo) {
  cJSON *metric = cJSON_GetObjectItem(root, "metric");
  if (!cJSON_IsString(metric)) {
X
Xiaoyu Wang 已提交
1545
    return TSDB_CODE_TSC_INVALID_JSON;
1546 1547 1548
  }

  tinfo->sTableNameLen = strlen(metric->valuestring);
1549
  if (IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
X
Xiaoyu Wang 已提交
1550
    uError("OTD:0x%" PRIx64 " Metric lenght is 0 or large than 192", info->id);
1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
  }

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

static int32_t smlParseTSFromJSONObj(SSmlHandle *info, cJSON *root, int64_t *tsVal) {
  int32_t size = cJSON_GetArraySize(root);
  if (size != OTD_JSON_SUB_FIELDS_NUM) {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

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

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

  double timeDouble = value->valuedouble;
X
Xiaoyu Wang 已提交
1574
  if (smlDoubleToInt64OverFlow(timeDouble)) {
1575
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1576
    return TSDB_CODE_INVALID_TIMESTAMP;
1577
  }
wmmhello's avatar
wmmhello 已提交
1578 1579 1580 1581 1582 1583 1584 1585

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

  if (timeDouble < 0) {
    return TSDB_CODE_INVALID_TIMESTAMP;
1586 1587
  }

1588
  *tsVal = timeDouble;
1589
  size_t typeLen = strlen(type->valuestring);
wmmhello's avatar
wmmhello 已提交
1590
  if (typeLen == 1 && (type->valuestring[0] == 's' || type->valuestring[0] == 'S')) {
X
Xiaoyu Wang 已提交
1591
    // seconds
1592 1593
    *tsVal = *tsVal * NANOSECOND_PER_SEC;
    timeDouble = timeDouble * NANOSECOND_PER_SEC;
X
Xiaoyu Wang 已提交
1594
    if (smlDoubleToInt64OverFlow(timeDouble)) {
1595
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1596
      return TSDB_CODE_INVALID_TIMESTAMP;
1597
    }
wmmhello's avatar
wmmhello 已提交
1598
  } else if (typeLen == 2 && (type->valuestring[1] == 's' || type->valuestring[1] == 'S')) {
1599 1600
    switch (type->valuestring[0]) {
      case 'm':
wmmhello's avatar
wmmhello 已提交
1601
      case 'M':
X
Xiaoyu Wang 已提交
1602
        // milliseconds
1603 1604
        *tsVal = *tsVal * NANOSECOND_PER_MSEC;
        timeDouble = timeDouble * NANOSECOND_PER_MSEC;
X
Xiaoyu Wang 已提交
1605
        if (smlDoubleToInt64OverFlow(timeDouble)) {
1606
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1607
          return TSDB_CODE_INVALID_TIMESTAMP;
1608 1609 1610
        }
        break;
      case 'u':
wmmhello's avatar
wmmhello 已提交
1611
      case 'U':
X
Xiaoyu Wang 已提交
1612
        // microseconds
1613 1614
        *tsVal = *tsVal * NANOSECOND_PER_USEC;
        timeDouble = timeDouble * NANOSECOND_PER_USEC;
X
Xiaoyu Wang 已提交
1615
        if (smlDoubleToInt64OverFlow(timeDouble)) {
1616
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1617
          return TSDB_CODE_INVALID_TIMESTAMP;
1618 1619 1620
        }
        break;
      case 'n':
wmmhello's avatar
wmmhello 已提交
1621
      case 'N':
1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642
        break;
      default:
        return TSDB_CODE_TSC_INVALID_JSON;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON;
  }

  return TSDB_CODE_SUCCESS;
}

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

static int32_t smlParseTSFromJSON(SSmlHandle *info, cJSON *root, SArray *cols) {
X
Xiaoyu Wang 已提交
1643
  // Timestamp must be the first KV to parse
1644 1645 1646 1647
  int64_t tsVal = 0;

  cJSON *timestamp = cJSON_GetObjectItem(root, "timestamp");
  if (cJSON_IsNumber(timestamp)) {
X
Xiaoyu Wang 已提交
1648
    // timestamp value 0 indicates current system time
1649
    double timeDouble = timestamp->valuedouble;
X
Xiaoyu Wang 已提交
1650
    if (smlDoubleToInt64OverFlow(timeDouble)) {
1651
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1652
      return TSDB_CODE_INVALID_TIMESTAMP;
1653
    }
wmmhello's avatar
wmmhello 已提交
1654

X
Xiaoyu Wang 已提交
1655
    if (timeDouble < 0) {
wmmhello's avatar
wmmhello 已提交
1656
      return TSDB_CODE_INVALID_TIMESTAMP;
1657
    }
1658

1659
    uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);
1660
    tsVal = (int64_t)timeDouble;
1661
    if (tsLen == TSDB_TIME_PRECISION_SEC_DIGITS) {
1662 1663
      tsVal = tsVal * NANOSECOND_PER_SEC;
      timeDouble = timeDouble * NANOSECOND_PER_SEC;
X
Xiaoyu Wang 已提交
1664
      if (smlDoubleToInt64OverFlow(timeDouble)) {
1665
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1666
        return TSDB_CODE_INVALID_TIMESTAMP;
1667 1668
      }
    } else if (tsLen == TSDB_TIME_PRECISION_MILLI_DIGITS) {
1669 1670
      tsVal = tsVal * NANOSECOND_PER_MSEC;
      timeDouble = timeDouble * NANOSECOND_PER_MSEC;
X
Xiaoyu Wang 已提交
1671
      if (smlDoubleToInt64OverFlow(timeDouble)) {
1672
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1673
        return TSDB_CODE_INVALID_TIMESTAMP;
1674
      }
X
Xiaoyu Wang 已提交
1675
    } else if (timeDouble == 0) {
wmmhello's avatar
wmmhello 已提交
1676
      tsVal = taosGetTimestampNs();
X
Xiaoyu Wang 已提交
1677
    } else {
wmmhello's avatar
wmmhello 已提交
1678
      return TSDB_CODE_INVALID_TIMESTAMP;
1679 1680 1681 1682
    }
  } else if (cJSON_IsObject(timestamp)) {
    int32_t ret = smlParseTSFromJSONObj(info, timestamp, &tsVal);
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
1683
      uError("SML:0x%" PRIx64 " Failed to parse timestamp from JSON Obj", info->id);
1684 1685 1686 1687
      return ret;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
1688 1689
  }

wmmhello's avatar
wmmhello 已提交
1690
  // add ts to
wmmhello's avatar
wmmhello 已提交
1691
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1692
  if (!kv) {
wmmhello's avatar
wmmhello 已提交
1693 1694 1695
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  kv->key = TS;
1696 1697
  kv->keyLen = TS_LEN;
  kv->i = tsVal;
wmmhello's avatar
wmmhello 已提交
1698
  kv->type = TSDB_DATA_TYPE_TIMESTAMP;
wmmhello's avatar
wmmhello 已提交
1699
  kv->length = (int16_t)tDataTypes[kv->type].bytes;
X
Xiaoyu Wang 已提交
1700
  if (cols) taosArrayPush(cols, &kv);
wmmhello's avatar
wmmhello 已提交
1701 1702 1703
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1704
static int32_t smlConvertJSONBool(SSmlKv *pVal, char *typeStr, cJSON *value) {
1705 1706 1707 1708 1709 1710 1711
  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 已提交
1712

1713 1714
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
1715

X
Xiaoyu Wang 已提交
1716 1717 1718
static int32_t smlConvertJSONNumber(SSmlKv *pVal, char *typeStr, cJSON *value) {
  // tinyint
  if (strcasecmp(typeStr, "i8") == 0 || strcasecmp(typeStr, "tinyint") == 0) {
1719 1720 1721 1722 1723 1724 1725 1726 1727
    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 已提交
1728 1729
  // smallint
  if (strcasecmp(typeStr, "i16") == 0 || strcasecmp(typeStr, "smallint") == 0) {
1730 1731 1732 1733 1734 1735 1736 1737 1738
    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 已提交
1739 1740
  // int
  if (strcasecmp(typeStr, "i32") == 0 || strcasecmp(typeStr, "int") == 0) {
1741 1742 1743 1744 1745 1746 1747 1748 1749
    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 已提交
1750 1751
  // bigint
  if (strcasecmp(typeStr, "i64") == 0 || strcasecmp(typeStr, "bigint") == 0) {
1752 1753
    pVal->type = TSDB_DATA_TYPE_BIGINT;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
X
Xiaoyu Wang 已提交
1754
    if (value->valuedouble >= (double)INT64_MAX) {
1755
      pVal->i = INT64_MAX;
X
Xiaoyu Wang 已提交
1756
    } else if (value->valuedouble <= (double)INT64_MIN) {
1757
      pVal->i = INT64_MIN;
X
Xiaoyu Wang 已提交
1758
    } else {
1759
      pVal->i = value->valuedouble;
1760 1761 1762
    }
    return TSDB_CODE_SUCCESS;
  }
X
Xiaoyu Wang 已提交
1763 1764
  // float
  if (strcasecmp(typeStr, "f32") == 0 || strcasecmp(typeStr, "float") == 0) {
1765 1766 1767
    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 已提交
1768
    }
1769 1770 1771 1772 1773
    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 已提交
1774 1775
  // double
  if (strcasecmp(typeStr, "f64") == 0 || strcasecmp(typeStr, "double") == 0) {
1776 1777 1778 1779
    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 已提交
1780 1781
  }

X
Xiaoyu Wang 已提交
1782
  // if reach here means type is unsupported
1783 1784 1785
  uError("OTD:invalid type(%s) for JSON Number", typeStr);
  return TSDB_CODE_TSC_INVALID_JSON_TYPE;
}
1786

X
Xiaoyu Wang 已提交
1787
static int32_t smlConvertJSONString(SSmlKv *pVal, char *typeStr, cJSON *value) {
1788 1789 1790 1791 1792 1793 1794 1795 1796
  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 已提交
1797 1798 1799 1800 1801 1802 1803 1804

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

wmmhello's avatar
wmmhello 已提交
1805
  return smlJsonCreateSring(&pVal->value, value->valuestring, pVal->length);
1806 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
}

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 已提交
1832
      }
1833
      break;
wmmhello's avatar
wmmhello 已提交
1834
    }
1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850
    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 已提交
1851
  }
1852 1853

  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1854 1855
}

1856 1857 1858 1859 1860 1861 1862 1863
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 已提交
1864
    }
1865 1866 1867 1868 1869 1870 1871 1872 1873 1874
    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 已提交
1875

X
Xiaoyu Wang 已提交
1876
      char *tsDefaultJSONStrType = "nchar";  // todo
1877 1878
      smlConvertJSONString(kv, tsDefaultJSONStrType, root);
      break;
wmmhello's avatar
wmmhello 已提交
1879
    }
1880 1881 1882 1883 1884 1885 1886 1887 1888 1889
    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 已提交
1890
  }
1891 1892 1893 1894 1895 1896 1897 1898 1899 1900 1901

  return TSDB_CODE_SUCCESS;
}

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

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1902
  if (!kv) {
1903 1904
    return TSDB_CODE_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
1905
  if (cols) taosArrayPush(cols, &kv);
1906 1907 1908 1909 1910 1911 1912 1913

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

X
Xiaoyu Wang 已提交
1916 1917
static int32_t smlParseTagsFromJSON(cJSON *root, SArray *pKVs, char *childTableName, SHashObj *dumplicateKey,
                                    SSmlMsgBuf *msg) {
1918 1919 1920 1921 1922 1923
  int32_t ret = TSDB_CODE_SUCCESS;

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

X
Xiaoyu Wang 已提交
1925
  size_t  childTableNameLen = strlen(tsSmlChildTableName);
1926 1927 1928 1929 1930
  int32_t tagNum = cJSON_GetArraySize(tags);
  for (int32_t i = 0; i < tagNum; ++i) {
    cJSON *tag = cJSON_GetArrayItem(tags, i);
    if (tag == NULL) {
      return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
1931
    }
1932 1933 1934 1935 1936
    size_t keyLen = strlen(tag->string);
    if (IS_INVALID_COL_LEN(keyLen)) {
      uError("OTD:Tag key length is 0 or too large than 64");
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
    }
X
Xiaoyu Wang 已提交
1937
    // check duplicate keys
1938
    if (smlCheckDuplicateKey(tag->string, keyLen, dumplicateKey)) {
wmmhello's avatar
wmmhello 已提交
1939
      return TSDB_CODE_TSC_DUP_NAMES;
wmmhello's avatar
wmmhello 已提交
1940 1941
    }

X
Xiaoyu Wang 已提交
1942 1943
    // handle child table name
    if (childTableNameLen != 0 && strcmp(tag->string, tsSmlChildTableName) == 0) {
1944 1945 1946 1947 1948 1949 1950 1951 1952
      if (!cJSON_IsString(tag)) {
        uError("OTD:ID must be JSON string");
        return TSDB_CODE_TSC_INVALID_JSON;
      }
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, tag->valuestring, TSDB_TABLE_NAME_LEN);
      continue;
    }

1953 1954
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1955 1956
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if (pKVs) taosArrayPush(pKVs, &kv);
1957

X
Xiaoyu Wang 已提交
1958
    // key
1959
    kv->keyLen = keyLen;
1960 1961 1962 1963
    ret = smlJsonCreateSring(&kv->key, tag->string, kv->keyLen);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
X
Xiaoyu Wang 已提交
1964
    // value
1965 1966 1967 1968
    ret = smlParseValueFromJSON(tag, kv);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
wmmhello's avatar
wmmhello 已提交
1969 1970
  }

1971
  return ret;
wmmhello's avatar
wmmhello 已提交
1972 1973
}

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

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

1982
  int32_t size = cJSON_GetArraySize(root);
X
Xiaoyu Wang 已提交
1983
  // outmost json fields has to be exactly 4
1984
  if (size != OTD_JSON_FIELDS_NUM) {
X
Xiaoyu Wang 已提交
1985
    uError("OTD:0x%" PRIx64 " Invalid number of JSON fields in data point %d", info->id, size);
1986
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
1987
  }
1988

X
Xiaoyu Wang 已提交
1989
  // Parse metric
1990 1991
  ret = smlParseMetricFromJSON(info, root, tinfo);
  if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
1992
    uError("OTD:0x%" PRIx64 " Unable to parse metric from JSON payload", info->id);
1993
    return ret;
wmmhello's avatar
wmmhello 已提交
1994
  }
X
Xiaoyu Wang 已提交
1995
  uDebug("OTD:0x%" PRIx64 " Parse metric from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
1996

X
Xiaoyu Wang 已提交
1997
  // Parse timestamp
1998 1999
  ret = smlParseTSFromJSON(info, root, cols);
  if (ret) {
X
Xiaoyu Wang 已提交
2000
    uError("OTD:0x%" PRIx64 " Unable to parse timestamp from JSON payload", info->id);
2001
    return ret;
wmmhello's avatar
wmmhello 已提交
2002
  }
X
Xiaoyu Wang 已提交
2003
  uDebug("OTD:0x%" PRIx64 " Parse timestamp from JSON payload finished", info->id);
2004

X
Xiaoyu Wang 已提交
2005
  // Parse metric value
2006 2007
  ret = smlParseColsFromJSON(root, cols);
  if (ret) {
X
Xiaoyu Wang 已提交
2008
    uError("OTD:0x%" PRIx64 " Unable to parse metric value from JSON payload", info->id);
2009
    return ret;
2010
  }
X
Xiaoyu Wang 已提交
2011
  uDebug("OTD:0x%" PRIx64 " Parse metric value from JSON payload finished", info->id);
2012

X
Xiaoyu Wang 已提交
2013
  // Parse tags
2014
  ret = smlParseTagsFromJSON(root, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
2015
  if (ret) {
X
Xiaoyu Wang 已提交
2016
    uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
2017
    return ret;
2018
  }
X
Xiaoyu Wang 已提交
2019
  uDebug("OTD:0x%" PRIx64 " Parse tags from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2020

2021
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2022
}
2023
/************* TSDB_SML_JSON_PROTOCOL function end **************/
wmmhello's avatar
wmmhello 已提交
2024

X
Xiaoyu Wang 已提交
2025
static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql) {
wmmhello's avatar
wmmhello 已提交
2026
  SSmlLineInfo elements = {0};
2027
  uDebug("SML:0x%" PRIx64 " smlParseInfluxLine sql:%s, hello", info->id, sql);
2028

X
Xiaoyu Wang 已提交
2029 2030 2031
  int          ret = smlParseInfluxString(sql, &elements, &info->msgBuf);
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseInfluxLine failed", info->id);
wmmhello's avatar
wmmhello 已提交
2032 2033 2034
    return ret;
  }

2035
  SArray *cols = NULL;
X
Xiaoyu Wang 已提交
2036
  if (info->dataFormat) {  // if dataFormat, cols need new memory to save data
2037 2038
    cols = taosArrayInit(16, POINTER_BYTES);
    if (cols == NULL) {
X
Xiaoyu Wang 已提交
2039
      uError("SML:0x%" PRIx64 " smlParseInfluxLine failed to allocate memory", info->id);
2040 2041
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
X
Xiaoyu Wang 已提交
2042
  } else {  // if dataFormat is false, cols do not need to save data, there is another new memory to save data
2043
    cols = info->colsContainer;
wmmhello's avatar
wmmhello 已提交
2044
  }
wmmhello's avatar
wmmhello 已提交
2045

wmmhello's avatar
wmmhello 已提交
2046
  ret = smlParseTS(info, elements.timestamp, elements.timestampLen, cols);
X
Xiaoyu Wang 已提交
2047 2048 2049
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseTS failed", info->id);
    if (info->dataFormat) taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2050 2051
    return ret;
  }
2052
  ret = smlParseCols(elements.cols, elements.colsLen, cols, NULL, false, info->dumplicateKey, &info->msgBuf);
X
Xiaoyu Wang 已提交
2053 2054
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseCols parse cloums fields failed", info->id);
2055
    smlDestroyCols(cols);
X
Xiaoyu Wang 已提交
2056
    if (info->dataFormat) taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2057 2058
    return ret;
  }
X
Xiaoyu Wang 已提交
2059
  if (taosArrayGetSize(cols) > TSDB_MAX_COLUMNS) {
wmmhello's avatar
wmmhello 已提交
2060
    smlBuildInvalidDataMsg(&info->msgBuf, "too many columns than 4096", NULL);
wmmhello's avatar
wmmhello 已提交
2061
    return TSDB_CODE_PAR_TOO_MANY_COLUMNS;
wmmhello's avatar
wmmhello 已提交
2062
  }
wmmhello's avatar
wmmhello 已提交
2063

X
Xiaoyu Wang 已提交
2064 2065 2066 2067 2068
  bool            hasTable = true;
  SSmlTableInfo  *tinfo = NULL;
  SSmlTableInfo **oneTable =
      (SSmlTableInfo **)taosHashGet(info->childTables, elements.measure, elements.measureTagsLen);
  if (!oneTable) {
2069
    tinfo = smlBuildTableInfo();
X
Xiaoyu Wang 已提交
2070
    if (!tinfo) {
wmmhello's avatar
wmmhello 已提交
2071 2072
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
2073
    taosHashPut(info->childTables, elements.measure, elements.measureTagsLen, &tinfo, POINTER_BYTES);
2074
    oneTable = &tinfo;
2075 2076 2077 2078
    hasTable = false;
  }

  ret = smlDealCols(*oneTable, info->dataFormat, cols);
X
Xiaoyu Wang 已提交
2079
  if (ret != TSDB_CODE_SUCCESS) {
2080 2081
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2082

X
Xiaoyu Wang 已提交
2083 2084 2085 2086 2087
  if (!hasTable) {
    ret = smlParseCols(elements.tags, elements.tagsLen, (*oneTable)->tags, (*oneTable)->childTableName, true,
                       info->dumplicateKey, &info->msgBuf);
    if (ret != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlParseCols parse tag fields failed", info->id);
wmmhello's avatar
wmmhello 已提交
2088 2089 2090
      return ret;
    }

X
Xiaoyu Wang 已提交
2091
    if (taosArrayGetSize((*oneTable)->tags) > TSDB_MAX_TAGS) {
wmmhello's avatar
wmmhello 已提交
2092
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
wmmhello's avatar
wmmhello 已提交
2093
      return TSDB_CODE_PAR_INVALID_TAGS_NUM;
wmmhello's avatar
wmmhello 已提交
2094 2095
    }

2096 2097
    (*oneTable)->sTableName = elements.measure;
    (*oneTable)->sTableNameLen = elements.measureLen;
X
Xiaoyu Wang 已提交
2098 2099 2100
    if (strlen((*oneTable)->childTableName) == 0) {
      RandTableName rName = {(*oneTable)->tags, (*oneTable)->sTableName, (uint8_t)(*oneTable)->sTableNameLen,
                             (*oneTable)->childTableName, 0};
2101 2102 2103

      buildChildTableName(&rName);
      (*oneTable)->uid = rName.uid;
X
Xiaoyu Wang 已提交
2104 2105
    } else {
      (*oneTable)->uid = *(uint64_t *)((*oneTable)->childTableName);
2106
    }
2107
  }
wmmhello's avatar
wmmhello 已提交
2108

X
Xiaoyu Wang 已提交
2109 2110
  SSmlSTableMeta **tableMeta = (SSmlSTableMeta **)taosHashGet(info->superTables, elements.measure, elements.measureLen);
  if (tableMeta) {  // update meta
wmmhello's avatar
wmmhello 已提交
2111
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
2112
    if (!hasTable && ret == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2113 2114
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
wmmhello's avatar
wmmhello 已提交
2115
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2116
      uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
wmmhello's avatar
wmmhello 已提交
2117
      return ret;
wmmhello's avatar
wmmhello 已提交
2118
    }
X
Xiaoyu Wang 已提交
2119
  } else {
2120
    SSmlSTableMeta *meta = smlBuildSTableMeta();
wmmhello's avatar
wmmhello 已提交
2121 2122
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
2123
    taosHashPut(info->superTables, elements.measure, elements.measureLen, &meta, POINTER_BYTES);
wmmhello's avatar
wmmhello 已提交
2124
  }
2125

X
Xiaoyu Wang 已提交
2126
  if (!info->dataFormat) {
2127 2128 2129
    taosArrayClear(info->colsContainer);
  }
  taosHashClear(info->dumplicateKey);
wmmhello's avatar
wmmhello 已提交
2130 2131 2132
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2133 2134
static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) {
  int            ret = TSDB_CODE_SUCCESS;
2135
  SSmlTableInfo *tinfo = smlBuildTableInfo();
X
Xiaoyu Wang 已提交
2136
  if (!tinfo) {
2137
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2138 2139
  }

2140 2141
  SArray *cols = taosArrayInit(16, POINTER_BYTES);
  if (cols == NULL) {
X
Xiaoyu Wang 已提交
2142
    uError("SML:0x%" PRIx64 " smlParseTelnetLine failed to allocate memory", info->id);
2143
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2144 2145
  }

X
Xiaoyu Wang 已提交
2146 2147 2148
  if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
    ret = smlParseTelnetString(info, (const char *)data, tinfo, cols);
  } else if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
2149
    ret = smlParseJSONString(info, (cJSON *)data, tinfo, cols);
X
Xiaoyu Wang 已提交
2150
  } else {
2151 2152
    ASSERT(0);
  }
X
Xiaoyu Wang 已提交
2153 2154
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseTelnetLine failed", info->id);
wmmhello's avatar
wmmhello 已提交
2155
    smlDestroyTableInfo(info, tinfo);
wmmhello's avatar
wmmhello 已提交
2156
    smlDestroyCols(cols);
2157 2158
    taosArrayDestroy(cols);
    return ret;
wmmhello's avatar
wmmhello 已提交
2159
  }
wmmhello's avatar
wmmhello 已提交
2160

X
Xiaoyu Wang 已提交
2161
  if (taosArrayGetSize(tinfo->tags) <= 0 || taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS) {
2162
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate tags length:[1,128]", NULL);
wmmhello's avatar
wmmhello 已提交
2163 2164 2165
    smlDestroyTableInfo(info, tinfo);
    smlDestroyCols(cols);
    taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2166
    return TSDB_CODE_PAR_INVALID_TAGS_NUM;
wmmhello's avatar
wmmhello 已提交
2167
  }
2168 2169
  taosHashClear(info->dumplicateKey);

X
Xiaoyu Wang 已提交
2170 2171
  if (strlen(tinfo->childTableName) == 0) {
    RandTableName rName = {tinfo->tags, tinfo->sTableName, (uint8_t)tinfo->sTableNameLen, tinfo->childTableName, 0};
2172 2173
    buildChildTableName(&rName);
    tinfo->uid = rName.uid;
X
Xiaoyu Wang 已提交
2174 2175
  } else {
    tinfo->uid = *(uint64_t *)(tinfo->childTableName);  // generate uid by name simple
2176 2177
  }

X
Xiaoyu Wang 已提交
2178 2179 2180 2181
  bool            hasTable = true;
  SSmlTableInfo **oneTable =
      (SSmlTableInfo **)taosHashGet(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName));
  if (!oneTable) {
2182
    taosHashPut(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName), &tinfo, POINTER_BYTES);
2183
    oneTable = &tinfo;
2184
    hasTable = false;
X
Xiaoyu Wang 已提交
2185
  } else {
wmmhello's avatar
wmmhello 已提交
2186
    smlDestroyTableInfo(info, tinfo);
wmmhello's avatar
wmmhello 已提交
2187
  }
wmmhello's avatar
wmmhello 已提交
2188

2189
  taosArrayPush((*oneTable)->cols, &cols);
X
Xiaoyu Wang 已提交
2190 2191 2192
  SSmlSTableMeta **tableMeta =
      (SSmlSTableMeta **)taosHashGet(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen);
  if (tableMeta) {  // update meta
wmmhello's avatar
wmmhello 已提交
2193
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
2194
    if (!hasTable && ret == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2195 2196
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
wmmhello's avatar
wmmhello 已提交
2197
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2198
      uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
wmmhello's avatar
wmmhello 已提交
2199
      return ret;
2200
    }
X
Xiaoyu Wang 已提交
2201
  } else {
2202
    SSmlSTableMeta *meta = smlBuildSTableMeta();
wmmhello's avatar
wmmhello 已提交
2203 2204
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
2205
    taosHashPut(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen, &meta, POINTER_BYTES);
wmmhello's avatar
wmmhello 已提交
2206 2207
  }

2208 2209
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
2210

X
Xiaoyu Wang 已提交
2211
static int32_t smlParseJSON(SSmlHandle *info, char *payload) {
2212 2213
  int32_t payloadNum = 0;
  int32_t ret = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2214

2215
  if (payload == NULL) {
X
Xiaoyu Wang 已提交
2216
    uError("SML:0x%" PRIx64 " empty JSON Payload", info->id);
2217
    return TSDB_CODE_TSC_INVALID_JSON;
2218
  }
2219 2220 2221

  cJSON *root = cJSON_Parse(payload);
  if (root == NULL) {
X
Xiaoyu Wang 已提交
2222
    uError("SML:0x%" PRIx64 " parse json failed:%s", info->id, payload);
2223 2224
    return TSDB_CODE_TSC_INVALID_JSON;
  }
X
Xiaoyu Wang 已提交
2225
  // multiple data points must be sent in JSON array
2226 2227 2228 2229 2230
  if (cJSON_IsObject(root)) {
    payloadNum = 1;
  } else if (cJSON_IsArray(root)) {
    payloadNum = cJSON_GetArraySize(root);
  } else {
X
Xiaoyu Wang 已提交
2231
    uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2232 2233
    ret = TSDB_CODE_TSC_INVALID_JSON;
    goto end;
wmmhello's avatar
wmmhello 已提交
2234
  }
wmmhello's avatar
wmmhello 已提交
2235

2236 2237 2238
  for (int32_t i = 0; i < payloadNum; ++i) {
    cJSON *dataPoint = (payloadNum == 1 && cJSON_IsObject(root)) ? root : cJSON_GetArrayItem(root, i);
    ret = smlParseTelnetLine(info, dataPoint);
X
Xiaoyu Wang 已提交
2239 2240
    if (ret != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2241 2242 2243 2244 2245 2246 2247
      goto end;
    }
  }

end:
  cJSON_Delete(root);
  return ret;
wmmhello's avatar
wmmhello 已提交
2248
}
2249

X
Xiaoyu Wang 已提交
2250
static int32_t smlInsertData(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
2251 2252
  int32_t code = TSDB_CODE_SUCCESS;

X
Xiaoyu Wang 已提交
2253
  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL);
wmmhello's avatar
wmmhello 已提交
2254
  while (oneTable) {
X
Xiaoyu Wang 已提交
2255
    SSmlTableInfo *tableData = *oneTable;
wmmhello's avatar
wmmhello 已提交
2256 2257 2258 2259

    SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
    strcpy(pName.dbname, info->pRequest->pDb);
    memcpy(pName.tname, tableData->childTableName, strlen(tableData->childTableName));
D
dapan1121 已提交
2260 2261 2262 2263 2264 2265

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

wmmhello's avatar
wmmhello 已提交
2267
    SVgroupInfo vg;
D
dapan1121 已提交
2268
    code = catalogGetTableHashVgroup(info->pCatalog, &conn, &pName, &vg);
2269
    if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2270
      uError("SML:0x%" PRIx64 " catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName);
wmmhello's avatar
wmmhello 已提交
2271 2272
      return code;
    }
X
Xiaoyu Wang 已提交
2273
    taosHashPut(info->pVgHash, (const char *)&vg.vgId, sizeof(vg.vgId), (char *)&vg, sizeof(vg));
wmmhello's avatar
wmmhello 已提交
2274

X
Xiaoyu Wang 已提交
2275 2276 2277
    SSmlSTableMeta **pMeta =
        (SSmlSTableMeta **)taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen);
    ASSERT(NULL != pMeta && NULL != *pMeta);
wmmhello's avatar
wmmhello 已提交
2278

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

2283
    code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat,
wmmhello's avatar
wmmhello 已提交
2284
                       (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen, info->msgBuf.buf, info->msgBuf.len);
X
Xiaoyu Wang 已提交
2285 2286
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
wmmhello's avatar
wmmhello 已提交
2287 2288
      return code;
    }
X
Xiaoyu Wang 已提交
2289
    oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, oneTable);
wmmhello's avatar
wmmhello 已提交
2290
  }
wmmhello's avatar
wmmhello 已提交
2291

2292 2293
  code = smlBuildOutput(info->exec, info->pVgHash);
  if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2294
    uError("SML:0x%" PRIx64 " smlBuildOutput failed", info->id);
2295 2296
    return code;
  }
2297 2298
  info->cost.insertRpcTime = taosGetTimestampUs();

X
Xiaoyu Wang 已提交
2299 2300 2301
  // launchQueryImpl(info->pRequest, info->pQuery, false, NULL);
  //  info->affectedRows = taos_affected_rows(info->pRequest);
  //  return info->pRequest->code;
wmmhello's avatar
wmmhello 已提交
2302

D
dapan1121 已提交
2303
  launchAsyncQuery(info->pRequest, info->pQuery, NULL);
wmmhello's avatar
wmmhello 已提交
2304
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2305 2306
}

X
Xiaoyu Wang 已提交
2307 2308 2309 2310 2311 2312 2313 2314 2315
static void smlPrintStatisticInfo(SSmlHandle *info) {
  uError("SML:0x%" PRIx64
         " smlInsertLines result, code:%d,lineNum:%d,stable num:%d,ctable num:%d,create stable num:%d \
        parse cost:%" PRId64 ",schema cost:%" PRId64 ",bind cost:%" PRId64 ",rpc cost:%" PRId64 ",total cost:%" PRId64
         "",
         info->id, info->cost.code, info->cost.lineNum, info->cost.numOfSTables, info->cost.numOfCTables,
         info->cost.numOfCreateSTables, info->cost.schemaTime - info->cost.parseTime,
         info->cost.insertBindTime - info->cost.schemaTime, info->cost.insertRpcTime - info->cost.insertBindTime,
         info->cost.endTime - info->cost.insertRpcTime, info->cost.endTime - info->cost.parseTime);
2316 2317
}

X
Xiaoyu Wang 已提交
2318
static int32_t smlParseLine(SSmlHandle *info, char *lines[], int numLines) {
wmmhello's avatar
wmmhello 已提交
2319
  int32_t code = TSDB_CODE_SUCCESS;
2320 2321 2322 2323 2324 2325
  if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
    code = smlParseJSON(info, *lines);
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, *lines);
      return code;
    }
wmmhello's avatar
wmmhello 已提交
2326
    return code;
wmmhello's avatar
wmmhello 已提交
2327
  }
wmmhello's avatar
wmmhello 已提交
2328

wmmhello's avatar
wmmhello 已提交
2329
  for (int32_t i = 0; i < numLines; ++i) {
X
Xiaoyu Wang 已提交
2330
    if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
2331
      code = smlParseInfluxLine(info, lines[i]);
X
Xiaoyu Wang 已提交
2332
    } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
2333
      code = smlParseTelnetLine(info, lines[i]);
X
Xiaoyu Wang 已提交
2334
    } else {
2335 2336
      ASSERT(0);
    }
wmmhello's avatar
wmmhello 已提交
2337
    if (code != TSDB_CODE_SUCCESS) {
2338 2339
      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, lines[i]);
      return code;
wmmhello's avatar
wmmhello 已提交
2340 2341
    }
  }
2342 2343 2344
  return code;
}

X
Xiaoyu Wang 已提交
2345
static int smlProcess(SSmlHandle *info, char *lines[], int numLines) {
2346
  int32_t code = TSDB_CODE_SUCCESS;
2347 2348
  int32_t retryNum = 0;

2349 2350 2351 2352
  info->cost.parseTime = taosGetTimestampUs();

  code = smlParseLine(info, lines, numLines);
  if (code != 0) {
X
Xiaoyu Wang 已提交
2353
    uError("SML:0x%" PRIx64 " smlParseLine error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2354
    return code;
2355
  }
wmmhello's avatar
wmmhello 已提交
2356

2357 2358 2359 2360 2361
  info->cost.lineNum = numLines;
  info->cost.numOfSTables = taosHashGetSize(info->superTables);
  info->cost.numOfCTables = taosHashGetSize(info->childTables);

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

X
Xiaoyu Wang 已提交
2363
  do {
2364 2365
    code = smlModifyDBSchemas(info);
    if (code == 0) break;
wmmhello's avatar
wmmhello 已提交
2366
  } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);
2367

wmmhello's avatar
wmmhello 已提交
2368
  if (code != 0) {
X
Xiaoyu Wang 已提交
2369
    uError("SML:0x%" PRIx64 " smlModifyDBSchemas error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2370
    return code;
wmmhello's avatar
wmmhello 已提交
2371
  }
wmmhello's avatar
wmmhello 已提交
2372

2373
  info->cost.insertBindTime = taosGetTimestampUs();
wmmhello's avatar
wmmhello 已提交
2374 2375
  code = smlInsertData(info);
  if (code != 0) {
X
Xiaoyu Wang 已提交
2376
    uError("SML:0x%" PRIx64 " smlInsertData error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2377
    return code;
wmmhello's avatar
wmmhello 已提交
2378 2379 2380 2381 2382
  }

  return code;
}

X
Xiaoyu Wang 已提交
2383
static int32_t isSchemalessDb(STscObj *taos, SRequestObj *request) {
wmmhello's avatar
wmmhello 已提交
2384 2385 2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411
//  SCatalog *catalog = NULL;
//  int32_t   code = catalogGetHandle(((STscObj *)taos)->pAppInfo->clusterId, &catalog);
//  if (code != TSDB_CODE_SUCCESS) {
//    uError("SML get catalog error %d", code);
//    return code;
//  }
//
//  SName name;
//  tNameSetDbName(&name, taos->acctId, taos->db, strlen(taos->db));
//  char dbFname[TSDB_DB_FNAME_LEN] = {0};
//  tNameGetFullDbName(&name, dbFname);
//  SDbCfgInfo pInfo = {0};
//
//  SRequestConnInfo conn = {0};
//  conn.pTrans = taos->pAppInfo->pTransporter;
//  conn.requestId = request->requestId;
//  conn.requestObjRefId = request->self;
//  conn.mgmtEps = getEpSet_s(&taos->pAppInfo->mgmtEp);
//
//  code = catalogGetDBCfg(catalog, &conn, dbFname, &pInfo);
//  if (code != TSDB_CODE_SUCCESS) {
//    return code;
//  }
//  taosArrayDestroy(pInfo.pRetensions);
//
//  if (!pInfo.schemaless) {
//    return TSDB_CODE_SML_INVALID_DB_CONF;
//  }
wmmhello's avatar
wmmhello 已提交
2412 2413 2414
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2415
static void smlInsertCallback(void *param, void *res, int32_t code) {
wmmhello's avatar
wmmhello 已提交
2416
  SRequestObj *pRequest = (SRequestObj *)res;
X
Xiaoyu Wang 已提交
2417
  SSmlHandle  *info = (SSmlHandle *)param;
wmmhello's avatar
wmmhello 已提交
2418
  int32_t rows = taos_affected_rows(pRequest);
wmmhello's avatar
wmmhello 已提交
2419

X
Xiaoyu Wang 已提交
2420
  uDebug("SML:0x%" PRIx64 " result. code:%d, msg:%s", info->id, pRequest->code, pRequest->msgBuf);
wmmhello's avatar
wmmhello 已提交
2421
  // lock
wmmhello's avatar
wmmhello 已提交
2422 2423
  taosThreadSpinLock(&info->params->lock);
  info->params->request->body.resInfo.numOfRows += rows;
X
Xiaoyu Wang 已提交
2424
  if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2425
    info->params->request->code = code;
wmmhello's avatar
wmmhello 已提交
2426
  }
wmmhello's avatar
wmmhello 已提交
2427
  taosThreadSpinUnlock(&info->params->lock);
wmmhello's avatar
wmmhello 已提交
2428 2429
  // unlock

wmmhello's avatar
wmmhello 已提交
2430
  uDebug("SML:0x%" PRIx64 " insert finished, code: %d, rows: %d, total: %d", info->id, code, rows, info->affectedRows);
wmmhello's avatar
wmmhello 已提交
2431
  Params *pParam = info->params;
X
Xiaoyu Wang 已提交
2432
  bool    isLast = info->isLast;
wmmhello's avatar
wmmhello 已提交
2433 2434 2435
  info->cost.endTime = taosGetTimestampUs();
  info->cost.code = code;
  smlPrintStatisticInfo(info);
wmmhello's avatar
wmmhello 已提交
2436 2437
  smlDestroyInfo(info);

X
Xiaoyu Wang 已提交
2438
  if (isLast) {
wmmhello's avatar
wmmhello 已提交
2439
    tsem_post(&pParam->sem);
wmmhello's avatar
wmmhello 已提交
2440 2441 2442
  }
}

wmmhello's avatar
wmmhello 已提交
2443 2444 2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463
/**
 * taos_schemaless_insert() parse and insert data points into database according to
 * different protocol.
 *
 * @param $lines input array may contain multiple lines, each line indicates a data point.
 *               If protocol=2 is used input array should contain single JSON
 *               string(e.g. char *lines[] = {"$JSON_string"}). If need to insert
 *               multiple data points in JSON format, should include them in $JSON_string
 *               as a JSON array.
 * @param $numLines indicates how many data points in $lines.
 *                  If protocol = 2 is used this param will be ignored as $lines should
 *                  contain single JSON string.
 * @param $protocol indicates which protocol to use for parsing:
 *                  0 - influxDB line protocol
 *                  1 - OpenTSDB telnet line protocol
 *                  2 - OpenTSDB JSON format protocol
 * @return return zero for successful insertion. Otherwise return none-zero error code of
 *         failure reason.
 *
 */

wmmhello's avatar
wmmhello 已提交
2464
TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision) {
D
dapan1121 已提交
2465
  if (NULL == taos) {
2466 2467 2468
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }
D
dapan1121 已提交
2469

2470
  SRequestObj* request = (SRequestObj*)createRequest(*(int64_t*)taos, TSDB_SQL_INSERT);
wmmhello's avatar
wmmhello 已提交
2471
  if(!request){
2472
    uError("SML:taos_schemaless_insert error request is null");
2473
    return NULL;
wmmhello's avatar
wmmhello 已提交
2474 2475
  }

wmmhello's avatar
wmmhello 已提交
2476
  int    batchs = 0;
2477 2478
  STscObj* pTscObj = request->pTscObj;

2479
  pTscObj->schemalessType = 1;
wmmhello's avatar
wmmhello 已提交
2480
  SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
wmmhello's avatar
wmmhello 已提交
2481

wmmhello's avatar
wmmhello 已提交
2482 2483
  Params params;
  params.request = request;
wmmhello's avatar
wmmhello 已提交
2484 2485 2486
  tsem_init(&params.sem, 0, 0);
  taosThreadSpinInit(&(params.lock), 0);

X
Xiaoyu Wang 已提交
2487
  if (request->pDb == NULL) {
wmmhello's avatar
wmmhello 已提交
2488
    request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
wmmhello's avatar
wmmhello 已提交
2489
    smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
wmmhello's avatar
wmmhello 已提交
2490 2491 2492
    goto end;
  }

2493
  if(isSchemalessDb(pTscObj, request) != TSDB_CODE_SUCCESS){
wmmhello's avatar
wmmhello 已提交
2494
    request->code = TSDB_CODE_SML_INVALID_DB_CONF;
wmmhello's avatar
wmmhello 已提交
2495
    smlBuildInvalidDataMsg(&msg, "Cannot write data to a non schemaless database", NULL);
wmmhello's avatar
wmmhello 已提交
2496 2497 2498
    goto end;
  }

2499
  if (!lines) {
2500
    request->code = TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
2501
    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
2502 2503 2504
    goto end;
  }

X
Xiaoyu Wang 已提交
2505
  if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
2506
    request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
wmmhello's avatar
wmmhello 已提交
2507
    smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
2508
    goto end;
wmmhello's avatar
wmmhello 已提交
2509 2510
  }

X
Xiaoyu Wang 已提交
2511 2512
  if (protocol == TSDB_SML_LINE_PROTOCOL &&
      (precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)) {
2513
    request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
wmmhello's avatar
wmmhello 已提交
2514
    smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
2515 2516 2517
    goto end;
  }

wmmhello's avatar
wmmhello 已提交
2518 2519 2520 2521 2522 2523 2524 2525
  if(protocol == TSDB_SML_JSON_PROTOCOL){
    numLines = 1;
  }else if(numLines <= 0){
    request->code = TSDB_CODE_SML_INVALID_DATA;
    smlBuildInvalidDataMsg(&msg, "line num is invalid", NULL);
    goto end;
  }

wmmhello's avatar
wmmhello 已提交
2526 2527
  batchs = ceil(((double)numLines) / LINE_BATCH);
  for (int i = 0; i < batchs; ++i) {
2528
    SRequestObj* req = (SRequestObj*)createRequest(pTscObj->id, TSDB_SQL_INSERT);
wmmhello's avatar
wmmhello 已提交
2529 2530 2531 2532 2533
    if(!req){
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error request is null");
      goto end;
    }
2534
    SSmlHandle* info = smlBuildSmlInfo(pTscObj, req, (SMLProtocolType)protocol, precision);
wmmhello's avatar
wmmhello 已提交
2535 2536 2537 2538 2539 2540
    if(!info){
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error SSmlHandle is null");
      goto end;
    }

wmmhello's avatar
wmmhello 已提交
2541 2542
    int32_t perBatch = LINE_BATCH;

X
Xiaoyu Wang 已提交
2543
    if (numLines > perBatch) {
wmmhello's avatar
wmmhello 已提交
2544 2545
      numLines -= perBatch;
      info->isLast = false;
X
Xiaoyu Wang 已提交
2546
    } else {
wmmhello's avatar
wmmhello 已提交
2547 2548 2549 2550 2551
      perBatch = numLines;
      numLines = 0;
      info->isLast = true;
    }

wmmhello's avatar
wmmhello 已提交
2552
    info->params = &params;
wmmhello's avatar
wmmhello 已提交
2553 2554
    info->affectedRows = perBatch;
    info->pRequest->body.queryFp = smlInsertCallback;
X
Xiaoyu Wang 已提交
2555
    info->pRequest->body.param = info;
wmmhello's avatar
wmmhello 已提交
2556
    int32_t code = smlProcess(info, lines, perBatch);
wmmhello's avatar
wmmhello 已提交
2557
    lines += perBatch;
X
Xiaoyu Wang 已提交
2558
    if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2559 2560 2561 2562
      info->pRequest->body.queryFp(info, req, code);
    }
  }
  tsem_wait(&params.sem);
2563 2564

end:
wmmhello's avatar
wmmhello 已提交
2565 2566
  taosThreadSpinDestroy(&params.lock);
  tsem_destroy(&params.sem);
2567
//  ((STscObj *)taos)->schemalessType = 0;
2568
  pTscObj->schemalessType = 1;
2569
  uDebug("resultend:%s", request->msgBuf);
wmmhello's avatar
wmmhello 已提交
2570
  return (TAOS_RES*)request;
wmmhello's avatar
wmmhello 已提交
2571
}