clientSml.c 81.2 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 87 88 89 90 91 92 93 94
//=================================================================================================
typedef TSDB_SML_PROTOCOL_TYPE SMLProtocolType;

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

typedef struct {
X
Xiaoyu Wang 已提交
95 96 97
  char    sTableName[TSDB_TABLE_NAME_LEN];
  SArray *tags;
  SArray *fields;
wmmhello's avatar
wmmhello 已提交
98 99 100
} SCreateSTableActionInfo;

typedef struct {
X
Xiaoyu Wang 已提交
101 102
  char    sTableName[TSDB_TABLE_NAME_LEN];
  SSmlKv *field;
wmmhello's avatar
wmmhello 已提交
103 104 105
} SAlterSTableActionInfo;

typedef struct {
X
Xiaoyu Wang 已提交
106
  ESchemaAction action;
wmmhello's avatar
wmmhello 已提交
107 108
  union {
    SCreateSTableActionInfo createSTable;
109
    SAlterSTableActionInfo  alterSTable;
wmmhello's avatar
wmmhello 已提交
110 111 112 113
  };
} SSchemaAction;

typedef struct {
X
Xiaoyu Wang 已提交
114 115 116 117
  const char *measure;
  const char *tags;
  const char *cols;
  const char *timestamp;
wmmhello's avatar
wmmhello 已提交
118 119 120 121 122 123 124 125 126

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

typedef struct {
X
Xiaoyu Wang 已提交
127 128 129 130
  const char *sTableName;  // super table name
  int32_t     sTableNameLen;
  char        childTableName[TSDB_TABLE_NAME_LEN];
  uint64_t    uid;
wmmhello's avatar
wmmhello 已提交
131

X
Xiaoyu Wang 已提交
132
  SArray *tags;
wmmhello's avatar
wmmhello 已提交
133

134 135
  // 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 已提交
136
  SArray *cols;
wmmhello's avatar
wmmhello 已提交
137 138 139
} SSmlTableInfo;

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

X
Xiaoyu Wang 已提交
143 144
  SArray   *cols;
  SHashObj *colHash;
145

wmmhello's avatar
wmmhello 已提交
146 147 148 149
  STableMeta *tableMeta;
} SSmlSTableMeta;

typedef struct {
X
Xiaoyu Wang 已提交
150 151
  int32_t len;
  char   *buf;
wmmhello's avatar
wmmhello 已提交
152 153
} SSmlMsgBuf;

154 155 156 157 158 159 160 161 162 163 164 165 166 167 168
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 已提交
169 170 171
typedef struct {
  SRequestObj     *request;
  tsem_t           sem;
wmmhello's avatar
wmmhello 已提交
172 173 174
  TdThreadSpinlock lock;
} Params;

wmmhello's avatar
wmmhello 已提交
175
typedef struct {
X
Xiaoyu Wang 已提交
176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198
  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 已提交
199
} SSmlHandle;
wmmhello's avatar
wmmhello 已提交
200 201
//=================================================================================================

wmmhello's avatar
wmmhello 已提交
202
//=================================================================================================
203
static volatile int64_t linesSmlHandleId = 0;
X
Xiaoyu Wang 已提交
204 205
static int64_t          smlGenId() {
           int64_t id;
wmmhello's avatar
wmmhello 已提交
206

X
Xiaoyu Wang 已提交
207 208
           do {
             id = atomic_add_fetch_64(&linesSmlHandleId, 1);
wmmhello's avatar
wmmhello 已提交
209 210
  } while (id == 0);

X
Xiaoyu Wang 已提交
211
           return id;
wmmhello's avatar
wmmhello 已提交
212 213
}

214
static inline bool smlDoubleToInt64OverFlow(double num) {
X
Xiaoyu Wang 已提交
215
  if (num >= (double)INT64_MAX || num <= (double)INT64_MIN) return true;
216 217 218 219 220 221 222 223 224 225 226 227
  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 已提交
228 229 230
static int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2) {
  memset(pBuf->buf, 0, pBuf->len);
  if (msg1) strncat(pBuf->buf, msg1, pBuf->len);
wmmhello's avatar
wmmhello 已提交
231
  int32_t left = pBuf->len - strlen(pBuf->buf);
X
Xiaoyu Wang 已提交
232
  if (left > 2 && msg2) {
wmmhello's avatar
wmmhello 已提交
233 234 235
    strncat(pBuf->buf, ":", left - 1);
    strncat(pBuf->buf, msg2, left - 2);
  }
wmmhello's avatar
wmmhello 已提交
236 237 238
  return TSDB_CODE_SML_INVALID_DATA;
}

X
Xiaoyu Wang 已提交
239 240
static int32_t smlGenerateSchemaAction(SSchema *colField, SHashObj *colHash, SSmlKv *kv, bool isTag,
                                       SSchemaAction *action, bool *actionNeeded, SSmlHandle *info) {
241
  uint16_t *index = (uint16_t *)taosHashGet(colHash, kv->key, kv->keyLen);
242 243
  if (index) {
    if (colField[*index].type != kv->type) {
X
Xiaoyu Wang 已提交
244 245
      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);
246 247 248
      return TSDB_CODE_TSC_INVALID_VALUE;
    }

X
Xiaoyu Wang 已提交
249 250 251 252
    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))) {
253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271
      if (isTag) {
        action->action = SCHEMA_ACTION_CHANGE_TAG_SIZE;
      } else {
        action->action = SCHEMA_ACTION_CHANGE_COLUMN_SIZE;
      }
      action->alterSTable.field = kv;
      *actionNeeded = true;
    }
  } else {
    if (isTag) {
      action->action = SCHEMA_ACTION_ADD_TAG;
    } else {
      action->action = SCHEMA_ACTION_ADD_COLUMN;
    }
    action->alterSTable.field = kv;
    *actionNeeded = true;
  }
  if (*actionNeeded) {
    uDebug("SML:0x%" PRIx64 " generate schema action. column name: %s, action: %d", info->id, colField->name,
X
Xiaoyu Wang 已提交
272
           action->action);
273
  }
wmmhello's avatar
wmmhello 已提交
274 275 276
  return 0;
}

wmmhello's avatar
wmmhello 已提交
277
static int32_t smlFindNearestPowerOf2(int32_t length, uint8_t type) {
wmmhello's avatar
wmmhello 已提交
278
  int32_t result = 1;
X
Xiaoyu Wang 已提交
279
  while (result <= length) {
wmmhello's avatar
wmmhello 已提交
280 281
    result *= 2;
  }
wmmhello's avatar
wmmhello 已提交
282 283 284 285 286
  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 已提交
287 288 289
  return result;
}

X
Xiaoyu Wang 已提交
290
static int32_t smlBuildColumnDescription(SSmlKv *field, char *buf, int32_t bufSize, int32_t *outBytes) {
wmmhello's avatar
wmmhello 已提交
291
  uint8_t type = field->type;
wmmhello's avatar
wmmhello 已提交
292 293
  char    tname[TSDB_TABLE_NAME_LEN] = {0};
  memcpy(tname, field->key, field->keyLen);
wmmhello's avatar
wmmhello 已提交
294
  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
295
    int32_t bytes = smlFindNearestPowerOf2(field->length, type);
X
Xiaoyu Wang 已提交
296
    int     out = snprintf(buf, bufSize, "`%s` %s(%d)", tname, tDataTypes[field->type].name, bytes);
wmmhello's avatar
wmmhello 已提交
297 298
    *outBytes = out;
  } else {
299
    int out = snprintf(buf, bufSize, "`%s` %s", tname, tDataTypes[type].name);
wmmhello's avatar
wmmhello 已提交
300 301 302 303 304 305
    *outBytes = out;
  }

  return 0;
}

X
Xiaoyu Wang 已提交
306
static int32_t smlApplySchemaAction(SSmlHandle *info, SSchemaAction *action) {
wmmhello's avatar
wmmhello 已提交
307 308
  int32_t code = 0;
  int32_t outBytes = 0;
X
Xiaoyu Wang 已提交
309
  char   *result = (char *)taosMemoryCalloc(1, TSDB_MAX_ALLOWED_SQL_LEN);
wmmhello's avatar
wmmhello 已提交
310
  int32_t capacity = TSDB_MAX_ALLOWED_SQL_LEN;
wmmhello's avatar
wmmhello 已提交
311

X
Xiaoyu Wang 已提交
312
  uDebug("SML:0x%" PRIx64 " apply schema action. action: %d", info->id, action->action);
wmmhello's avatar
wmmhello 已提交
313 314
  switch (action->action) {
    case SCHEMA_ACTION_ADD_COLUMN: {
315
      int n = sprintf(result, "alter stable `%s` add column ", action->alterSTable.sTableName);
X
Xiaoyu Wang 已提交
316
      smlBuildColumnDescription(action->alterSTable.field, result + n, capacity - n, &outBytes);
D
dapan1121 已提交
317
      TAOS_RES *res = taos_query((TAOS*)&info->taos->id, result);  // TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
318
      code = taos_errno(res);
X
Xiaoyu Wang 已提交
319
      const char *errStr = taos_errstr(res);
wmmhello's avatar
wmmhello 已提交
320
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
321
        uError("SML:0x%" PRIx64 " apply schema action. error: %s", info->id, errStr);
322
        taosMsleep(100);
wmmhello's avatar
wmmhello 已提交
323 324 325 326 327 328
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_ADD_TAG: {
329
      int n = sprintf(result, "alter stable `%s` add tag ", action->alterSTable.sTableName);
X
Xiaoyu Wang 已提交
330
      smlBuildColumnDescription(action->alterSTable.field, result + n, capacity - n, &outBytes);
D
dapan1121 已提交
331
      TAOS_RES *res = taos_query((TAOS*)&info->taos->id, result);  // TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
332
      code = taos_errno(res);
X
Xiaoyu Wang 已提交
333
      const char *errStr = taos_errstr(res);
wmmhello's avatar
wmmhello 已提交
334
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
335
        uError("SML:0x%" PRIx64 " apply schema action. error : %s", info->id, taos_errstr(res));
336
        taosMsleep(100);
wmmhello's avatar
wmmhello 已提交
337 338 339 340 341 342
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_CHANGE_COLUMN_SIZE: {
343
      int n = sprintf(result, "alter stable `%s` modify column ", action->alterSTable.sTableName);
X
Xiaoyu Wang 已提交
344
      smlBuildColumnDescription(action->alterSTable.field, result + n, capacity - n, &outBytes);
D
dapan1121 已提交
345
      TAOS_RES *res = taos_query((TAOS*)&info->taos->id, result);  // TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
346 347
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
348
        uError("SML:0x%" PRIx64 " apply schema action. error : %s", info->id, taos_errstr(res));
349
        taosMsleep(100);
wmmhello's avatar
wmmhello 已提交
350 351 352 353 354 355
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_CHANGE_TAG_SIZE: {
356
      int n = sprintf(result, "alter stable `%s` modify tag ", action->alterSTable.sTableName);
X
Xiaoyu Wang 已提交
357
      smlBuildColumnDescription(action->alterSTable.field, result + n, capacity - n, &outBytes);
D
dapan1121 已提交
358
      TAOS_RES *res = taos_query((TAOS*)&info->taos->id, result);  // TODO async doAsyncQuery
wmmhello's avatar
wmmhello 已提交
359 360
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
361
        uError("SML:0x%" PRIx64 " apply schema action. error : %s", info->id, taos_errstr(res));
362
        taosMsleep(100);
wmmhello's avatar
wmmhello 已提交
363 364 365 366 367 368
      }
      taos_free_result(res);

      break;
    }
    case SCHEMA_ACTION_CREATE_STABLE: {
X
Xiaoyu Wang 已提交
369 370 371
      int   n = sprintf(result, "create stable `%s` (", action->createSTable.sTableName);
      char *pos = result + n;
      int   freeBytes = capacity - n;
wmmhello's avatar
wmmhello 已提交
372

373
      SArray *cols = action->createSTable.fields;
wmmhello's avatar
wmmhello 已提交
374

X
Xiaoyu Wang 已提交
375
      for (int i = 0; i < taosArrayGetSize(cols); i++) {
wmmhello's avatar
wmmhello 已提交
376
        SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
377
        smlBuildColumnDescription(kv, pos, freeBytes, &outBytes);
X
Xiaoyu Wang 已提交
378 379 380 381 382
        pos += outBytes;
        freeBytes -= outBytes;
        *pos = ',';
        ++pos;
        --freeBytes;
wmmhello's avatar
wmmhello 已提交
383
      }
wmmhello's avatar
wmmhello 已提交
384

X
Xiaoyu Wang 已提交
385 386
      --pos;
      ++freeBytes;
wmmhello's avatar
wmmhello 已提交
387 388

      outBytes = snprintf(pos, freeBytes, ") tags (");
X
Xiaoyu Wang 已提交
389 390
      pos += outBytes;
      freeBytes -= outBytes;
wmmhello's avatar
wmmhello 已提交
391

392
      cols = action->createSTable.tags;
X
Xiaoyu Wang 已提交
393
      for (int i = 0; i < taosArrayGetSize(cols); i++) {
wmmhello's avatar
wmmhello 已提交
394
        SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
395
        smlBuildColumnDescription(kv, pos, freeBytes, &outBytes);
X
Xiaoyu Wang 已提交
396 397 398 399 400
        pos += outBytes;
        freeBytes -= outBytes;
        *pos = ',';
        ++pos;
        --freeBytes;
wmmhello's avatar
wmmhello 已提交
401
      }
X
Xiaoyu Wang 已提交
402 403 404 405 406 407 408
      if (taosArrayGetSize(cols) == 0) {
        outBytes = snprintf(pos, freeBytes, "`%s` %s(%d)", tsSmlTagName, tDataTypes[TSDB_DATA_TYPE_NCHAR].name, 1);
        pos += outBytes;
        freeBytes -= outBytes;
        *pos = ',';
        ++pos;
        --freeBytes;
409
      }
X
Xiaoyu Wang 已提交
410 411
      pos--;
      ++freeBytes;
wmmhello's avatar
wmmhello 已提交
412
      outBytes = snprintf(pos, freeBytes, ")");
D
dapan1121 已提交
413
      TAOS_RES *res = taos_query((TAOS*)&info->taos->id, result);
wmmhello's avatar
wmmhello 已提交
414 415
      code = taos_errno(res);
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
416
        uError("SML:0x%" PRIx64 " apply schema action. error : %s", info->id, taos_errstr(res));
417
        taosMsleep(100);
wmmhello's avatar
wmmhello 已提交
418 419 420 421 422 423 424 425 426 427
      }
      taos_free_result(res);

      break;
    }

    default:
      break;
  }

wmmhello's avatar
wmmhello 已提交
428
  taosMemoryFreeClear(result);
wmmhello's avatar
wmmhello 已提交
429
  if (code != 0) {
X
Xiaoyu Wang 已提交
430
    uError("SML:0x%" PRIx64 " apply schema action failure. %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
431 432 433 434
  }
  return code;
}

X
Xiaoyu Wang 已提交
435 436
static int32_t smlProcessSchemaAction(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
                                      SSchemaAction *action, bool isTag) {
437 438
  int32_t code = TSDB_CODE_SUCCESS;
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
X
Xiaoyu Wang 已提交
439 440
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, j);
    bool    actionNeeded = false;
441
    code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, action, &actionNeeded, info);
X
Xiaoyu Wang 已提交
442
    if (code != TSDB_CODE_SUCCESS) {
443 444 445 446 447 448 449 450 451 452 453 454
      return code;
    }
    if (actionNeeded) {
      code = smlApplySchemaAction(info, action);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
455
static int32_t smlCheckMeta(SSchema *schema, int32_t length, SArray *cols) {
456
  SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
457
  for (uint16_t i = 0; i < length; i++) {
458 459 460
    taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
  }

X
Xiaoyu Wang 已提交
461 462 463
  for (int32_t i = 0; i < taosArrayGetSize(cols); i++) {
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
    if (taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL) {
464 465 466 467 468 469
      return -1;
    }
  }
  return 0;
}

X
Xiaoyu Wang 已提交
470
static int32_t smlModifyDBSchemas(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
471
  int32_t code = 0;
X
Xiaoyu Wang 已提交
472
  SName   pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
473
  strcpy(pName.dbname, info->pRequest->pDb);
wmmhello's avatar
wmmhello 已提交
474

D
dapan1121 已提交
475 476 477 478 479
  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 已提交
480 481

  SSmlSTableMeta **tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, NULL);
wmmhello's avatar
wmmhello 已提交
482
  while (tableMetaSml) {
X
Xiaoyu Wang 已提交
483 484 485
    SSmlSTableMeta *sTableData = *tableMetaSml;
    STableMeta     *pTableMeta = NULL;
    bool            needCheckMeta = false;  // for multi thread
wmmhello's avatar
wmmhello 已提交
486

wmmhello's avatar
wmmhello 已提交
487
    size_t superTableLen = 0;
X
Xiaoyu Wang 已提交
488
    void  *superTable = taosHashGetKey(tableMetaSml, &superTableLen);
489
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
wmmhello's avatar
wmmhello 已提交
490
    memcpy(pName.tname, superTable, superTableLen);
wmmhello's avatar
wmmhello 已提交
491

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

D
dapan1121 已提交
494
    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_INVALID_STB) {
495 496 497
      SSchemaAction schemaAction;
      schemaAction.action = SCHEMA_ACTION_CREATE_STABLE;
      memset(&schemaAction.createSTable, 0, sizeof(SCreateSTableActionInfo));
wmmhello's avatar
wmmhello 已提交
498
      memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
499 500
      schemaAction.createSTable.tags = sTableData->tags;
      schemaAction.createSTable.fields = sTableData->cols;
wmmhello's avatar
wmmhello 已提交
501
      code = smlApplySchemaAction(info, &schemaAction);
502
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
503 504
        uError("SML:0x%" PRIx64 " smlApplySchemaAction failed. can not create %s", info->id,
               schemaAction.createSTable.sTableName);
505
        goto end;
wmmhello's avatar
wmmhello 已提交
506
      }
507
      info->cost.numOfCreateSTables++;
X
Xiaoyu Wang 已提交
508 509 510 511 512
    } 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++) {
513 514
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
wmmhello's avatar
wmmhello 已提交
515

516 517
      SSchemaAction schemaAction;
      memset(&schemaAction, 0, sizeof(SSchemaAction));
518
      memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
519 520 521
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &schemaAction, true);
      if (code != TSDB_CODE_SUCCESS) {
        taosHashCleanup(hashTmp);
522
        goto end;
523 524 525
      }

      taosHashClear(hashTmp);
X
Xiaoyu Wang 已提交
526
      for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns; i++) {
527 528 529 530 531
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
      }
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->cols, &schemaAction, false);
      taosHashCleanup(hashTmp);
      if (code != TSDB_CODE_SUCCESS) {
532
        goto end;
wmmhello's avatar
wmmhello 已提交
533
      }
wmmhello's avatar
wmmhello 已提交
534

D
dapan1121 已提交
535
      code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
wmmhello's avatar
wmmhello 已提交
536
      if (code != TSDB_CODE_SUCCESS) {
537
        goto end;
wmmhello's avatar
wmmhello 已提交
538
      }
539
      needCheckMeta = true;
wmmhello's avatar
wmmhello 已提交
540
    } else {
X
Xiaoyu Wang 已提交
541
      uError("SML:0x%" PRIx64 " load table meta error: %s", info->id, tstrerror(code));
542
      goto end;
wmmhello's avatar
wmmhello 已提交
543
    }
X
Xiaoyu Wang 已提交
544
    if (pTableMeta) taosMemoryFree(pTableMeta);
545

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

X
Xiaoyu Wang 已提交
552 553 554
    if (needCheckMeta) {
      code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags,
                          sTableData->tags);
555
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
556
        uError("SML:0x%" PRIx64 " check tag failed. super table name %s", info->id, (char *)superTable);
557 558 559 560
        goto end;
      }
      code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols);
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
561
        uError("SML:0x%" PRIx64 " check cols failed. super table name %s", info->id, (char *)superTable);
562 563 564 565
        goto end;
      }
    }

566
    sTableData->tableMeta = pTableMeta;
wmmhello's avatar
wmmhello 已提交
567

X
Xiaoyu Wang 已提交
568
    tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, tableMetaSml);
wmmhello's avatar
wmmhello 已提交
569 570
  }
  return 0;
571 572

end:
D
dapan1121 已提交
573
  catalogRefreshTableMeta(info->pCatalog, &conn, &pName, 1);
574
  return code;
wmmhello's avatar
wmmhello 已提交
575 576
}

X
Xiaoyu Wang 已提交
577
static bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
578
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
579 580 581 582
  int32_t     len = kvVal->length;
  char       *endptr = NULL;
  double      result = taosStr2Double(pVal, &endptr);
  if (pVal == endptr) {
583
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
584 585 586
    return false;
  }

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

wmmhello's avatar
wmmhello 已提交
675
static bool smlParseBool(SSmlKv *kvVal) {
wmmhello's avatar
wmmhello 已提交
676
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
677
  int32_t     len = kvVal->length;
678
  if ((len == 1) && (pVal[0] == 't' || pVal[0] == 'T')) {
wmmhello's avatar
wmmhello 已提交
679
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
680 681 682
    return true;
  }

683
  if ((len == 1) && (pVal[0] == 'f' || pVal[0] == 'F')) {
wmmhello's avatar
wmmhello 已提交
684
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
685 686 687
    return true;
  }

X
Xiaoyu Wang 已提交
688
  if ((len == 4) && !strncasecmp(pVal, "true", len)) {
wmmhello's avatar
wmmhello 已提交
689
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
690 691
    return true;
  }
X
Xiaoyu Wang 已提交
692
  if ((len == 5) && !strncasecmp(pVal, "false", len)) {
wmmhello's avatar
wmmhello 已提交
693
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
694 695 696 697 698
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
699
static bool smlIsBinary(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
700
  // binary: "abc"
wmmhello's avatar
wmmhello 已提交
701 702 703 704 705 706 707 708 709
  if (len < 2) {
    return false;
  }
  if (pVal[0] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
710
static bool smlIsNchar(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
711
  // nchar: L"abc"
wmmhello's avatar
wmmhello 已提交
712 713 714
  if (len < 3) {
    return false;
  }
X
Xiaoyu Wang 已提交
715
  if ((pVal[0] == 'l' || pVal[0] == 'L') && pVal[1] == '"' && pVal[len - 1] == '"') {
wmmhello's avatar
wmmhello 已提交
716 717 718 719 720
    return true;
  }
  return false;
}

721
static int64_t smlGetTimeValue(const char *value, int32_t len, int8_t type) {
X
Xiaoyu Wang 已提交
722
  char   *endPtr = NULL;
wafwerar's avatar
wafwerar 已提交
723
  int64_t tsInt64 = taosStr2Int64(value, &endPtr, 10);
X
Xiaoyu Wang 已提交
724
  if (value + len != endPtr) {
725
    return -1;
wmmhello's avatar
wmmhello 已提交
726
  }
727
  double ts = tsInt64;
728 729
  switch (type) {
    case TSDB_TIME_PRECISION_HOURS:
wmmhello's avatar
wmmhello 已提交
730 731
      ts *= NANOSECOND_PER_HOUR;
      tsInt64 *= NANOSECOND_PER_HOUR;
732 733
      break;
    case TSDB_TIME_PRECISION_MINUTES:
wmmhello's avatar
wmmhello 已提交
734 735
      ts *= NANOSECOND_PER_MINUTE;
      tsInt64 *= NANOSECOND_PER_MINUTE;
736 737
      break;
    case TSDB_TIME_PRECISION_SECONDS:
wmmhello's avatar
wmmhello 已提交
738 739
      ts *= NANOSECOND_PER_SEC;
      tsInt64 *= NANOSECOND_PER_SEC;
740 741
      break;
    case TSDB_TIME_PRECISION_MILLI:
wmmhello's avatar
wmmhello 已提交
742 743
      ts *= NANOSECOND_PER_MSEC;
      tsInt64 *= NANOSECOND_PER_MSEC;
744 745
      break;
    case TSDB_TIME_PRECISION_MICRO:
wmmhello's avatar
wmmhello 已提交
746 747
      ts *= NANOSECOND_PER_USEC;
      tsInt64 *= NANOSECOND_PER_USEC;
748 749 750 751 752
      break;
    case TSDB_TIME_PRECISION_NANO:
      break;
    default:
      ASSERT(0);
wmmhello's avatar
wmmhello 已提交
753
  }
X
Xiaoyu Wang 已提交
754
  if (ts >= (double)INT64_MAX || ts < 0) {
755
    return -1;
wmmhello's avatar
wmmhello 已提交
756 757
  }

758
  return tsInt64;
759 760 761 762 763 764
}

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) {
765
    return TSDB_TIME_PRECISION_MILLI;
766 767
  } else {
    return -1;
wmmhello's avatar
wmmhello 已提交
768
  }
769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787
}

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 已提交
788
  }
789 790
}

X
Xiaoyu Wang 已提交
791 792
static int64_t smlParseInfluxTime(SSmlHandle *info, const char *data, int32_t len) {
  if (len == 0 || (len == 1 && data[0] == '0')) {
793
    return taosGetTimestampNs();
wmmhello's avatar
wmmhello 已提交
794 795
  }

796 797 798 799
  int8_t tsType = smlGetTsTypeByPrecision(info->precision);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp precision", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
800
  }
801 802

  int64_t ts = smlGetTimeValue(data, len, tsType);
X
Xiaoyu Wang 已提交
803
  if (ts == -1) {
804 805
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
wmmhello's avatar
wmmhello 已提交
806
  }
807 808 809
  return ts;
}

X
Xiaoyu Wang 已提交
810 811
static int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len) {
  if (!data) {
812 813
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp can not be null", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
814
  }
X
Xiaoyu Wang 已提交
815
  if (len == 1 && data[0] == '0') {
816 817
    return taosGetTimestampNs();
  }
818 819
  int8_t tsType = smlGetTsTypeByLen(len);
  if (tsType == -1) {
X
Xiaoyu Wang 已提交
820 821
    smlBuildInvalidDataMsg(&info->msgBuf,
                           "timestamp precision can only be seconds(10 digits) or milli seconds(13 digits)", data);
822 823 824
    return -1;
  }
  int64_t ts = smlGetTimeValue(data, len, tsType);
X
Xiaoyu Wang 已提交
825
  if (ts == -1) {
826 827 828 829 830 831
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
  }
  return ts;
}

X
Xiaoyu Wang 已提交
832
static int32_t smlParseTS(SSmlHandle *info, const char *data, int32_t len, SArray *cols) {
833
  int64_t ts = 0;
X
Xiaoyu Wang 已提交
834
  if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
835
    ts = smlParseInfluxTime(info, data, len);
X
Xiaoyu Wang 已提交
836
  } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
837
    ts = smlParseOpenTsdbTime(info, data, len);
X
Xiaoyu Wang 已提交
838
  } else {
839
    ASSERT(0);
840
  }
841

wmmhello's avatar
wmmhello 已提交
842
  if (ts == -1) return TSDB_CODE_INVALID_TIMESTAMP;
843 844 845

  // add ts to
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
846
  if (!kv) {
847 848 849 850 851 852 853 854
    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 已提交
855
  if (cols) taosArrayPush(cols, &kv);
856 857 858
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
859
static int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
X
Xiaoyu Wang 已提交
860
  // binary
wmmhello's avatar
wmmhello 已提交
861
  if (smlIsBinary(pVal->value, pVal->length)) {
862
    pVal->type = TSDB_DATA_TYPE_BINARY;
wmmhello's avatar
wmmhello 已提交
863
    pVal->length -= BINARY_ADD_LEN;
wmmhello's avatar
wmmhello 已提交
864 865 866
    if (pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
867
    pVal->value += (BINARY_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
868
    return TSDB_CODE_SUCCESS;
869
  }
X
Xiaoyu Wang 已提交
870
  // nchar
wmmhello's avatar
wmmhello 已提交
871
  if (smlIsNchar(pVal->value, pVal->length)) {
872
    pVal->type = TSDB_DATA_TYPE_NCHAR;
wmmhello's avatar
wmmhello 已提交
873
    pVal->length -= NCHAR_ADD_LEN;
wmmhello's avatar
wmmhello 已提交
874 875 876
    if(pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
877
    pVal->value += (NCHAR_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
878
    return TSDB_CODE_SUCCESS;
879 880
  }

X
Xiaoyu Wang 已提交
881
  // bool
882 883 884
  if (smlParseBool(pVal)) {
    pVal->type = TSDB_DATA_TYPE_BOOL;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
885
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
886
  }
X
Xiaoyu Wang 已提交
887
  // number
888
  if (smlParseNumber(pVal, msg)) {
wmmhello's avatar
wmmhello 已提交
889
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
890
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
891 892
  }

wmmhello's avatar
wmmhello 已提交
893
  return TSDB_CODE_TSC_INVALID_VALUE;
wmmhello's avatar
wmmhello 已提交
894 895
}

X
Xiaoyu Wang 已提交
896 897
static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
898
  JUMP_SPACE(sql)
X
Xiaoyu Wang 已提交
899
  if (*sql == COMMA) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
900
  elements->measure = sql;
wmmhello's avatar
wmmhello 已提交
901

wmmhello's avatar
wmmhello 已提交
902
  // parse measure
wmmhello's avatar
wmmhello 已提交
903
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
904 905
    if ((sql != elements->measure) && IS_SLASH_LETTER(sql)) {
      MOVE_FORWARD_ONE(sql, strlen(sql) + 1);
wmmhello's avatar
wmmhello 已提交
906 907
      continue;
    }
X
Xiaoyu Wang 已提交
908
    if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
909 910 911
      break;
    }

X
Xiaoyu Wang 已提交
912
    if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
913 914
      break;
    }
wmmhello's avatar
wmmhello 已提交
915 916
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
917
  elements->measureLen = sql - elements->measure;
X
Xiaoyu Wang 已提交
918
  if (IS_INVALID_TABLE_LEN(elements->measureLen)) {
919
    smlBuildInvalidDataMsg(msg, "measure is empty or too large than 192", NULL);
wmmhello's avatar
wmmhello 已提交
920
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
wmmhello's avatar
wmmhello 已提交
921
  }
wmmhello's avatar
wmmhello 已提交
922

wmmhello's avatar
wmmhello 已提交
923
  // parse tag
X
Xiaoyu Wang 已提交
924
  if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
925
    elements->tagsLen = 0;
X
Xiaoyu Wang 已提交
926 927
  } else {
    if (*sql == COMMA) sql++;
wmmhello's avatar
wmmhello 已提交
928 929
    elements->tags = sql;
    while (*sql != '\0') {
X
Xiaoyu Wang 已提交
930
      if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
931 932 933
        break;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
934
    }
wmmhello's avatar
wmmhello 已提交
935
    elements->tagsLen = sql - elements->tags;
936
  }
wmmhello's avatar
wmmhello 已提交
937
  elements->measureTagsLen = sql - elements->measure;
wmmhello's avatar
wmmhello 已提交
938

wmmhello's avatar
wmmhello 已提交
939 940 941
  // parse cols
  JUMP_SPACE(sql)
  elements->cols = sql;
wmmhello's avatar
wmmhello 已提交
942
  bool isInQuote = false;
wmmhello's avatar
wmmhello 已提交
943
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
944
    if (IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
945 946
      isInQuote = !isInQuote;
    }
X
Xiaoyu Wang 已提交
947
    if (!isInQuote && IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
948 949 950 951
      break;
    }
    sql++;
  }
X
Xiaoyu Wang 已提交
952
  if (isInQuote) {
953 954 955
    smlBuildInvalidDataMsg(msg, "only one quote", elements->cols);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
956
  elements->colsLen = sql - elements->cols;
X
Xiaoyu Wang 已提交
957
  if (elements->colsLen == 0) {
wmmhello's avatar
wmmhello 已提交
958 959 960
    smlBuildInvalidDataMsg(msg, "cols is empty", NULL);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
961

wmmhello's avatar
wmmhello 已提交
962 963 964
  // parse timestamp
  JUMP_SPACE(sql)
  elements->timestamp = sql;
wmmhello's avatar
wmmhello 已提交
965
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
966
    if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
967 968 969 970
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
971
  elements->timestampLen = sql - elements->timestamp;
wmmhello's avatar
wmmhello 已提交
972 973 974 975

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
976
static void smlParseTelnetElement(const char **sql, const char **data, int32_t *len) {
977
  while (**sql != '\0') {
X
Xiaoyu Wang 已提交
978
    if (**sql != SPACE && !(*data)) {
979
      *data = *sql;
X
Xiaoyu Wang 已提交
980
    } else if (**sql == SPACE && *data) {
981 982 983 984 985 986 987
      *len = *sql - *data;
      break;
    }
    (*sql)++;
  }
}

X
Xiaoyu Wang 已提交
988 989
static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTableName, SHashObj *dumplicateKey,
                                  SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
990
  const char *sql = data;
X
Xiaoyu Wang 已提交
991 992
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
  while (*sql != '\0') {
wmmhello's avatar
wmmhello 已提交
993
    JUMP_SPACE(sql)
X
Xiaoyu Wang 已提交
994
    if (*sql == '\0') break;
wmmhello's avatar
wmmhello 已提交
995

wmmhello's avatar
wmmhello 已提交
996
    const char *key = sql;
X
Xiaoyu Wang 已提交
997
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
998 999

    // parse key
X
Xiaoyu Wang 已提交
1000 1001
    while (*sql != '\0') {
      if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
1002 1003 1004
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
1005
      if (*sql == EQUAL) {
wmmhello's avatar
wmmhello 已提交
1006 1007
        keyLen = sql - key;
        sql++;
1008 1009
        break;
      }
wmmhello's avatar
wmmhello 已提交
1010
      sql++;
1011
    }
wmmhello's avatar
wmmhello 已提交
1012

X
Xiaoyu Wang 已提交
1013
    if (IS_INVALID_COL_LEN(keyLen)) {
1014
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1015
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
1016
    }
X
Xiaoyu Wang 已提交
1017
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1018
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1019
      return TSDB_CODE_TSC_DUP_NAMES;
1020 1021 1022
    }

    // parse value
wmmhello's avatar
wmmhello 已提交
1023
    const char *value = sql;
X
Xiaoyu Wang 已提交
1024 1025
    int32_t     valueLen = 0;
    while (*sql != '\0') {
wmmhello's avatar
wmmhello 已提交
1026 1027
      // parse value
      if (*sql == SPACE) {
1028 1029
        break;
      }
wmmhello's avatar
wmmhello 已提交
1030 1031 1032 1033 1034
      if (*sql == EQUAL) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
1035
    }
wmmhello's avatar
wmmhello 已提交
1036
    valueLen = sql - value;
wmmhello's avatar
wmmhello 已提交
1037

X
Xiaoyu Wang 已提交
1038
    if (valueLen == 0) {
1039
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1040
      return TSDB_CODE_TSC_INVALID_VALUE;
1041
    }
wmmhello's avatar
wmmhello 已提交
1042

X
Xiaoyu Wang 已提交
1043 1044
    // handle child table name
    if (childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1045 1046 1047 1048 1049
      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 已提交
1050 1051 1052 1053
    if(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }

1054 1055
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1056
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1057 1058 1059
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1060
    kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1061
    kv->type = TSDB_DATA_TYPE_NCHAR;
1062

X
Xiaoyu Wang 已提交
1063
    if (cols) taosArrayPush(cols, &kv);
1064 1065 1066 1067
  }

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

1069
// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
X
Xiaoyu Wang 已提交
1070 1071
static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTableInfo *tinfo, SArray *cols) {
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
1072 1073 1074

  // parse metric
  smlParseTelnetElement(&sql, &tinfo->sTableName, &tinfo->sTableNameLen);
1075
  if (!(tinfo->sTableName) || IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
1076
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1077
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
1078 1079 1080 1081
  }

  // parse timestamp
  const char *timestamp = NULL;
X
Xiaoyu Wang 已提交
1082
  int32_t     tLen = 0;
1083 1084 1085 1086 1087 1088 1089 1090 1091
  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 已提交
1092
    return ret;
1093 1094 1095 1096
  }

  // parse value
  const char *value = NULL;
X
Xiaoyu Wang 已提交
1097
  int32_t     valueLen = 0;
1098 1099 1100
  smlParseTelnetElement(&sql, &value, &valueLen);
  if (!value || valueLen == 0) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
wmmhello's avatar
wmmhello 已提交
1101
    return TSDB_CODE_TSC_INVALID_VALUE;
1102 1103 1104
  }

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1105
  if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1106 1107 1108 1109
  taosArrayPush(cols, &kv);
  kv->key = VALUE;
  kv->keyLen = VALUE_LEN;
  kv->value = value;
wmmhello's avatar
wmmhello 已提交
1110
  kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1111 1112
  if ((ret = smlParseValue(kv, &info->msgBuf)) != TSDB_CODE_SUCCESS) {
    return ret;
1113 1114 1115
  }

  // parse tags
1116
  ret = smlParseTelnetTags(sql, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
1117 1118
  if (ret != TSDB_CODE_SUCCESS) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1119
    return ret;
1120 1121 1122 1123 1124
  }

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1125 1126 1127
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 已提交
1128
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1129 1130
  }

X
Xiaoyu Wang 已提交
1131
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
wmmhello's avatar
wmmhello 已提交
1132
  const char *sql = data;
X
Xiaoyu Wang 已提交
1133
  while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1134
    const char *key = sql;
X
Xiaoyu Wang 已提交
1135
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
1136

X
Xiaoyu Wang 已提交
1137
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1138
      // parse key
X
Xiaoyu Wang 已提交
1139
      if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1140 1141 1142
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
1143
      if (IS_EQUAL(sql)) {
wmmhello's avatar
wmmhello 已提交
1144 1145
        keyLen = sql - key;
        sql++;
wmmhello's avatar
wmmhello 已提交
1146 1147
        break;
      }
wmmhello's avatar
wmmhello 已提交
1148
      sql++;
wmmhello's avatar
wmmhello 已提交
1149
    }
wmmhello's avatar
wmmhello 已提交
1150

X
Xiaoyu Wang 已提交
1151
    if (IS_INVALID_COL_LEN(keyLen)) {
wmmhello's avatar
wmmhello 已提交
1152
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1153
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
wmmhello's avatar
wmmhello 已提交
1154
    }
X
Xiaoyu Wang 已提交
1155
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1156
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1157
      return TSDB_CODE_TSC_DUP_NAMES;
1158 1159
    }

wmmhello's avatar
wmmhello 已提交
1160
    // parse value
wmmhello's avatar
wmmhello 已提交
1161
    const char *value = sql;
X
Xiaoyu Wang 已提交
1162 1163 1164
    int32_t     valueLen = 0;
    bool        isInQuote = false;
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1165
      // parse value
X
Xiaoyu Wang 已提交
1166
      if (!isTag && IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
1167
        isInQuote = !isInQuote;
wmmhello's avatar
wmmhello 已提交
1168 1169
        sql++;
        continue;
wmmhello's avatar
wmmhello 已提交
1170
      }
wmmhello's avatar
wmmhello 已提交
1171
      if (!isInQuote && IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1172 1173
        break;
      }
wmmhello's avatar
wmmhello 已提交
1174 1175 1176 1177 1178
      if (!isInQuote && IS_EQUAL(sql)) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
1179
    }
wmmhello's avatar
wmmhello 已提交
1180 1181 1182
    valueLen = sql - value;
    sql++;

X
Xiaoyu Wang 已提交
1183
    if (isInQuote) {
wmmhello's avatar
wmmhello 已提交
1184 1185 1186
      smlBuildInvalidDataMsg(msg, "only one quote", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
X
Xiaoyu Wang 已提交
1187
    if (valueLen == 0) {
wmmhello's avatar
wmmhello 已提交
1188
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1189 1190
      return TSDB_CODE_SML_INVALID_DATA;
    }
wmmhello's avatar
wmmhello 已提交
1191 1192
    PROCESS_SLASH(key, keyLen)
    PROCESS_SLASH(value, valueLen)
wmmhello's avatar
wmmhello 已提交
1193

X
Xiaoyu Wang 已提交
1194 1195
    // handle child table name
    if (childTableName && childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1196 1197 1198 1199 1200
      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 已提交
1201
    // add kv to SSmlKv
wmmhello's avatar
wmmhello 已提交
1202
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1203 1204
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if (cols) taosArrayPush(cols, &kv);
1205

wmmhello's avatar
wmmhello 已提交
1206 1207 1208
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1209
    kv->length = valueLen;
X
Xiaoyu Wang 已提交
1210
    if (isTag) {
wmmhello's avatar
wmmhello 已提交
1211
      kv->type = TSDB_DATA_TYPE_NCHAR;
X
Xiaoyu Wang 已提交
1212
    } else {
wmmhello's avatar
wmmhello 已提交
1213 1214 1215
      int32_t ret = smlParseValue(kv, msg);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
wmmhello's avatar
wmmhello 已提交
1216
      }
wmmhello's avatar
wmmhello 已提交
1217 1218
    }
  }
wmmhello's avatar
wmmhello 已提交
1219

wmmhello's avatar
wmmhello 已提交
1220 1221 1222
  return TSDB_CODE_SUCCESS;
}

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

wmmhello's avatar
wmmhello 已提交
1227
    int16_t *index = (int16_t *)taosHashGet(metaHash, kv->key, kv->keyLen);
X
Xiaoyu Wang 已提交
1228
    if (index) {
wmmhello's avatar
wmmhello 已提交
1229
      SSmlKv **value = (SSmlKv **)taosArrayGet(metaArray, *index);
X
Xiaoyu Wang 已提交
1230
      if (kv->type != (*value)->type) {
wmmhello's avatar
wmmhello 已提交
1231
        smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
wmmhello's avatar
wmmhello 已提交
1232
        return TSDB_CODE_SML_NOT_SAME_TYPE;
X
Xiaoyu Wang 已提交
1233 1234 1235
      } else {
        if (IS_VAR_DATA_TYPE(kv->type)) {  // update string len, if bigger
          if (kv->length > (*value)->length) {
wmmhello's avatar
wmmhello 已提交
1236
            *value = kv;
1237 1238 1239
          }
        }
      }
X
Xiaoyu Wang 已提交
1240
    } else {
wmmhello's avatar
wmmhello 已提交
1241 1242 1243 1244 1245
      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);
1246
    }
wmmhello's avatar
wmmhello 已提交
1247
  }
wmmhello's avatar
wmmhello 已提交
1248

wmmhello's avatar
wmmhello 已提交
1249
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1250 1251
}

X
Xiaoyu Wang 已提交
1252
static void smlInsertMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols) {
wmmhello's avatar
wmmhello 已提交
1253 1254 1255 1256
  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 已提交
1257 1258 1259
  }
}

X
Xiaoyu Wang 已提交
1260
static SSmlTableInfo *smlBuildTableInfo() {
1261
  SSmlTableInfo *tag = (SSmlTableInfo *)taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
X
Xiaoyu Wang 已提交
1262
  if (!tag) {
1263
    return NULL;
wmmhello's avatar
wmmhello 已提交
1264
  }
1265 1266 1267 1268 1269

  tag->cols = taosArrayInit(16, POINTER_BYTES);
  if (tag->cols == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1270
  }
1271 1272 1273 1274 1275

  tag->tags = taosArrayInit(16, POINTER_BYTES);
  if (tag->tags == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1276
  }
1277 1278 1279 1280 1281
  return tag;

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

X
Xiaoyu Wang 已提交
1284 1285 1286
static void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag) {
  if (info->dataFormat) {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1287 1288
      SArray *kvArray = (SArray *)taosArrayGetP(tag->cols, i);
      for (int j = 0; j < taosArrayGetSize(kvArray); ++j) {
wmmhello's avatar
wmmhello 已提交
1289
        SSmlKv *p = (SSmlKv *)taosArrayGetP(kvArray, j);
X
Xiaoyu Wang 已提交
1290 1291 1292
        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 已提交
1293
        }
1294 1295 1296 1297
        taosMemoryFree(p);
      }
      taosArrayDestroy(kvArray);
    }
X
Xiaoyu Wang 已提交
1298 1299
  } else {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1300
      SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
X
Xiaoyu Wang 已提交
1301
      void    **p1 = (void **)taosHashIterate(kvHash, NULL);
1302 1303
      while (p1) {
        taosMemoryFree(*p1);
X
Xiaoyu Wang 已提交
1304
        p1 = (void **)taosHashIterate(kvHash, p1);
1305 1306 1307 1308
      }
      taosHashCleanup(kvHash);
    }
  }
X
Xiaoyu Wang 已提交
1309
  for (size_t i = 0; i < taosArrayGetSize(tag->tags); i++) {
wmmhello's avatar
wmmhello 已提交
1310
    SSmlKv *p = (SSmlKv *)taosArrayGetP(tag->tags, i);
X
Xiaoyu Wang 已提交
1311 1312 1313 1314
    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 已提交
1315 1316 1317 1318
      }
    }
    taosMemoryFree(p);
  }
X
Xiaoyu Wang 已提交
1319 1320
  if (info->protocol == TSDB_SML_JSON_PROTOCOL && tag->sTableName) {
    taosMemoryFree((void *)tag->sTableName);
wmmhello's avatar
wmmhello 已提交
1321
  }
1322 1323 1324 1325 1326
  taosArrayDestroy(tag->cols);
  taosArrayDestroy(tag->tags);
  taosMemoryFree(tag);
}

X
Xiaoyu Wang 已提交
1327
static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) {
1328 1329
  SArray *s1 = *(SArray **)key1;
  SArray *s2 = *(SArray **)key2;
1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342
  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 已提交
1343
static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
1344 1345
  SHashObj *s1 = *(SHashObj **)key1;
  SHashObj *s2 = *(SHashObj **)key2;
1346 1347
  SSmlKv *kv1 = *(SSmlKv **)taosHashGet(s1, TS, TS_LEN);
  SSmlKv *kv2 = *(SSmlKv **)taosHashGet(s2, TS, TS_LEN);
1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358
  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;
  }
}

1359 1360
static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
  if(dataFormat){
1361
    void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GT);
1362 1363
    if(p == NULL){
      taosArrayPush(oneTable->cols, &cols);
1364 1365
    }else{
      taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
1366
    }
1367 1368 1369 1370
    return TSDB_CODE_SUCCESS;
  }

  SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1371
  if (!kvHash) {
1372 1373 1374
    uError("SML:smlDealCols failed to allocate memory");
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
1375
  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
1376
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
wmmhello's avatar
wmmhello 已提交
1377
    taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
1378 1379
  }

1380
  void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GT);
1381 1382
  if(p == NULL){
    taosArrayPush(oneTable->cols, &kvHash);
1383 1384
  }else{
    taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &kvHash);
1385
  }
1386 1387 1388
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1389 1390 1391
static SSmlSTableMeta *smlBuildSTableMeta() {
  SSmlSTableMeta *meta = (SSmlSTableMeta *)taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if (!meta) {
1392 1393 1394 1395 1396 1397 1398 1399
    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 已提交
1400 1401
  meta->colHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (meta->colHash == NULL) {
1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423
    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 已提交
1424
static void smlDestroySTableMeta(SSmlSTableMeta *meta) {
1425
  taosHashCleanup(meta->tagHash);
wmmhello's avatar
wmmhello 已提交
1426
  taosHashCleanup(meta->colHash);
1427 1428 1429
  taosArrayDestroy(meta->tags);
  taosArrayDestroy(meta->cols);
  taosMemoryFree(meta->tableMeta);
wmmhello's avatar
wmmhello 已提交
1430
  taosMemoryFree(meta);
1431 1432 1433 1434 1435
}

static void smlDestroyCols(SArray *cols) {
  if (!cols) return;
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
wmmhello's avatar
wmmhello 已提交
1436
    void *kv = taosArrayGetP(cols, i);
1437 1438 1439 1440
    taosMemoryFree(kv);
  }
}

X
Xiaoyu Wang 已提交
1441 1442
static void smlDestroyInfo(SSmlHandle *info) {
  if (!info) return;
1443 1444 1445 1446
  qDestroyQuery(info->pQuery);
  smlDestroyHandle(info->exec);

  // destroy info->childTables
X
Xiaoyu Wang 已提交
1447
  void **p1 = (void **)taosHashIterate(info->childTables, NULL);
1448
  while (p1) {
X
Xiaoyu Wang 已提交
1449 1450
    smlDestroyTableInfo(info, (SSmlTableInfo *)(*p1));
    p1 = (void **)taosHashIterate(info->childTables, p1);
1451 1452 1453 1454
  }
  taosHashCleanup(info->childTables);

  // destroy info->superTables
X
Xiaoyu Wang 已提交
1455
  p1 = (void **)taosHashIterate(info->superTables, NULL);
1456
  while (p1) {
X
Xiaoyu Wang 已提交
1457 1458
    smlDestroySTableMeta((SSmlSTableMeta *)(*p1));
    p1 = (void **)taosHashIterate(info->superTables, p1);
1459 1460 1461 1462 1463 1464
  }
  taosHashCleanup(info->superTables);

  // destroy info->pVgHash
  taosHashCleanup(info->pVgHash);
  taosHashCleanup(info->dumplicateKey);
X
Xiaoyu Wang 已提交
1465
  if (!info->dataFormat) {
wmmhello's avatar
wmmhello 已提交
1466 1467
    taosArrayDestroy(info->colsContainer);
  }
wmmhello's avatar
wmmhello 已提交
1468
  destroyRequest(info->pRequest);
1469 1470 1471
  taosMemoryFreeClear(info);
}

1472
static SSmlHandle* smlBuildSmlInfo(STscObj* pTscObj, SRequestObj* request, SMLProtocolType protocol, int8_t precision){
1473 1474 1475 1476 1477
  int32_t code = TSDB_CODE_SUCCESS;
  SSmlHandle* info = (SSmlHandle*)taosMemoryCalloc(1, sizeof(SSmlHandle));
  if (NULL == info) {
    return NULL;
  }
X
Xiaoyu Wang 已提交
1478
  info->id = smlGenId();
1479

X
Xiaoyu Wang 已提交
1480
  info->pQuery = (SQuery *)taosMemoryCalloc(1, sizeof(SQuery));
1481
  if (NULL == info->pQuery) {
X
Xiaoyu Wang 已提交
1482
    uError("SML:0x%" PRIx64 " create info->pQuery error", info->id);
1483 1484
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
1485
  info->pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
1486
  info->pQuery->haveResultSet = false;
X
Xiaoyu Wang 已提交
1487 1488 1489 1490
  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);
1491 1492
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
1493
  ((SVnodeModifOpStmt *)(info->pQuery->pRoot))->payloadType = PAYLOAD_TYPE_KV;
1494

1495
  info->taos        = pTscObj;
wmmhello's avatar
wmmhello 已提交
1496
  code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
X
Xiaoyu Wang 已提交
1497 1498
  if (code != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " get catalog error %d", info->id, code);
wmmhello's avatar
wmmhello 已提交
1499 1500
    goto cleanup;
  }
1501

X
Xiaoyu Wang 已提交
1502 1503 1504
  info->precision = precision;
  info->protocol = protocol;
  if (protocol == TSDB_SML_LINE_PROTOCOL) {
1505
    info->dataFormat = tsSmlDataFormat;
X
Xiaoyu Wang 已提交
1506
  } else {
1507 1508
    info->dataFormat = true;
  }
X
Xiaoyu Wang 已提交
1509 1510 1511
  info->pRequest = request;
  info->msgBuf.buf = info->pRequest->msgBuf;
  info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
1512

X
Xiaoyu Wang 已提交
1513
  info->exec = smlInitHandle(info->pQuery);
1514 1515
  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 已提交
1516
  info->pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
1517 1518

  info->dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1519
  if (!info->dataFormat) {
1520
    info->colsContainer = taosArrayInit(32, POINTER_BYTES);
X
Xiaoyu Wang 已提交
1521 1522
    if (NULL == info->colsContainer) {
      uError("SML:0x%" PRIx64 " create info failed", info->id);
1523 1524 1525
      goto cleanup;
    }
  }
X
Xiaoyu Wang 已提交
1526 1527 1528
  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);
1529 1530 1531 1532 1533 1534 1535 1536 1537 1538
    goto cleanup;
  }

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

/************* TSDB_SML_JSON_PROTOCOL function start **************/
X
Xiaoyu Wang 已提交
1539
static int32_t smlJsonCreateSring(const char **output, char *input, int32_t inputLen) {
1540
  *output = (const char *)taosMemoryMalloc(inputLen);
X
Xiaoyu Wang 已提交
1541
  if (*output == NULL) {
1542 1543 1544
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

X
Xiaoyu Wang 已提交
1545
  memcpy((void *)(*output), input, inputLen);
1546 1547 1548 1549 1550 1551
  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 已提交
1552
    return TSDB_CODE_TSC_INVALID_JSON;
1553 1554 1555
  }

  tinfo->sTableNameLen = strlen(metric->valuestring);
1556
  if (IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
X
Xiaoyu Wang 已提交
1557
    uError("OTD:0x%" PRIx64 " Metric lenght is 0 or large than 192", info->id);
1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580
    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 已提交
1581
  if (smlDoubleToInt64OverFlow(timeDouble)) {
1582
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1583
    return TSDB_CODE_INVALID_TIMESTAMP;
1584
  }
wmmhello's avatar
wmmhello 已提交
1585 1586 1587 1588 1589 1590 1591 1592

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

  if (timeDouble < 0) {
    return TSDB_CODE_INVALID_TIMESTAMP;
1593 1594
  }

1595
  *tsVal = timeDouble;
1596
  size_t typeLen = strlen(type->valuestring);
wmmhello's avatar
wmmhello 已提交
1597
  if (typeLen == 1 && (type->valuestring[0] == 's' || type->valuestring[0] == 'S')) {
X
Xiaoyu Wang 已提交
1598
    // seconds
1599 1600
    *tsVal = *tsVal * NANOSECOND_PER_SEC;
    timeDouble = timeDouble * NANOSECOND_PER_SEC;
X
Xiaoyu Wang 已提交
1601
    if (smlDoubleToInt64OverFlow(timeDouble)) {
1602
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1603
      return TSDB_CODE_INVALID_TIMESTAMP;
1604
    }
wmmhello's avatar
wmmhello 已提交
1605
  } else if (typeLen == 2 && (type->valuestring[1] == 's' || type->valuestring[1] == 'S')) {
1606 1607
    switch (type->valuestring[0]) {
      case 'm':
wmmhello's avatar
wmmhello 已提交
1608
      case 'M':
X
Xiaoyu Wang 已提交
1609
        // milliseconds
1610 1611
        *tsVal = *tsVal * NANOSECOND_PER_MSEC;
        timeDouble = timeDouble * NANOSECOND_PER_MSEC;
X
Xiaoyu Wang 已提交
1612
        if (smlDoubleToInt64OverFlow(timeDouble)) {
1613
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1614
          return TSDB_CODE_INVALID_TIMESTAMP;
1615 1616 1617
        }
        break;
      case 'u':
wmmhello's avatar
wmmhello 已提交
1618
      case 'U':
X
Xiaoyu Wang 已提交
1619
        // microseconds
1620 1621
        *tsVal = *tsVal * NANOSECOND_PER_USEC;
        timeDouble = timeDouble * NANOSECOND_PER_USEC;
X
Xiaoyu Wang 已提交
1622
        if (smlDoubleToInt64OverFlow(timeDouble)) {
1623
          smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1624
          return TSDB_CODE_INVALID_TIMESTAMP;
1625 1626 1627
        }
        break;
      case 'n':
wmmhello's avatar
wmmhello 已提交
1628
      case 'N':
1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649
        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 已提交
1650
  // Timestamp must be the first KV to parse
1651 1652 1653 1654
  int64_t tsVal = 0;

  cJSON *timestamp = cJSON_GetObjectItem(root, "timestamp");
  if (cJSON_IsNumber(timestamp)) {
X
Xiaoyu Wang 已提交
1655
    // timestamp value 0 indicates current system time
1656
    double timeDouble = timestamp->valuedouble;
X
Xiaoyu Wang 已提交
1657
    if (smlDoubleToInt64OverFlow(timeDouble)) {
1658
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1659
      return TSDB_CODE_INVALID_TIMESTAMP;
1660
    }
wmmhello's avatar
wmmhello 已提交
1661

X
Xiaoyu Wang 已提交
1662
    if (timeDouble < 0) {
wmmhello's avatar
wmmhello 已提交
1663
      return TSDB_CODE_INVALID_TIMESTAMP;
1664
    }
1665

1666
    uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);
1667
    tsVal = (int64_t)timeDouble;
1668
    if (tsLen == TSDB_TIME_PRECISION_SEC_DIGITS) {
1669 1670
      tsVal = tsVal * NANOSECOND_PER_SEC;
      timeDouble = timeDouble * NANOSECOND_PER_SEC;
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 1675
      }
    } else if (tsLen == TSDB_TIME_PRECISION_MILLI_DIGITS) {
1676 1677
      tsVal = tsVal * NANOSECOND_PER_MSEC;
      timeDouble = timeDouble * NANOSECOND_PER_MSEC;
X
Xiaoyu Wang 已提交
1678
      if (smlDoubleToInt64OverFlow(timeDouble)) {
1679
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1680
        return TSDB_CODE_INVALID_TIMESTAMP;
1681
      }
X
Xiaoyu Wang 已提交
1682
    } else if (timeDouble == 0) {
wmmhello's avatar
wmmhello 已提交
1683
      tsVal = taosGetTimestampNs();
X
Xiaoyu Wang 已提交
1684
    } else {
wmmhello's avatar
wmmhello 已提交
1685
      return TSDB_CODE_INVALID_TIMESTAMP;
1686 1687 1688 1689
    }
  } else if (cJSON_IsObject(timestamp)) {
    int32_t ret = smlParseTSFromJSONObj(info, timestamp, &tsVal);
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
1690
      uError("SML:0x%" PRIx64 " Failed to parse timestamp from JSON Obj", info->id);
1691 1692 1693 1694
      return ret;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
1695 1696
  }

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

X
Xiaoyu Wang 已提交
1711
static int32_t smlConvertJSONBool(SSmlKv *pVal, char *typeStr, cJSON *value) {
1712 1713 1714 1715 1716 1717 1718
  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 已提交
1719

1720 1721
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
1722

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

X
Xiaoyu Wang 已提交
1789
  // if reach here means type is unsupported
1790 1791 1792
  uError("OTD:invalid type(%s) for JSON Number", typeStr);
  return TSDB_CODE_TSC_INVALID_JSON_TYPE;
}
1793

X
Xiaoyu Wang 已提交
1794
static int32_t smlConvertJSONString(SSmlKv *pVal, char *typeStr, cJSON *value) {
1795 1796 1797 1798 1799 1800 1801 1802 1803
  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 已提交
1804 1805 1806 1807 1808 1809 1810 1811

  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 已提交
1812
  return smlJsonCreateSring(&pVal->value, value->valuestring, pVal->length);
1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838
}

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 已提交
1839
      }
1840
      break;
wmmhello's avatar
wmmhello 已提交
1841
    }
1842 1843 1844 1845 1846 1847 1848 1849 1850 1851 1852 1853 1854 1855 1856 1857
    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 已提交
1858
  }
1859 1860

  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1861 1862
}

1863 1864 1865 1866 1867 1868 1869 1870
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 已提交
1871
    }
1872 1873 1874 1875 1876 1877 1878 1879 1880 1881
    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 已提交
1882

X
Xiaoyu Wang 已提交
1883
      char *tsDefaultJSONStrType = "nchar";  // todo
1884 1885
      smlConvertJSONString(kv, tsDefaultJSONStrType, root);
      break;
wmmhello's avatar
wmmhello 已提交
1886
    }
1887 1888 1889 1890 1891 1892 1893 1894 1895 1896
    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 已提交
1897
  }
1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908

  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 已提交
1909
  if (!kv) {
1910 1911
    return TSDB_CODE_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
1912
  if (cols) taosArrayPush(cols, &kv);
1913 1914 1915 1916 1917 1918 1919 1920

  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 已提交
1921 1922
}

X
Xiaoyu Wang 已提交
1923 1924
static int32_t smlParseTagsFromJSON(cJSON *root, SArray *pKVs, char *childTableName, SHashObj *dumplicateKey,
                                    SSmlMsgBuf *msg) {
1925 1926 1927 1928 1929 1930
  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 已提交
1931

X
Xiaoyu Wang 已提交
1932
  size_t  childTableNameLen = strlen(tsSmlChildTableName);
1933 1934 1935 1936 1937
  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 已提交
1938
    }
1939 1940 1941 1942 1943
    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 已提交
1944
    // check duplicate keys
1945
    if (smlCheckDuplicateKey(tag->string, keyLen, dumplicateKey)) {
wmmhello's avatar
wmmhello 已提交
1946
      return TSDB_CODE_TSC_DUP_NAMES;
wmmhello's avatar
wmmhello 已提交
1947 1948
    }

X
Xiaoyu Wang 已提交
1949 1950
    // handle child table name
    if (childTableNameLen != 0 && strcmp(tag->string, tsSmlChildTableName) == 0) {
1951 1952 1953 1954 1955 1956 1957 1958 1959
      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;
    }

1960 1961
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1962 1963
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if (pKVs) taosArrayPush(pKVs, &kv);
1964

X
Xiaoyu Wang 已提交
1965
    // key
1966
    kv->keyLen = keyLen;
1967 1968 1969 1970
    ret = smlJsonCreateSring(&kv->key, tag->string, kv->keyLen);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
X
Xiaoyu Wang 已提交
1971
    // value
1972 1973 1974 1975
    ret = smlParseValueFromJSON(tag, kv);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
wmmhello's avatar
wmmhello 已提交
1976 1977
  }

1978
  return ret;
wmmhello's avatar
wmmhello 已提交
1979 1980
}

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

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

1989
  int32_t size = cJSON_GetArraySize(root);
X
Xiaoyu Wang 已提交
1990
  // outmost json fields has to be exactly 4
1991
  if (size != OTD_JSON_FIELDS_NUM) {
X
Xiaoyu Wang 已提交
1992
    uError("OTD:0x%" PRIx64 " Invalid number of JSON fields in data point %d", info->id, size);
1993
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
1994
  }
1995

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

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

X
Xiaoyu Wang 已提交
2012
  // Parse metric value
2013 2014
  ret = smlParseColsFromJSON(root, cols);
  if (ret) {
X
Xiaoyu Wang 已提交
2015
    uError("OTD:0x%" PRIx64 " Unable to parse metric value from JSON payload", info->id);
2016
    return ret;
2017
  }
X
Xiaoyu Wang 已提交
2018
  uDebug("OTD:0x%" PRIx64 " Parse metric value from JSON payload finished", info->id);
2019

X
Xiaoyu Wang 已提交
2020
  // Parse tags
2021
  ret = smlParseTagsFromJSON(root, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
2022
  if (ret) {
X
Xiaoyu Wang 已提交
2023
    uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
2024
    return ret;
2025
  }
X
Xiaoyu Wang 已提交
2026
  uDebug("OTD:0x%" PRIx64 " Parse tags from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2027

2028
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2029
}
2030
/************* TSDB_SML_JSON_PROTOCOL function end **************/
wmmhello's avatar
wmmhello 已提交
2031

X
Xiaoyu Wang 已提交
2032
static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql) {
wmmhello's avatar
wmmhello 已提交
2033
  SSmlLineInfo elements = {0};
X
Xiaoyu Wang 已提交
2034 2035 2036
  int          ret = smlParseInfluxString(sql, &elements, &info->msgBuf);
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseInfluxLine failed", info->id);
wmmhello's avatar
wmmhello 已提交
2037 2038 2039
    return ret;
  }

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

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

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

  ret = smlDealCols(*oneTable, info->dataFormat, cols);
X
Xiaoyu Wang 已提交
2084
  if (ret != TSDB_CODE_SUCCESS) {
2085 2086
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2087

X
Xiaoyu Wang 已提交
2088 2089 2090 2091 2092
  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 已提交
2093 2094 2095
      return ret;
    }

X
Xiaoyu Wang 已提交
2096
    if (taosArrayGetSize((*oneTable)->tags) > TSDB_MAX_TAGS) {
wmmhello's avatar
wmmhello 已提交
2097
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
wmmhello's avatar
wmmhello 已提交
2098
      return TSDB_CODE_PAR_INVALID_TAGS_NUM;
wmmhello's avatar
wmmhello 已提交
2099 2100
    }

2101 2102
    (*oneTable)->sTableName = elements.measure;
    (*oneTable)->sTableNameLen = elements.measureLen;
X
Xiaoyu Wang 已提交
2103 2104 2105
    if (strlen((*oneTable)->childTableName) == 0) {
      RandTableName rName = {(*oneTable)->tags, (*oneTable)->sTableName, (uint8_t)(*oneTable)->sTableNameLen,
                             (*oneTable)->childTableName, 0};
2106 2107 2108

      buildChildTableName(&rName);
      (*oneTable)->uid = rName.uid;
X
Xiaoyu Wang 已提交
2109 2110
    } else {
      (*oneTable)->uid = *(uint64_t *)((*oneTable)->childTableName);
2111
    }
2112
  }
wmmhello's avatar
wmmhello 已提交
2113

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

X
Xiaoyu Wang 已提交
2131
  if (!info->dataFormat) {
2132 2133 2134
    taosArrayClear(info->colsContainer);
  }
  taosHashClear(info->dumplicateKey);
wmmhello's avatar
wmmhello 已提交
2135 2136 2137
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2138 2139
static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) {
  int            ret = TSDB_CODE_SUCCESS;
2140
  SSmlTableInfo *tinfo = smlBuildTableInfo();
X
Xiaoyu Wang 已提交
2141
  if (!tinfo) {
2142
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2143 2144
  }

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

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

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

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

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

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

2213 2214
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
2215

X
Xiaoyu Wang 已提交
2216
static int32_t smlParseJSON(SSmlHandle *info, char *payload) {
2217 2218
  int32_t payloadNum = 0;
  int32_t ret = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2219

2220
  if (payload == NULL) {
X
Xiaoyu Wang 已提交
2221
    uError("SML:0x%" PRIx64 " empty JSON Payload", info->id);
2222
    return TSDB_CODE_TSC_INVALID_JSON;
2223
  }
2224 2225 2226

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

2241 2242 2243
  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 已提交
2244 2245
    if (ret != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2246 2247 2248 2249 2250 2251 2252
      goto end;
    }
  }

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

X
Xiaoyu Wang 已提交
2255
static int32_t smlInsertData(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
2256 2257
  int32_t code = TSDB_CODE_SUCCESS;

X
Xiaoyu Wang 已提交
2258
  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL);
wmmhello's avatar
wmmhello 已提交
2259
  while (oneTable) {
X
Xiaoyu Wang 已提交
2260
    SSmlTableInfo *tableData = *oneTable;
wmmhello's avatar
wmmhello 已提交
2261 2262 2263 2264

    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 已提交
2265 2266 2267 2268 2269 2270

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

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

X
Xiaoyu Wang 已提交
2280 2281 2282
    SSmlSTableMeta **pMeta =
        (SSmlSTableMeta **)taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen);
    ASSERT(NULL != pMeta && NULL != *pMeta);
wmmhello's avatar
wmmhello 已提交
2283

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

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

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

X
Xiaoyu Wang 已提交
2304 2305 2306
  // launchQueryImpl(info->pRequest, info->pQuery, false, NULL);
  //  info->affectedRows = taos_affected_rows(info->pRequest);
  //  return info->pRequest->code;
wmmhello's avatar
wmmhello 已提交
2307

D
dapan1121 已提交
2308
  launchAsyncQuery(info->pRequest, info->pQuery, NULL);
wmmhello's avatar
wmmhello 已提交
2309
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2310 2311
}

X
Xiaoyu Wang 已提交
2312 2313 2314 2315 2316 2317 2318 2319 2320
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);
2321 2322
}

X
Xiaoyu Wang 已提交
2323
static int32_t smlParseLine(SSmlHandle *info, char *lines[], int numLines) {
wmmhello's avatar
wmmhello 已提交
2324
  int32_t code = TSDB_CODE_SUCCESS;
2325 2326 2327 2328 2329 2330
  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 已提交
2331
    return code;
wmmhello's avatar
wmmhello 已提交
2332
  }
wmmhello's avatar
wmmhello 已提交
2333

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

X
Xiaoyu Wang 已提交
2350
static int smlProcess(SSmlHandle *info, char *lines[], int numLines) {
2351
  int32_t code = TSDB_CODE_SUCCESS;
2352 2353
  int32_t retryNum = 0;

2354 2355 2356 2357
  info->cost.parseTime = taosGetTimestampUs();

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

2362 2363 2364 2365 2366
  info->cost.lineNum = numLines;
  info->cost.numOfSTables = taosHashGetSize(info->superTables);
  info->cost.numOfCTables = taosHashGetSize(info->childTables);

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

X
Xiaoyu Wang 已提交
2368
  do {
2369 2370
    code = smlModifyDBSchemas(info);
    if (code == 0) break;
wmmhello's avatar
wmmhello 已提交
2371
  } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);
2372

wmmhello's avatar
wmmhello 已提交
2373
  if (code != 0) {
X
Xiaoyu Wang 已提交
2374
    uError("SML:0x%" PRIx64 " smlModifyDBSchemas error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2375
    return code;
wmmhello's avatar
wmmhello 已提交
2376
  }
wmmhello's avatar
wmmhello 已提交
2377

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

  return code;
}

X
Xiaoyu Wang 已提交
2388
static int32_t isSchemalessDb(STscObj *taos, SRequestObj *request) {
wmmhello's avatar
wmmhello 已提交
2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416
//  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 已提交
2417 2418 2419
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2420
static void smlInsertCallback(void *param, void *res, int32_t code) {
wmmhello's avatar
wmmhello 已提交
2421
  SRequestObj *pRequest = (SRequestObj *)res;
X
Xiaoyu Wang 已提交
2422
  SSmlHandle  *info = (SSmlHandle *)param;
wmmhello's avatar
wmmhello 已提交
2423
  int32_t rows = taos_affected_rows(pRequest);
wmmhello's avatar
wmmhello 已提交
2424

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

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

X
Xiaoyu Wang 已提交
2443
  if (isLast) {
wmmhello's avatar
wmmhello 已提交
2444
    tsem_post(&pParam->sem);
wmmhello's avatar
wmmhello 已提交
2445 2446 2447
  }
}

wmmhello's avatar
wmmhello 已提交
2448 2449 2450 2451 2452 2453 2454 2455 2456 2457 2458 2459 2460 2461 2462 2463 2464 2465 2466 2467 2468
/**
 * 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 已提交
2469
TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision) {
D
dapan1121 已提交
2470
  if (NULL == taos) {
2471 2472 2473
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }
D
dapan1121 已提交
2474

2475
  SRequestObj* request = (SRequestObj*)createRequest(*(int64_t*)taos, TSDB_SQL_INSERT);
wmmhello's avatar
wmmhello 已提交
2476
  if(!request){
2477
    uError("SML:taos_schemaless_insert error request is null");
2478
    return NULL;
wmmhello's avatar
wmmhello 已提交
2479 2480
  }

wmmhello's avatar
wmmhello 已提交
2481
  int    batchs = 0;
2482 2483
  STscObj* pTscObj = request->pTscObj;

2484
  pTscObj->schemalessType = 1;
wmmhello's avatar
wmmhello 已提交
2485
  SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
wmmhello's avatar
wmmhello 已提交
2486

wmmhello's avatar
wmmhello 已提交
2487 2488
  Params params;
  params.request = request;
wmmhello's avatar
wmmhello 已提交
2489 2490 2491
  tsem_init(&params.sem, 0, 0);
  taosThreadSpinInit(&(params.lock), 0);

X
Xiaoyu Wang 已提交
2492
  if (request->pDb == NULL) {
wmmhello's avatar
wmmhello 已提交
2493
    request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
wmmhello's avatar
wmmhello 已提交
2494
    smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
wmmhello's avatar
wmmhello 已提交
2495 2496 2497
    goto end;
  }

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

2504
  if (!lines) {
2505
    request->code = TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
2506
    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
2507 2508 2509
    goto end;
  }

X
Xiaoyu Wang 已提交
2510
  if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
2511
    request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
wmmhello's avatar
wmmhello 已提交
2512
    smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
2513
    goto end;
wmmhello's avatar
wmmhello 已提交
2514 2515
  }

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

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

wmmhello's avatar
wmmhello 已提交
2546 2547
    int32_t perBatch = LINE_BATCH;

X
Xiaoyu Wang 已提交
2548
    if (numLines > perBatch) {
wmmhello's avatar
wmmhello 已提交
2549 2550
      numLines -= perBatch;
      info->isLast = false;
X
Xiaoyu Wang 已提交
2551
    } else {
wmmhello's avatar
wmmhello 已提交
2552 2553 2554 2555 2556
      perBatch = numLines;
      numLines = 0;
      info->isLast = true;
    }

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

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