clientSml.c 81.6 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
      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) {
271
    uDebug("SML:0x%" PRIx64 " generate schema action. kv->name: %s, action: %d", info->id, kv->key,
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) {
439
    if(j == 0 && !isTag) continue;
X
Xiaoyu Wang 已提交
440 441
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, j);
    bool    actionNeeded = false;
442
    code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, action, &actionNeeded, info);
X
Xiaoyu Wang 已提交
443
    if (code != TSDB_CODE_SUCCESS) {
444 445 446 447 448 449 450 451 452 453 454 455
      return code;
    }
    if (actionNeeded) {
      code = smlApplySchemaAction(info, action);
      if (code != TSDB_CODE_SUCCESS) {
        return code;
      }
    }
  }
  return TSDB_CODE_SUCCESS;
}

456
static int32_t smlCheckMeta(SSchema *schema, int32_t length, SArray *cols, bool isTag) {
457
  SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
458 459
  int32_t i = 0;
  for ( ;i < length; i++) {
460 461 462
    taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
  }

463 464 465 466 467 468
  if (isTag){
    i = 0;
  } else {
    i = 1;
  }
  for (; i < taosArrayGetSize(cols); i++) {
X
Xiaoyu Wang 已提交
469 470
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
    if (taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL) {
471 472 473
      return -1;
    }
  }
474
  taosHashCleanup(hashTmp);
475 476 477
  return 0;
}

X
Xiaoyu Wang 已提交
478
static int32_t smlModifyDBSchemas(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
479
  int32_t code = 0;
X
Xiaoyu Wang 已提交
480
  SName   pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
481
  strcpy(pName.dbname, info->pRequest->pDb);
wmmhello's avatar
wmmhello 已提交
482

D
dapan1121 已提交
483 484 485 486 487
  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 已提交
488 489

  SSmlSTableMeta **tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, NULL);
wmmhello's avatar
wmmhello 已提交
490
  while (tableMetaSml) {
X
Xiaoyu Wang 已提交
491 492 493
    SSmlSTableMeta *sTableData = *tableMetaSml;
    STableMeta     *pTableMeta = NULL;
    bool            needCheckMeta = false;  // for multi thread
wmmhello's avatar
wmmhello 已提交
494

wmmhello's avatar
wmmhello 已提交
495
    size_t superTableLen = 0;
X
Xiaoyu Wang 已提交
496
    void  *superTable = taosHashGetKey(tableMetaSml, &superTableLen);
497
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
wmmhello's avatar
wmmhello 已提交
498
    memcpy(pName.tname, superTable, superTableLen);
wmmhello's avatar
wmmhello 已提交
499

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

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

524 525
      SSchemaAction schemaAction;
      memset(&schemaAction, 0, sizeof(SSchemaAction));
526
      memcpy(schemaAction.createSTable.sTableName, superTable, superTableLen);
527 528 529
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &schemaAction, true);
      if (code != TSDB_CODE_SUCCESS) {
        taosHashCleanup(hashTmp);
530
        goto end;
531 532 533
      }

      taosHashClear(hashTmp);
534
      for (uint16_t i = 1; i < pTableMeta->tableInfo.numOfColumns; i++) {
535 536 537 538 539
        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) {
540
        goto end;
wmmhello's avatar
wmmhello 已提交
541
      }
wmmhello's avatar
wmmhello 已提交
542

D
dapan1121 已提交
543
      code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
wmmhello's avatar
wmmhello 已提交
544
      if (code != TSDB_CODE_SUCCESS) {
545
        goto end;
wmmhello's avatar
wmmhello 已提交
546
      }
547
      needCheckMeta = true;
wmmhello's avatar
wmmhello 已提交
548
    } else {
X
Xiaoyu Wang 已提交
549
      uError("SML:0x%" PRIx64 " load table meta error: %s", info->id, tstrerror(code));
550
      goto end;
wmmhello's avatar
wmmhello 已提交
551
    }
X
Xiaoyu Wang 已提交
552
    if (pTableMeta) taosMemoryFree(pTableMeta);
553

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

X
Xiaoyu Wang 已提交
560 561
    if (needCheckMeta) {
      code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags,
562
                          sTableData->tags, true);
563
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
564
        uError("SML:0x%" PRIx64 " check tag failed. super table name %s", info->id, (char *)superTable);
565 566
        goto end;
      }
567
      code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols, false);
568
      if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
569
        uError("SML:0x%" PRIx64 " check cols failed. super table name %s", info->id, (char *)superTable);
570 571 572 573
        goto end;
      }
    }

574
    sTableData->tableMeta = pTableMeta;
wmmhello's avatar
wmmhello 已提交
575

X
Xiaoyu Wang 已提交
576
    tableMetaSml = (SSmlSTableMeta **)taosHashIterate(info->superTables, tableMetaSml);
wmmhello's avatar
wmmhello 已提交
577 578
  }
  return 0;
579 580

end:
D
dapan1121 已提交
581
  catalogRefreshTableMeta(info->pCatalog, &conn, &pName, 1);
582
  return code;
wmmhello's avatar
wmmhello 已提交
583 584
}

X
Xiaoyu Wang 已提交
585
static bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
586
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
587 588 589 590
  int32_t     len = kvVal->length;
  char       *endptr = NULL;
  double      result = taosStr2Double(pVal, &endptr);
  if (pVal == endptr) {
591
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
592 593 594
    return false;
  }

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

wmmhello's avatar
wmmhello 已提交
683
static bool smlParseBool(SSmlKv *kvVal) {
wmmhello's avatar
wmmhello 已提交
684
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
685
  int32_t     len = kvVal->length;
686
  if ((len == 1) && (pVal[0] == 't' || pVal[0] == 'T')) {
wmmhello's avatar
wmmhello 已提交
687
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
688 689 690
    return true;
  }

691
  if ((len == 1) && (pVal[0] == 'f' || pVal[0] == 'F')) {
wmmhello's avatar
wmmhello 已提交
692
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
693 694 695
    return true;
  }

X
Xiaoyu Wang 已提交
696
  if ((len == 4) && !strncasecmp(pVal, "true", len)) {
wmmhello's avatar
wmmhello 已提交
697
    kvVal->i = true;
wmmhello's avatar
wmmhello 已提交
698 699
    return true;
  }
X
Xiaoyu Wang 已提交
700
  if ((len == 5) && !strncasecmp(pVal, "false", len)) {
wmmhello's avatar
wmmhello 已提交
701
    kvVal->i = false;
wmmhello's avatar
wmmhello 已提交
702 703 704 705 706
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
707
static bool smlIsBinary(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
708
  // binary: "abc"
wmmhello's avatar
wmmhello 已提交
709 710 711 712 713 714 715 716 717
  if (len < 2) {
    return false;
  }
  if (pVal[0] == '"' && pVal[len - 1] == '"') {
    return true;
  }
  return false;
}

wmmhello's avatar
wmmhello 已提交
718
static bool smlIsNchar(const char *pVal, uint16_t len) {
X
Xiaoyu Wang 已提交
719
  // nchar: L"abc"
wmmhello's avatar
wmmhello 已提交
720 721 722
  if (len < 3) {
    return false;
  }
X
Xiaoyu Wang 已提交
723
  if ((pVal[0] == 'l' || pVal[0] == 'L') && pVal[1] == '"' && pVal[len - 1] == '"') {
wmmhello's avatar
wmmhello 已提交
724 725 726 727 728
    return true;
  }
  return false;
}

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

766
  return tsInt64;
767 768 769 770 771 772
}

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) {
773
    return TSDB_TIME_PRECISION_MILLI;
774 775
  } else {
    return -1;
wmmhello's avatar
wmmhello 已提交
776
  }
777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795
}

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 已提交
796
  }
797 798
}

X
Xiaoyu Wang 已提交
799 800
static int64_t smlParseInfluxTime(SSmlHandle *info, const char *data, int32_t len) {
  if (len == 0 || (len == 1 && data[0] == '0')) {
801
    return taosGetTimestampNs();
wmmhello's avatar
wmmhello 已提交
802 803
  }

804 805 806 807
  int8_t tsType = smlGetTsTypeByPrecision(info->precision);
  if (tsType == -1) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp precision", NULL);
    return -1;
wmmhello's avatar
wmmhello 已提交
808
  }
809 810

  int64_t ts = smlGetTimeValue(data, len, tsType);
X
Xiaoyu Wang 已提交
811
  if (ts == -1) {
812 813
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", data);
    return -1;
wmmhello's avatar
wmmhello 已提交
814
  }
815 816 817
  return ts;
}

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

X
Xiaoyu Wang 已提交
840
static int32_t smlParseTS(SSmlHandle *info, const char *data, int32_t len, SArray *cols) {
841
  int64_t ts = 0;
X
Xiaoyu Wang 已提交
842
  if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
843
//    uError("SML:data:%s,len:%d", data, len);
844
    ts = smlParseInfluxTime(info, data, len);
X
Xiaoyu Wang 已提交
845
  } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
846
    ts = smlParseOpenTsdbTime(info, data, len);
X
Xiaoyu Wang 已提交
847
  } else {
848
    ASSERT(0);
849
  }
850

wmmhello's avatar
wmmhello 已提交
851
  if (ts == -1) return TSDB_CODE_INVALID_TIMESTAMP;
852 853 854

  // add ts to
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
855
  if (!kv) {
856 857 858 859 860 861 862 863
    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 已提交
864
  if (cols) taosArrayPush(cols, &kv);
865 866 867
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
868
static int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
X
Xiaoyu Wang 已提交
869
  // binary
wmmhello's avatar
wmmhello 已提交
870
  if (smlIsBinary(pVal->value, pVal->length)) {
871
    pVal->type = TSDB_DATA_TYPE_BINARY;
wmmhello's avatar
wmmhello 已提交
872
    pVal->length -= BINARY_ADD_LEN;
wmmhello's avatar
wmmhello 已提交
873 874 875
    if (pVal->length > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
876
    pVal->value += (BINARY_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
877
    return TSDB_CODE_SUCCESS;
878
  }
X
Xiaoyu Wang 已提交
879
  // nchar
wmmhello's avatar
wmmhello 已提交
880
  if (smlIsNchar(pVal->value, pVal->length)) {
881
    pVal->type = TSDB_DATA_TYPE_NCHAR;
wmmhello's avatar
wmmhello 已提交
882
    pVal->length -= NCHAR_ADD_LEN;
wmmhello's avatar
wmmhello 已提交
883 884 885
    if(pVal->length > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }
886
    pVal->value += (NCHAR_ADD_LEN - 1);
wmmhello's avatar
wmmhello 已提交
887
    return TSDB_CODE_SUCCESS;
888 889
  }

X
Xiaoyu Wang 已提交
890
  // bool
891 892 893
  if (smlParseBool(pVal)) {
    pVal->type = TSDB_DATA_TYPE_BOOL;
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
894
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
895
  }
X
Xiaoyu Wang 已提交
896
  // number
897
  if (smlParseNumber(pVal, msg)) {
wmmhello's avatar
wmmhello 已提交
898
    pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
wmmhello's avatar
wmmhello 已提交
899
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
900 901
  }

wmmhello's avatar
wmmhello 已提交
902
  return TSDB_CODE_TSC_INVALID_VALUE;
wmmhello's avatar
wmmhello 已提交
903 904
}

X
Xiaoyu Wang 已提交
905 906
static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSmlMsgBuf *msg) {
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
907
  JUMP_SPACE(sql)
X
Xiaoyu Wang 已提交
908
  if (*sql == COMMA) return TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
909
  elements->measure = sql;
wmmhello's avatar
wmmhello 已提交
910

wmmhello's avatar
wmmhello 已提交
911
  // parse measure
wmmhello's avatar
wmmhello 已提交
912
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
913 914
    if ((sql != elements->measure) && IS_SLASH_LETTER(sql)) {
      MOVE_FORWARD_ONE(sql, strlen(sql) + 1);
wmmhello's avatar
wmmhello 已提交
915 916
      continue;
    }
X
Xiaoyu Wang 已提交
917
    if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
918 919 920
      break;
    }

X
Xiaoyu Wang 已提交
921
    if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
922 923
      break;
    }
wmmhello's avatar
wmmhello 已提交
924 925
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
926
  elements->measureLen = sql - elements->measure;
X
Xiaoyu Wang 已提交
927
  if (IS_INVALID_TABLE_LEN(elements->measureLen)) {
928
    smlBuildInvalidDataMsg(msg, "measure is empty or too large than 192", NULL);
wmmhello's avatar
wmmhello 已提交
929
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
wmmhello's avatar
wmmhello 已提交
930
  }
wmmhello's avatar
wmmhello 已提交
931

wmmhello's avatar
wmmhello 已提交
932
  // parse tag
X
Xiaoyu Wang 已提交
933
  if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
934
    elements->tagsLen = 0;
X
Xiaoyu Wang 已提交
935 936
  } else {
    if (*sql == COMMA) sql++;
wmmhello's avatar
wmmhello 已提交
937 938
    elements->tags = sql;
    while (*sql != '\0') {
X
Xiaoyu Wang 已提交
939
      if (IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
940 941 942
        break;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
943
    }
wmmhello's avatar
wmmhello 已提交
944
    elements->tagsLen = sql - elements->tags;
945
  }
wmmhello's avatar
wmmhello 已提交
946
  elements->measureTagsLen = sql - elements->measure;
wmmhello's avatar
wmmhello 已提交
947

wmmhello's avatar
wmmhello 已提交
948 949 950
  // parse cols
  JUMP_SPACE(sql)
  elements->cols = sql;
wmmhello's avatar
wmmhello 已提交
951
  bool isInQuote = false;
wmmhello's avatar
wmmhello 已提交
952
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
953
    if (IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
954 955
      isInQuote = !isInQuote;
    }
X
Xiaoyu Wang 已提交
956
    if (!isInQuote && IS_SPACE(sql)) {
wmmhello's avatar
wmmhello 已提交
957 958 959 960
      break;
    }
    sql++;
  }
X
Xiaoyu Wang 已提交
961
  if (isInQuote) {
962 963 964
    smlBuildInvalidDataMsg(msg, "only one quote", elements->cols);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
965
  elements->colsLen = sql - elements->cols;
X
Xiaoyu Wang 已提交
966
  if (elements->colsLen == 0) {
wmmhello's avatar
wmmhello 已提交
967 968 969
    smlBuildInvalidDataMsg(msg, "cols is empty", NULL);
    return TSDB_CODE_SML_INVALID_DATA;
  }
wmmhello's avatar
wmmhello 已提交
970

wmmhello's avatar
wmmhello 已提交
971 972 973
  // parse timestamp
  JUMP_SPACE(sql)
  elements->timestamp = sql;
wmmhello's avatar
wmmhello 已提交
974
  while (*sql != '\0') {
X
Xiaoyu Wang 已提交
975
    if (*sql == SPACE) {
wmmhello's avatar
wmmhello 已提交
976 977 978 979
      break;
    }
    sql++;
  }
wmmhello's avatar
wmmhello 已提交
980
  elements->timestampLen = sql - elements->timestamp;
wmmhello's avatar
wmmhello 已提交
981 982 983 984

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
985
static void smlParseTelnetElement(const char **sql, const char **data, int32_t *len) {
986
  while (**sql != '\0') {
X
Xiaoyu Wang 已提交
987
    if (**sql != SPACE && !(*data)) {
988
      *data = *sql;
X
Xiaoyu Wang 已提交
989
    } else if (**sql == SPACE && *data) {
990 991 992 993 994 995 996
      *len = *sql - *data;
      break;
    }
    (*sql)++;
  }
}

X
Xiaoyu Wang 已提交
997 998
static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTableName, SHashObj *dumplicateKey,
                                  SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
999
  const char *sql = data;
X
Xiaoyu Wang 已提交
1000 1001
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
  while (*sql != '\0') {
wmmhello's avatar
wmmhello 已提交
1002
    JUMP_SPACE(sql)
X
Xiaoyu Wang 已提交
1003
    if (*sql == '\0') break;
wmmhello's avatar
wmmhello 已提交
1004

wmmhello's avatar
wmmhello 已提交
1005
    const char *key = sql;
X
Xiaoyu Wang 已提交
1006
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
1007 1008

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

X
Xiaoyu Wang 已提交
1022
    if (IS_INVALID_COL_LEN(keyLen)) {
1023
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1024
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
1025
    }
X
Xiaoyu Wang 已提交
1026
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1027
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1028
      return TSDB_CODE_TSC_DUP_NAMES;
1029 1030 1031
    }

    // parse value
wmmhello's avatar
wmmhello 已提交
1032
    const char *value = sql;
X
Xiaoyu Wang 已提交
1033 1034
    int32_t     valueLen = 0;
    while (*sql != '\0') {
wmmhello's avatar
wmmhello 已提交
1035 1036
      // parse value
      if (*sql == SPACE) {
1037 1038
        break;
      }
wmmhello's avatar
wmmhello 已提交
1039 1040 1041 1042 1043
      if (*sql == EQUAL) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
1044
    }
wmmhello's avatar
wmmhello 已提交
1045
    valueLen = sql - value;
wmmhello's avatar
wmmhello 已提交
1046

X
Xiaoyu Wang 已提交
1047
    if (valueLen == 0) {
1048
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1049
      return TSDB_CODE_TSC_INVALID_VALUE;
1050
    }
wmmhello's avatar
wmmhello 已提交
1051

X
Xiaoyu Wang 已提交
1052 1053
    // handle child table name
    if (childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1054 1055 1056 1057 1058
      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 已提交
1059 1060 1061 1062
    if(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
      return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
    }

1063 1064
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1065
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1066 1067 1068
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1069
    kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1070
    kv->type = TSDB_DATA_TYPE_NCHAR;
1071

X
Xiaoyu Wang 已提交
1072
    if (cols) taosArrayPush(cols, &kv);
1073 1074 1075 1076
  }

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

1078
// format: <metric> <timestamp> <value> <tagk_1>=<tagv_1>[ <tagk_n>=<tagv_n>]
X
Xiaoyu Wang 已提交
1079 1080
static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTableInfo *tinfo, SArray *cols) {
  if (!sql) return TSDB_CODE_SML_INVALID_DATA;
1081 1082 1083

  // parse metric
  smlParseTelnetElement(&sql, &tinfo->sTableName, &tinfo->sTableNameLen);
1084
  if (!(tinfo->sTableName) || IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
1085
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1086
    return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
1087 1088 1089 1090
  }

  // parse timestamp
  const char *timestamp = NULL;
X
Xiaoyu Wang 已提交
1091
  int32_t     tLen = 0;
1092 1093 1094 1095 1096 1097 1098 1099 1100
  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 已提交
1101
    return ret;
1102 1103 1104 1105
  }

  // parse value
  const char *value = NULL;
X
Xiaoyu Wang 已提交
1106
  int32_t     valueLen = 0;
1107 1108 1109
  smlParseTelnetElement(&sql, &value, &valueLen);
  if (!value || valueLen == 0) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql);
wmmhello's avatar
wmmhello 已提交
1110
    return TSDB_CODE_TSC_INVALID_VALUE;
1111 1112 1113
  }

  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1114
  if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
1115 1116 1117 1118
  taosArrayPush(cols, &kv);
  kv->key = VALUE;
  kv->keyLen = VALUE_LEN;
  kv->value = value;
wmmhello's avatar
wmmhello 已提交
1119
  kv->length = valueLen;
wmmhello's avatar
wmmhello 已提交
1120 1121
  if ((ret = smlParseValue(kv, &info->msgBuf)) != TSDB_CODE_SUCCESS) {
    return ret;
1122 1123 1124
  }

  // parse tags
1125
  ret = smlParseTelnetTags(sql, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
1126 1127
  if (ret != TSDB_CODE_SUCCESS) {
    smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
wmmhello's avatar
wmmhello 已提交
1128
    return ret;
1129 1130 1131 1132 1133
  }

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1134 1135 1136
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 已提交
1137
    return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1138 1139
  }

X
Xiaoyu Wang 已提交
1140
  size_t      childTableNameLen = strlen(tsSmlChildTableName);
wmmhello's avatar
wmmhello 已提交
1141
  const char *sql = data;
X
Xiaoyu Wang 已提交
1142
  while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1143
    const char *key = sql;
X
Xiaoyu Wang 已提交
1144
    int32_t     keyLen = 0;
wmmhello's avatar
wmmhello 已提交
1145

X
Xiaoyu Wang 已提交
1146
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1147
      // parse key
X
Xiaoyu Wang 已提交
1148
      if (IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1149 1150 1151
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
1152
      if (IS_EQUAL(sql)) {
wmmhello's avatar
wmmhello 已提交
1153 1154
        keyLen = sql - key;
        sql++;
wmmhello's avatar
wmmhello 已提交
1155 1156
        break;
      }
wmmhello's avatar
wmmhello 已提交
1157
      sql++;
wmmhello's avatar
wmmhello 已提交
1158
    }
wmmhello's avatar
wmmhello 已提交
1159

X
Xiaoyu Wang 已提交
1160
    if (IS_INVALID_COL_LEN(keyLen)) {
wmmhello's avatar
wmmhello 已提交
1161
      smlBuildInvalidDataMsg(msg, "invalid key or key is too long than 64", key);
wmmhello's avatar
wmmhello 已提交
1162
      return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
wmmhello's avatar
wmmhello 已提交
1163
    }
X
Xiaoyu Wang 已提交
1164
    if (smlCheckDuplicateKey(key, keyLen, dumplicateKey)) {
1165
      smlBuildInvalidDataMsg(msg, "dumplicate key", key);
wmmhello's avatar
wmmhello 已提交
1166
      return TSDB_CODE_TSC_DUP_NAMES;
1167 1168
    }

wmmhello's avatar
wmmhello 已提交
1169
    // parse value
wmmhello's avatar
wmmhello 已提交
1170
    const char *value = sql;
X
Xiaoyu Wang 已提交
1171 1172 1173
    int32_t     valueLen = 0;
    bool        isInQuote = false;
    while (sql < data + len) {
wmmhello's avatar
wmmhello 已提交
1174
      // parse value
X
Xiaoyu Wang 已提交
1175
      if (!isTag && IS_QUOTE(sql)) {
wmmhello's avatar
wmmhello 已提交
1176
        isInQuote = !isInQuote;
wmmhello's avatar
wmmhello 已提交
1177 1178
        sql++;
        continue;
wmmhello's avatar
wmmhello 已提交
1179
      }
wmmhello's avatar
wmmhello 已提交
1180
      if (!isInQuote && IS_COMMA(sql)) {
wmmhello's avatar
wmmhello 已提交
1181 1182
        break;
      }
wmmhello's avatar
wmmhello 已提交
1183 1184 1185 1186 1187
      if (!isInQuote && IS_EQUAL(sql)) {
        smlBuildInvalidDataMsg(msg, "invalid data", sql);
        return TSDB_CODE_SML_INVALID_DATA;
      }
      sql++;
wmmhello's avatar
wmmhello 已提交
1188
    }
wmmhello's avatar
wmmhello 已提交
1189 1190 1191
    valueLen = sql - value;
    sql++;

X
Xiaoyu Wang 已提交
1192
    if (isInQuote) {
wmmhello's avatar
wmmhello 已提交
1193 1194 1195
      smlBuildInvalidDataMsg(msg, "only one quote", value);
      return TSDB_CODE_SML_INVALID_DATA;
    }
X
Xiaoyu Wang 已提交
1196
    if (valueLen == 0) {
wmmhello's avatar
wmmhello 已提交
1197
      smlBuildInvalidDataMsg(msg, "invalid value", value);
wmmhello's avatar
wmmhello 已提交
1198 1199
      return TSDB_CODE_SML_INVALID_DATA;
    }
wmmhello's avatar
wmmhello 已提交
1200 1201
    PROCESS_SLASH(key, keyLen)
    PROCESS_SLASH(value, valueLen)
wmmhello's avatar
wmmhello 已提交
1202

X
Xiaoyu Wang 已提交
1203 1204
    // handle child table name
    if (childTableName && childTableNameLen != 0 && strncmp(key, tsSmlChildTableName, keyLen) == 0) {
1205 1206 1207 1208 1209
      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 已提交
1210
    // add kv to SSmlKv
wmmhello's avatar
wmmhello 已提交
1211
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1212 1213
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if (cols) taosArrayPush(cols, &kv);
1214

wmmhello's avatar
wmmhello 已提交
1215 1216 1217
    kv->key = key;
    kv->keyLen = keyLen;
    kv->value = value;
wmmhello's avatar
wmmhello 已提交
1218
    kv->length = valueLen;
X
Xiaoyu Wang 已提交
1219
    if (isTag) {
1220 1221 1222
      if(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
        return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
      }
wmmhello's avatar
wmmhello 已提交
1223
      kv->type = TSDB_DATA_TYPE_NCHAR;
X
Xiaoyu Wang 已提交
1224
    } else {
wmmhello's avatar
wmmhello 已提交
1225 1226 1227
      int32_t ret = smlParseValue(kv, msg);
      if (ret != TSDB_CODE_SUCCESS) {
        return ret;
wmmhello's avatar
wmmhello 已提交
1228
      }
wmmhello's avatar
wmmhello 已提交
1229 1230
    }
  }
wmmhello's avatar
wmmhello 已提交
1231

wmmhello's avatar
wmmhello 已提交
1232 1233 1234
  return TSDB_CODE_SUCCESS;
}

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

wmmhello's avatar
wmmhello 已提交
1239
    int16_t *index = (int16_t *)taosHashGet(metaHash, kv->key, kv->keyLen);
X
Xiaoyu Wang 已提交
1240
    if (index) {
wmmhello's avatar
wmmhello 已提交
1241
      SSmlKv **value = (SSmlKv **)taosArrayGet(metaArray, *index);
X
Xiaoyu Wang 已提交
1242
      if (kv->type != (*value)->type) {
wmmhello's avatar
wmmhello 已提交
1243
        smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
wmmhello's avatar
wmmhello 已提交
1244
        return TSDB_CODE_SML_NOT_SAME_TYPE;
X
Xiaoyu Wang 已提交
1245 1246 1247
      } else {
        if (IS_VAR_DATA_TYPE(kv->type)) {  // update string len, if bigger
          if (kv->length > (*value)->length) {
wmmhello's avatar
wmmhello 已提交
1248
            *value = kv;
1249 1250 1251
          }
        }
      }
X
Xiaoyu Wang 已提交
1252
    } else {
wmmhello's avatar
wmmhello 已提交
1253 1254 1255 1256 1257
      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);
1258
    }
wmmhello's avatar
wmmhello 已提交
1259
  }
wmmhello's avatar
wmmhello 已提交
1260

wmmhello's avatar
wmmhello 已提交
1261
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1262 1263
}

X
Xiaoyu Wang 已提交
1264
static void smlInsertMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols) {
wmmhello's avatar
wmmhello 已提交
1265 1266 1267 1268
  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 已提交
1269 1270 1271
  }
}

X
Xiaoyu Wang 已提交
1272
static SSmlTableInfo *smlBuildTableInfo() {
1273
  SSmlTableInfo *tag = (SSmlTableInfo *)taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
X
Xiaoyu Wang 已提交
1274
  if (!tag) {
1275
    return NULL;
wmmhello's avatar
wmmhello 已提交
1276
  }
1277 1278 1279 1280 1281

  tag->cols = taosArrayInit(16, POINTER_BYTES);
  if (tag->cols == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1282
  }
1283 1284 1285 1286 1287

  tag->tags = taosArrayInit(16, POINTER_BYTES);
  if (tag->tags == NULL) {
    uError("SML:smlBuildTableInfo failed to allocate memory");
    goto cleanup;
wmmhello's avatar
wmmhello 已提交
1288
  }
1289 1290 1291 1292 1293
  return tag;

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

X
Xiaoyu Wang 已提交
1296 1297 1298
static void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag) {
  if (info->dataFormat) {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1299 1300
      SArray *kvArray = (SArray *)taosArrayGetP(tag->cols, i);
      for (int j = 0; j < taosArrayGetSize(kvArray); ++j) {
wmmhello's avatar
wmmhello 已提交
1301
        SSmlKv *p = (SSmlKv *)taosArrayGetP(kvArray, j);
X
Xiaoyu Wang 已提交
1302 1303 1304
        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 已提交
1305
        }
1306 1307 1308 1309
        taosMemoryFree(p);
      }
      taosArrayDestroy(kvArray);
    }
X
Xiaoyu Wang 已提交
1310 1311
  } else {
    for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
1312
      SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
X
Xiaoyu Wang 已提交
1313
      void    **p1 = (void **)taosHashIterate(kvHash, NULL);
1314 1315
      while (p1) {
        taosMemoryFree(*p1);
X
Xiaoyu Wang 已提交
1316
        p1 = (void **)taosHashIterate(kvHash, p1);
1317 1318 1319 1320
      }
      taosHashCleanup(kvHash);
    }
  }
X
Xiaoyu Wang 已提交
1321
  for (size_t i = 0; i < taosArrayGetSize(tag->tags); i++) {
wmmhello's avatar
wmmhello 已提交
1322
    SSmlKv *p = (SSmlKv *)taosArrayGetP(tag->tags, i);
X
Xiaoyu Wang 已提交
1323 1324 1325 1326
    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 已提交
1327 1328 1329 1330
      }
    }
    taosMemoryFree(p);
  }
X
Xiaoyu Wang 已提交
1331 1332
  if (info->protocol == TSDB_SML_JSON_PROTOCOL && tag->sTableName) {
    taosMemoryFree((void *)tag->sTableName);
wmmhello's avatar
wmmhello 已提交
1333
  }
1334 1335 1336 1337 1338
  taosArrayDestroy(tag->cols);
  taosArrayDestroy(tag->tags);
  taosMemoryFree(tag);
}

X
Xiaoyu Wang 已提交
1339
static int32_t smlKvTimeArrayCompare(const void *key1, const void *key2) {
1340 1341
  SArray *s1 = *(SArray **)key1;
  SArray *s2 = *(SArray **)key2;
1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354
  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 已提交
1355
static int32_t smlKvTimeHashCompare(const void *key1, const void *key2) {
1356 1357
  SHashObj *s1 = *(SHashObj **)key1;
  SHashObj *s2 = *(SHashObj **)key2;
1358 1359
  SSmlKv *kv1 = *(SSmlKv **)taosHashGet(s1, TS, TS_LEN);
  SSmlKv *kv2 = *(SSmlKv **)taosHashGet(s2, TS, TS_LEN);
1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370
  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;
  }
}

1371 1372
static int32_t smlDealCols(SSmlTableInfo* oneTable, bool dataFormat, SArray *cols){
  if(dataFormat){
1373
    void *p = taosArraySearch(oneTable->cols, &cols, smlKvTimeArrayCompare, TD_GT);
1374 1375
    if(p == NULL){
      taosArrayPush(oneTable->cols, &cols);
1376 1377
    }else{
      taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &cols);
1378
    }
1379 1380 1381 1382
    return TSDB_CODE_SUCCESS;
  }

  SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1383
  if (!kvHash) {
1384 1385 1386
    uError("SML:smlDealCols failed to allocate memory");
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
1387
  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
1388
    SSmlKv *kv = (SSmlKv *)taosArrayGetP(cols, i);
wmmhello's avatar
wmmhello 已提交
1389
    taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
1390 1391
  }

1392
  void *p = taosArraySearch(oneTable->cols, &kvHash, smlKvTimeHashCompare, TD_GT);
1393 1394
  if(p == NULL){
    taosArrayPush(oneTable->cols, &kvHash);
1395 1396
  }else{
    taosArrayInsert(oneTable->cols, TARRAY_ELEM_IDX(oneTable->cols, p), &kvHash);
1397
  }
1398 1399 1400
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1401 1402 1403
static SSmlSTableMeta *smlBuildSTableMeta() {
  SSmlSTableMeta *meta = (SSmlSTableMeta *)taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if (!meta) {
1404 1405 1406 1407 1408 1409 1410 1411
    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 已提交
1412 1413
  meta->colHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (meta->colHash == NULL) {
1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435
    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 已提交
1436
static void smlDestroySTableMeta(SSmlSTableMeta *meta) {
1437
  taosHashCleanup(meta->tagHash);
wmmhello's avatar
wmmhello 已提交
1438
  taosHashCleanup(meta->colHash);
1439 1440 1441
  taosArrayDestroy(meta->tags);
  taosArrayDestroy(meta->cols);
  taosMemoryFree(meta->tableMeta);
wmmhello's avatar
wmmhello 已提交
1442
  taosMemoryFree(meta);
1443 1444 1445 1446 1447
}

static void smlDestroyCols(SArray *cols) {
  if (!cols) return;
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
wmmhello's avatar
wmmhello 已提交
1448
    void *kv = taosArrayGetP(cols, i);
1449 1450 1451 1452
    taosMemoryFree(kv);
  }
}

X
Xiaoyu Wang 已提交
1453 1454
static void smlDestroyInfo(SSmlHandle *info) {
  if (!info) return;
1455 1456 1457 1458
  qDestroyQuery(info->pQuery);
  smlDestroyHandle(info->exec);

  // destroy info->childTables
X
Xiaoyu Wang 已提交
1459
  void **p1 = (void **)taosHashIterate(info->childTables, NULL);
1460
  while (p1) {
X
Xiaoyu Wang 已提交
1461 1462
    smlDestroyTableInfo(info, (SSmlTableInfo *)(*p1));
    p1 = (void **)taosHashIterate(info->childTables, p1);
1463 1464 1465 1466
  }
  taosHashCleanup(info->childTables);

  // destroy info->superTables
X
Xiaoyu Wang 已提交
1467
  p1 = (void **)taosHashIterate(info->superTables, NULL);
1468
  while (p1) {
X
Xiaoyu Wang 已提交
1469 1470
    smlDestroySTableMeta((SSmlSTableMeta *)(*p1));
    p1 = (void **)taosHashIterate(info->superTables, p1);
1471 1472 1473 1474 1475 1476
  }
  taosHashCleanup(info->superTables);

  // destroy info->pVgHash
  taosHashCleanup(info->pVgHash);
  taosHashCleanup(info->dumplicateKey);
X
Xiaoyu Wang 已提交
1477
  if (!info->dataFormat) {
wmmhello's avatar
wmmhello 已提交
1478 1479
    taosArrayDestroy(info->colsContainer);
  }
wmmhello's avatar
wmmhello 已提交
1480
  destroyRequest(info->pRequest);
1481 1482 1483
  taosMemoryFreeClear(info);
}

1484
static SSmlHandle* smlBuildSmlInfo(STscObj* pTscObj, SRequestObj* request, SMLProtocolType protocol, int8_t precision){
1485 1486 1487 1488 1489
  int32_t code = TSDB_CODE_SUCCESS;
  SSmlHandle* info = (SSmlHandle*)taosMemoryCalloc(1, sizeof(SSmlHandle));
  if (NULL == info) {
    return NULL;
  }
X
Xiaoyu Wang 已提交
1490
  info->id = smlGenId();
1491

1492
  info->pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY);
1493
  if (NULL == info->pQuery) {
X
Xiaoyu Wang 已提交
1494
    uError("SML:0x%" PRIx64 " create info->pQuery error", info->id);
1495 1496
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
1497
  info->pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
1498
  info->pQuery->haveResultSet = false;
X
Xiaoyu Wang 已提交
1499 1500 1501 1502
  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);
1503 1504
    goto cleanup;
  }
X
Xiaoyu Wang 已提交
1505
  ((SVnodeModifOpStmt *)(info->pQuery->pRoot))->payloadType = PAYLOAD_TYPE_KV;
1506

1507
  info->taos        = pTscObj;
wmmhello's avatar
wmmhello 已提交
1508
  code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
X
Xiaoyu Wang 已提交
1509 1510
  if (code != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " get catalog error %d", info->id, code);
wmmhello's avatar
wmmhello 已提交
1511 1512
    goto cleanup;
  }
1513

X
Xiaoyu Wang 已提交
1514 1515 1516
  info->precision = precision;
  info->protocol = protocol;
  if (protocol == TSDB_SML_LINE_PROTOCOL) {
1517
    info->dataFormat = tsSmlDataFormat;
X
Xiaoyu Wang 已提交
1518
  } else {
1519 1520
    info->dataFormat = true;
  }
X
Xiaoyu Wang 已提交
1521 1522 1523
  info->pRequest = request;
  info->msgBuf.buf = info->pRequest->msgBuf;
  info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
1524

X
Xiaoyu Wang 已提交
1525
  info->exec = smlInitHandle(info->pQuery);
1526 1527
  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 已提交
1528
  info->pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
1529 1530

  info->dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
X
Xiaoyu Wang 已提交
1531
  if (!info->dataFormat) {
1532
    info->colsContainer = taosArrayInit(32, POINTER_BYTES);
X
Xiaoyu Wang 已提交
1533 1534
    if (NULL == info->colsContainer) {
      uError("SML:0x%" PRIx64 " create info failed", info->id);
1535 1536 1537
      goto cleanup;
    }
  }
X
Xiaoyu Wang 已提交
1538 1539 1540
  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);
1541 1542 1543 1544 1545 1546 1547 1548 1549 1550
    goto cleanup;
  }

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

/************* TSDB_SML_JSON_PROTOCOL function start **************/
X
Xiaoyu Wang 已提交
1551
static int32_t smlJsonCreateSring(const char **output, char *input, int32_t inputLen) {
1552
  *output = (const char *)taosMemoryMalloc(inputLen);
X
Xiaoyu Wang 已提交
1553
  if (*output == NULL) {
1554 1555 1556
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

X
Xiaoyu Wang 已提交
1557
  memcpy((void *)(*output), input, inputLen);
1558 1559 1560 1561 1562 1563
  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 已提交
1564
    return TSDB_CODE_TSC_INVALID_JSON;
1565 1566 1567
  }

  tinfo->sTableNameLen = strlen(metric->valuestring);
1568
  if (IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) {
X
Xiaoyu Wang 已提交
1569
    uError("OTD:0x%" PRIx64 " Metric lenght is 0 or large than 192", info->id);
1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592
    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 已提交
1593
  if (smlDoubleToInt64OverFlow(timeDouble)) {
1594
    smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1595
    return TSDB_CODE_INVALID_TIMESTAMP;
1596
  }
wmmhello's avatar
wmmhello 已提交
1597 1598 1599 1600 1601 1602 1603 1604

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

  if (timeDouble < 0) {
    return TSDB_CODE_INVALID_TIMESTAMP;
1605 1606
  }

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

  cJSON *timestamp = cJSON_GetObjectItem(root, "timestamp");
  if (cJSON_IsNumber(timestamp)) {
X
Xiaoyu Wang 已提交
1667
    // timestamp value 0 indicates current system time
1668
    double timeDouble = timestamp->valuedouble;
X
Xiaoyu Wang 已提交
1669
    if (smlDoubleToInt64OverFlow(timeDouble)) {
1670
      smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1671
      return TSDB_CODE_INVALID_TIMESTAMP;
1672
    }
wmmhello's avatar
wmmhello 已提交
1673

X
Xiaoyu Wang 已提交
1674
    if (timeDouble < 0) {
wmmhello's avatar
wmmhello 已提交
1675
      return TSDB_CODE_INVALID_TIMESTAMP;
1676
    }
1677

1678
    uint8_t tsLen = smlGetTimestampLen((int64_t)timeDouble);
1679
    tsVal = (int64_t)timeDouble;
1680
    if (tsLen == TSDB_TIME_PRECISION_SEC_DIGITS) {
1681 1682
      tsVal = tsVal * NANOSECOND_PER_SEC;
      timeDouble = timeDouble * NANOSECOND_PER_SEC;
X
Xiaoyu Wang 已提交
1683
      if (smlDoubleToInt64OverFlow(timeDouble)) {
1684
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1685
        return TSDB_CODE_INVALID_TIMESTAMP;
1686 1687
      }
    } else if (tsLen == TSDB_TIME_PRECISION_MILLI_DIGITS) {
1688 1689
      tsVal = tsVal * NANOSECOND_PER_MSEC;
      timeDouble = timeDouble * NANOSECOND_PER_MSEC;
X
Xiaoyu Wang 已提交
1690
      if (smlDoubleToInt64OverFlow(timeDouble)) {
1691
        smlBuildInvalidDataMsg(&info->msgBuf, "timestamp is too large", NULL);
wmmhello's avatar
wmmhello 已提交
1692
        return TSDB_CODE_INVALID_TIMESTAMP;
1693
      }
X
Xiaoyu Wang 已提交
1694
    } else if (timeDouble == 0) {
wmmhello's avatar
wmmhello 已提交
1695
      tsVal = taosGetTimestampNs();
X
Xiaoyu Wang 已提交
1696
    } else {
wmmhello's avatar
wmmhello 已提交
1697
      return TSDB_CODE_INVALID_TIMESTAMP;
1698 1699 1700 1701
    }
  } else if (cJSON_IsObject(timestamp)) {
    int32_t ret = smlParseTSFromJSONObj(info, timestamp, &tsVal);
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
1702
      uError("SML:0x%" PRIx64 " Failed to parse timestamp from JSON Obj", info->id);
1703 1704 1705 1706
      return ret;
    }
  } else {
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
1707 1708
  }

wmmhello's avatar
wmmhello 已提交
1709
  // add ts to
wmmhello's avatar
wmmhello 已提交
1710
  SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1711
  if (!kv) {
wmmhello's avatar
wmmhello 已提交
1712 1713 1714
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  kv->key = TS;
1715 1716
  kv->keyLen = TS_LEN;
  kv->i = tsVal;
wmmhello's avatar
wmmhello 已提交
1717
  kv->type = TSDB_DATA_TYPE_TIMESTAMP;
wmmhello's avatar
wmmhello 已提交
1718
  kv->length = (int16_t)tDataTypes[kv->type].bytes;
X
Xiaoyu Wang 已提交
1719
  if (cols) taosArrayPush(cols, &kv);
wmmhello's avatar
wmmhello 已提交
1720 1721 1722
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1723
static int32_t smlConvertJSONBool(SSmlKv *pVal, char *typeStr, cJSON *value) {
1724 1725 1726 1727 1728 1729 1730
  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 已提交
1731

1732 1733
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
1734

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

X
Xiaoyu Wang 已提交
1801
  // if reach here means type is unsupported
1802 1803 1804
  uError("OTD:invalid type(%s) for JSON Number", typeStr);
  return TSDB_CODE_TSC_INVALID_JSON_TYPE;
}
1805

X
Xiaoyu Wang 已提交
1806
static int32_t smlConvertJSONString(SSmlKv *pVal, char *typeStr, cJSON *value) {
1807 1808 1809 1810 1811 1812 1813 1814 1815
  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 已提交
1816 1817 1818 1819 1820 1821 1822 1823

  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 已提交
1824
  return smlJsonCreateSring(&pVal->value, value->valuestring, pVal->length);
1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850
}

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 已提交
1851
      }
1852
      break;
wmmhello's avatar
wmmhello 已提交
1853
    }
1854 1855 1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869
    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 已提交
1870
  }
1871 1872

  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1873 1874
}

1875 1876 1877 1878 1879 1880 1881 1882
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 已提交
1883
    }
1884 1885 1886 1887 1888 1889 1890 1891 1892 1893
    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 已提交
1894

X
Xiaoyu Wang 已提交
1895
      char *tsDefaultJSONStrType = "nchar";  // todo
1896 1897
      smlConvertJSONString(kv, tsDefaultJSONStrType, root);
      break;
wmmhello's avatar
wmmhello 已提交
1898
    }
1899 1900 1901 1902 1903 1904 1905 1906 1907 1908
    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 已提交
1909
  }
1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920

  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 已提交
1921
  if (!kv) {
1922 1923
    return TSDB_CODE_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
1924
  if (cols) taosArrayPush(cols, &kv);
1925 1926 1927 1928 1929 1930 1931 1932

  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 已提交
1933 1934
}

X
Xiaoyu Wang 已提交
1935 1936
static int32_t smlParseTagsFromJSON(cJSON *root, SArray *pKVs, char *childTableName, SHashObj *dumplicateKey,
                                    SSmlMsgBuf *msg) {
1937 1938 1939 1940 1941 1942
  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 已提交
1943

X
Xiaoyu Wang 已提交
1944
  size_t  childTableNameLen = strlen(tsSmlChildTableName);
1945 1946 1947 1948 1949
  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 已提交
1950
    }
1951 1952 1953 1954 1955
    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 已提交
1956
    // check duplicate keys
1957
    if (smlCheckDuplicateKey(tag->string, keyLen, dumplicateKey)) {
wmmhello's avatar
wmmhello 已提交
1958
      return TSDB_CODE_TSC_DUP_NAMES;
wmmhello's avatar
wmmhello 已提交
1959 1960
    }

X
Xiaoyu Wang 已提交
1961 1962
    // handle child table name
    if (childTableNameLen != 0 && strcmp(tag->string, tsSmlChildTableName) == 0) {
1963 1964 1965 1966 1967 1968 1969 1970 1971
      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;
    }

1972 1973
    // add kv to SSmlKv
    SSmlKv *kv = (SSmlKv *)taosMemoryCalloc(sizeof(SSmlKv), 1);
X
Xiaoyu Wang 已提交
1974 1975
    if (!kv) return TSDB_CODE_OUT_OF_MEMORY;
    if (pKVs) taosArrayPush(pKVs, &kv);
1976

X
Xiaoyu Wang 已提交
1977
    // key
1978
    kv->keyLen = keyLen;
1979 1980 1981 1982
    ret = smlJsonCreateSring(&kv->key, tag->string, kv->keyLen);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
X
Xiaoyu Wang 已提交
1983
    // value
1984 1985 1986 1987
    ret = smlParseValueFromJSON(tag, kv);
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }
wmmhello's avatar
wmmhello 已提交
1988 1989
  }

1990
  return ret;
wmmhello's avatar
wmmhello 已提交
1991 1992
}

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

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

2001
  int32_t size = cJSON_GetArraySize(root);
X
Xiaoyu Wang 已提交
2002
  // outmost json fields has to be exactly 4
2003
  if (size != OTD_JSON_FIELDS_NUM) {
X
Xiaoyu Wang 已提交
2004
    uError("OTD:0x%" PRIx64 " Invalid number of JSON fields in data point %d", info->id, size);
2005
    return TSDB_CODE_TSC_INVALID_JSON;
wmmhello's avatar
wmmhello 已提交
2006
  }
2007

X
Xiaoyu Wang 已提交
2008
  // Parse metric
2009 2010
  ret = smlParseMetricFromJSON(info, root, tinfo);
  if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2011
    uError("OTD:0x%" PRIx64 " Unable to parse metric from JSON payload", info->id);
2012
    return ret;
wmmhello's avatar
wmmhello 已提交
2013
  }
X
Xiaoyu Wang 已提交
2014
  uDebug("OTD:0x%" PRIx64 " Parse metric from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2015

X
Xiaoyu Wang 已提交
2016
  // Parse timestamp
2017 2018
  ret = smlParseTSFromJSON(info, root, cols);
  if (ret) {
X
Xiaoyu Wang 已提交
2019
    uError("OTD:0x%" PRIx64 " Unable to parse timestamp from JSON payload", info->id);
2020
    return ret;
wmmhello's avatar
wmmhello 已提交
2021
  }
X
Xiaoyu Wang 已提交
2022
  uDebug("OTD:0x%" PRIx64 " Parse timestamp from JSON payload finished", info->id);
2023

X
Xiaoyu Wang 已提交
2024
  // Parse metric value
2025 2026
  ret = smlParseColsFromJSON(root, cols);
  if (ret) {
X
Xiaoyu Wang 已提交
2027
    uError("OTD:0x%" PRIx64 " Unable to parse metric value from JSON payload", info->id);
2028
    return ret;
2029
  }
X
Xiaoyu Wang 已提交
2030
  uDebug("OTD:0x%" PRIx64 " Parse metric value from JSON payload finished", info->id);
2031

X
Xiaoyu Wang 已提交
2032
  // Parse tags
2033
  ret = smlParseTagsFromJSON(root, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf);
2034
  if (ret) {
X
Xiaoyu Wang 已提交
2035
    uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
2036
    return ret;
2037
  }
X
Xiaoyu Wang 已提交
2038
  uDebug("OTD:0x%" PRIx64 " Parse tags from JSON payload finished", info->id);
wmmhello's avatar
wmmhello 已提交
2039

2040
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2041
}
2042
/************* TSDB_SML_JSON_PROTOCOL function end **************/
wmmhello's avatar
wmmhello 已提交
2043

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

X
Xiaoyu Wang 已提交
2048 2049 2050
  int          ret = smlParseInfluxString(sql, &elements, &info->msgBuf);
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseInfluxLine failed", info->id);
wmmhello's avatar
wmmhello 已提交
2051 2052 2053
    return ret;
  }

2054
  SArray *cols = NULL;
X
Xiaoyu Wang 已提交
2055
  if (info->dataFormat) {  // if dataFormat, cols need new memory to save data
2056 2057
    cols = taosArrayInit(16, POINTER_BYTES);
    if (cols == NULL) {
X
Xiaoyu Wang 已提交
2058
      uError("SML:0x%" PRIx64 " smlParseInfluxLine failed to allocate memory", info->id);
2059 2060
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
X
Xiaoyu Wang 已提交
2061
  } else {  // if dataFormat is false, cols do not need to save data, there is another new memory to save data
2062
    cols = info->colsContainer;
wmmhello's avatar
wmmhello 已提交
2063
  }
wmmhello's avatar
wmmhello 已提交
2064

wmmhello's avatar
wmmhello 已提交
2065
  ret = smlParseTS(info, elements.timestamp, elements.timestampLen, cols);
X
Xiaoyu Wang 已提交
2066 2067 2068
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseTS failed", info->id);
    if (info->dataFormat) taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2069 2070
    return ret;
  }
2071
  ret = smlParseCols(elements.cols, elements.colsLen, cols, NULL, false, info->dumplicateKey, &info->msgBuf);
X
Xiaoyu Wang 已提交
2072 2073
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseCols parse cloums fields failed", info->id);
2074
    smlDestroyCols(cols);
X
Xiaoyu Wang 已提交
2075
    if (info->dataFormat) taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2076 2077
    return ret;
  }
X
Xiaoyu Wang 已提交
2078
  if (taosArrayGetSize(cols) > TSDB_MAX_COLUMNS) {
wmmhello's avatar
wmmhello 已提交
2079
    smlBuildInvalidDataMsg(&info->msgBuf, "too many columns than 4096", NULL);
wmmhello's avatar
wmmhello 已提交
2080
    return TSDB_CODE_PAR_TOO_MANY_COLUMNS;
wmmhello's avatar
wmmhello 已提交
2081
  }
wmmhello's avatar
wmmhello 已提交
2082

X
Xiaoyu Wang 已提交
2083 2084 2085 2086 2087
  bool            hasTable = true;
  SSmlTableInfo  *tinfo = NULL;
  SSmlTableInfo **oneTable =
      (SSmlTableInfo **)taosHashGet(info->childTables, elements.measure, elements.measureTagsLen);
  if (!oneTable) {
2088
    tinfo = smlBuildTableInfo();
X
Xiaoyu Wang 已提交
2089
    if (!tinfo) {
wmmhello's avatar
wmmhello 已提交
2090 2091
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
2092
    taosHashPut(info->childTables, elements.measure, elements.measureTagsLen, &tinfo, POINTER_BYTES);
2093
    oneTable = &tinfo;
2094 2095 2096 2097
    hasTable = false;
  }

  ret = smlDealCols(*oneTable, info->dataFormat, cols);
X
Xiaoyu Wang 已提交
2098
  if (ret != TSDB_CODE_SUCCESS) {
2099 2100
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2101

X
Xiaoyu Wang 已提交
2102 2103 2104 2105 2106
  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 已提交
2107 2108 2109
      return ret;
    }

X
Xiaoyu Wang 已提交
2110
    if (taosArrayGetSize((*oneTable)->tags) > TSDB_MAX_TAGS) {
wmmhello's avatar
wmmhello 已提交
2111
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
wmmhello's avatar
wmmhello 已提交
2112
      return TSDB_CODE_PAR_INVALID_TAGS_NUM;
wmmhello's avatar
wmmhello 已提交
2113 2114
    }

2115 2116
    (*oneTable)->sTableName = elements.measure;
    (*oneTable)->sTableNameLen = elements.measureLen;
X
Xiaoyu Wang 已提交
2117 2118 2119
    if (strlen((*oneTable)->childTableName) == 0) {
      RandTableName rName = {(*oneTable)->tags, (*oneTable)->sTableName, (uint8_t)(*oneTable)->sTableNameLen,
                             (*oneTable)->childTableName, 0};
2120 2121 2122

      buildChildTableName(&rName);
      (*oneTable)->uid = rName.uid;
X
Xiaoyu Wang 已提交
2123 2124
    } else {
      (*oneTable)->uid = *(uint64_t *)((*oneTable)->childTableName);
2125
    }
2126
  }
wmmhello's avatar
wmmhello 已提交
2127

X
Xiaoyu Wang 已提交
2128 2129
  SSmlSTableMeta **tableMeta = (SSmlSTableMeta **)taosHashGet(info->superTables, elements.measure, elements.measureLen);
  if (tableMeta) {  // update meta
wmmhello's avatar
wmmhello 已提交
2130
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
2131
    if (!hasTable && ret == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2132 2133
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
wmmhello's avatar
wmmhello 已提交
2134
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2135
      uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
wmmhello's avatar
wmmhello 已提交
2136
      return ret;
wmmhello's avatar
wmmhello 已提交
2137
    }
X
Xiaoyu Wang 已提交
2138
  } else {
2139
    SSmlSTableMeta *meta = smlBuildSTableMeta();
wmmhello's avatar
wmmhello 已提交
2140 2141
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
2142
    taosHashPut(info->superTables, elements.measure, elements.measureLen, &meta, POINTER_BYTES);
wmmhello's avatar
wmmhello 已提交
2143
  }
2144

X
Xiaoyu Wang 已提交
2145
  if (!info->dataFormat) {
2146 2147 2148
    taosArrayClear(info->colsContainer);
  }
  taosHashClear(info->dumplicateKey);
wmmhello's avatar
wmmhello 已提交
2149 2150 2151
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2152 2153
static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) {
  int            ret = TSDB_CODE_SUCCESS;
2154
  SSmlTableInfo *tinfo = smlBuildTableInfo();
X
Xiaoyu Wang 已提交
2155
  if (!tinfo) {
2156
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2157 2158
  }

2159 2160
  SArray *cols = taosArrayInit(16, POINTER_BYTES);
  if (cols == NULL) {
X
Xiaoyu Wang 已提交
2161
    uError("SML:0x%" PRIx64 " smlParseTelnetLine failed to allocate memory", info->id);
2162
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2163 2164
  }

X
Xiaoyu Wang 已提交
2165 2166 2167
  if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
    ret = smlParseTelnetString(info, (const char *)data, tinfo, cols);
  } else if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
2168
    ret = smlParseJSONString(info, (cJSON *)data, tinfo, cols);
X
Xiaoyu Wang 已提交
2169
  } else {
2170 2171
    ASSERT(0);
  }
X
Xiaoyu Wang 已提交
2172 2173
  if (ret != TSDB_CODE_SUCCESS) {
    uError("SML:0x%" PRIx64 " smlParseTelnetLine failed", info->id);
wmmhello's avatar
wmmhello 已提交
2174
    smlDestroyTableInfo(info, tinfo);
wmmhello's avatar
wmmhello 已提交
2175
    smlDestroyCols(cols);
2176 2177
    taosArrayDestroy(cols);
    return ret;
wmmhello's avatar
wmmhello 已提交
2178
  }
wmmhello's avatar
wmmhello 已提交
2179

X
Xiaoyu Wang 已提交
2180
  if (taosArrayGetSize(tinfo->tags) <= 0 || taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS) {
2181
    smlBuildInvalidDataMsg(&info->msgBuf, "invalidate tags length:[1,128]", NULL);
wmmhello's avatar
wmmhello 已提交
2182 2183 2184
    smlDestroyTableInfo(info, tinfo);
    smlDestroyCols(cols);
    taosArrayDestroy(cols);
wmmhello's avatar
wmmhello 已提交
2185
    return TSDB_CODE_PAR_INVALID_TAGS_NUM;
wmmhello's avatar
wmmhello 已提交
2186
  }
2187 2188
  taosHashClear(info->dumplicateKey);

X
Xiaoyu Wang 已提交
2189 2190
  if (strlen(tinfo->childTableName) == 0) {
    RandTableName rName = {tinfo->tags, tinfo->sTableName, (uint8_t)tinfo->sTableNameLen, tinfo->childTableName, 0};
2191 2192
    buildChildTableName(&rName);
    tinfo->uid = rName.uid;
X
Xiaoyu Wang 已提交
2193 2194
  } else {
    tinfo->uid = *(uint64_t *)(tinfo->childTableName);  // generate uid by name simple
2195 2196
  }

X
Xiaoyu Wang 已提交
2197 2198 2199 2200
  bool            hasTable = true;
  SSmlTableInfo **oneTable =
      (SSmlTableInfo **)taosHashGet(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName));
  if (!oneTable) {
2201
    taosHashPut(info->childTables, tinfo->childTableName, strlen(tinfo->childTableName), &tinfo, POINTER_BYTES);
2202
    oneTable = &tinfo;
2203
    hasTable = false;
X
Xiaoyu Wang 已提交
2204
  } else {
wmmhello's avatar
wmmhello 已提交
2205
    smlDestroyTableInfo(info, tinfo);
wmmhello's avatar
wmmhello 已提交
2206
  }
wmmhello's avatar
wmmhello 已提交
2207

2208
  taosArrayPush((*oneTable)->cols, &cols);
X
Xiaoyu Wang 已提交
2209 2210 2211
  SSmlSTableMeta **tableMeta =
      (SSmlSTableMeta **)taosHashGet(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen);
  if (tableMeta) {  // update meta
wmmhello's avatar
wmmhello 已提交
2212
    ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, cols, &info->msgBuf);
wmmhello's avatar
wmmhello 已提交
2213
    if (!hasTable && ret == TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2214 2215
      ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, (*oneTable)->tags, &info->msgBuf);
    }
wmmhello's avatar
wmmhello 已提交
2216
    if (ret != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2217
      uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
wmmhello's avatar
wmmhello 已提交
2218
      return ret;
2219
    }
X
Xiaoyu Wang 已提交
2220
  } else {
2221
    SSmlSTableMeta *meta = smlBuildSTableMeta();
wmmhello's avatar
wmmhello 已提交
2222 2223
    smlInsertMeta(meta->tagHash, meta->tags, (*oneTable)->tags);
    smlInsertMeta(meta->colHash, meta->cols, cols);
2224
    taosHashPut(info->superTables, (*oneTable)->sTableName, (*oneTable)->sTableNameLen, &meta, POINTER_BYTES);
wmmhello's avatar
wmmhello 已提交
2225 2226
  }

2227 2228
  return TSDB_CODE_SUCCESS;
}
wmmhello's avatar
wmmhello 已提交
2229

X
Xiaoyu Wang 已提交
2230
static int32_t smlParseJSON(SSmlHandle *info, char *payload) {
2231 2232
  int32_t payloadNum = 0;
  int32_t ret = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2233

2234
  if (payload == NULL) {
X
Xiaoyu Wang 已提交
2235
    uError("SML:0x%" PRIx64 " empty JSON Payload", info->id);
2236
    return TSDB_CODE_TSC_INVALID_JSON;
2237
  }
2238 2239 2240

  cJSON *root = cJSON_Parse(payload);
  if (root == NULL) {
X
Xiaoyu Wang 已提交
2241
    uError("SML:0x%" PRIx64 " parse json failed:%s", info->id, payload);
2242 2243
    return TSDB_CODE_TSC_INVALID_JSON;
  }
X
Xiaoyu Wang 已提交
2244
  // multiple data points must be sent in JSON array
2245 2246 2247 2248 2249
  if (cJSON_IsObject(root)) {
    payloadNum = 1;
  } else if (cJSON_IsArray(root)) {
    payloadNum = cJSON_GetArraySize(root);
  } else {
X
Xiaoyu Wang 已提交
2250
    uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2251 2252
    ret = TSDB_CODE_TSC_INVALID_JSON;
    goto end;
wmmhello's avatar
wmmhello 已提交
2253
  }
wmmhello's avatar
wmmhello 已提交
2254

2255 2256 2257
  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 已提交
2258 2259
    if (ret != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id);
2260 2261 2262 2263 2264 2265 2266
      goto end;
    }
  }

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

X
Xiaoyu Wang 已提交
2269
static int32_t smlInsertData(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
2270 2271
  int32_t code = TSDB_CODE_SUCCESS;

X
Xiaoyu Wang 已提交
2272
  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL);
wmmhello's avatar
wmmhello 已提交
2273
  while (oneTable) {
X
Xiaoyu Wang 已提交
2274
    SSmlTableInfo *tableData = *oneTable;
wmmhello's avatar
wmmhello 已提交
2275 2276 2277 2278

    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 已提交
2279 2280 2281 2282 2283 2284

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

wmmhello's avatar
wmmhello 已提交
2286
    SVgroupInfo vg;
D
dapan1121 已提交
2287
    code = catalogGetTableHashVgroup(info->pCatalog, &conn, &pName, &vg);
2288
    if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2289
      uError("SML:0x%" PRIx64 " catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName);
wmmhello's avatar
wmmhello 已提交
2290 2291
      return code;
    }
X
Xiaoyu Wang 已提交
2292
    taosHashPut(info->pVgHash, (const char *)&vg.vgId, sizeof(vg.vgId), (char *)&vg, sizeof(vg));
wmmhello's avatar
wmmhello 已提交
2293

X
Xiaoyu Wang 已提交
2294 2295 2296
    SSmlSTableMeta **pMeta =
        (SSmlSTableMeta **)taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen);
    ASSERT(NULL != pMeta && NULL != *pMeta);
wmmhello's avatar
wmmhello 已提交
2297

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

2302
    code = smlBindData(info->exec, tableData->tags, (*pMeta)->cols, tableData->cols, info->dataFormat,
wmmhello's avatar
wmmhello 已提交
2303
                       (*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen, info->msgBuf.buf, info->msgBuf.len);
X
Xiaoyu Wang 已提交
2304 2305
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
wmmhello's avatar
wmmhello 已提交
2306 2307
      return code;
    }
X
Xiaoyu Wang 已提交
2308
    oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, oneTable);
wmmhello's avatar
wmmhello 已提交
2309
  }
wmmhello's avatar
wmmhello 已提交
2310

2311 2312
  code = smlBuildOutput(info->exec, info->pVgHash);
  if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
2313
    uError("SML:0x%" PRIx64 " smlBuildOutput failed", info->id);
2314 2315
    return code;
  }
2316 2317
  info->cost.insertRpcTime = taosGetTimestampUs();

X
Xiaoyu Wang 已提交
2318 2319 2320
  // launchQueryImpl(info->pRequest, info->pQuery, false, NULL);
  //  info->affectedRows = taos_affected_rows(info->pRequest);
  //  return info->pRequest->code;
wmmhello's avatar
wmmhello 已提交
2321

D
dapan1121 已提交
2322
  launchAsyncQuery(info->pRequest, info->pQuery, NULL);
wmmhello's avatar
wmmhello 已提交
2323
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2324 2325
}

X
Xiaoyu Wang 已提交
2326 2327 2328 2329 2330 2331 2332 2333 2334
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);
2335 2336
}

X
Xiaoyu Wang 已提交
2337
static int32_t smlParseLine(SSmlHandle *info, char *lines[], int numLines) {
wmmhello's avatar
wmmhello 已提交
2338
  int32_t code = TSDB_CODE_SUCCESS;
2339 2340 2341 2342 2343 2344
  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 已提交
2345
    return code;
wmmhello's avatar
wmmhello 已提交
2346
  }
wmmhello's avatar
wmmhello 已提交
2347

wmmhello's avatar
wmmhello 已提交
2348
  for (int32_t i = 0; i < numLines; ++i) {
X
Xiaoyu Wang 已提交
2349
    if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
2350
      code = smlParseInfluxLine(info, lines[i]);
X
Xiaoyu Wang 已提交
2351
    } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
2352
      code = smlParseTelnetLine(info, lines[i]);
X
Xiaoyu Wang 已提交
2353
    } else {
2354 2355
      ASSERT(0);
    }
wmmhello's avatar
wmmhello 已提交
2356
    if (code != TSDB_CODE_SUCCESS) {
2357 2358
      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, lines[i]);
      return code;
wmmhello's avatar
wmmhello 已提交
2359 2360
    }
  }
2361 2362 2363
  return code;
}

X
Xiaoyu Wang 已提交
2364
static int smlProcess(SSmlHandle *info, char *lines[], int numLines) {
2365
  int32_t code = TSDB_CODE_SUCCESS;
2366 2367
  int32_t retryNum = 0;

2368 2369 2370 2371
  info->cost.parseTime = taosGetTimestampUs();

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

2376 2377 2378 2379 2380
  info->cost.lineNum = numLines;
  info->cost.numOfSTables = taosHashGetSize(info->superTables);
  info->cost.numOfCTables = taosHashGetSize(info->childTables);

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

X
Xiaoyu Wang 已提交
2382
  do {
2383 2384
    code = smlModifyDBSchemas(info);
    if (code == 0) break;
wmmhello's avatar
wmmhello 已提交
2385
  } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);
2386

wmmhello's avatar
wmmhello 已提交
2387
  if (code != 0) {
X
Xiaoyu Wang 已提交
2388
    uError("SML:0x%" PRIx64 " smlModifyDBSchemas error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2389
    return code;
wmmhello's avatar
wmmhello 已提交
2390
  }
wmmhello's avatar
wmmhello 已提交
2391

2392
  info->cost.insertBindTime = taosGetTimestampUs();
wmmhello's avatar
wmmhello 已提交
2393 2394
  code = smlInsertData(info);
  if (code != 0) {
X
Xiaoyu Wang 已提交
2395
    uError("SML:0x%" PRIx64 " smlInsertData error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2396
    return code;
wmmhello's avatar
wmmhello 已提交
2397 2398 2399 2400 2401
  }

  return code;
}

X
Xiaoyu Wang 已提交
2402
static int32_t isSchemalessDb(STscObj *taos, SRequestObj *request) {
wmmhello's avatar
wmmhello 已提交
2403 2404 2405 2406 2407 2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424 2425 2426 2427 2428 2429 2430
//  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 已提交
2431 2432 2433
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2434
static void smlInsertCallback(void *param, void *res, int32_t code) {
wmmhello's avatar
wmmhello 已提交
2435
  SRequestObj *pRequest = (SRequestObj *)res;
X
Xiaoyu Wang 已提交
2436
  SSmlHandle  *info = (SSmlHandle *)param;
wmmhello's avatar
wmmhello 已提交
2437
  int32_t rows = taos_affected_rows(pRequest);
wmmhello's avatar
wmmhello 已提交
2438

X
Xiaoyu Wang 已提交
2439
  uDebug("SML:0x%" PRIx64 " result. code:%d, msg:%s", info->id, pRequest->code, pRequest->msgBuf);
wmmhello's avatar
wmmhello 已提交
2440
  // lock
wmmhello's avatar
wmmhello 已提交
2441 2442
  taosThreadSpinLock(&info->params->lock);
  info->params->request->body.resInfo.numOfRows += rows;
X
Xiaoyu Wang 已提交
2443
  if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2444
    info->params->request->code = code;
wmmhello's avatar
wmmhello 已提交
2445
  }
wmmhello's avatar
wmmhello 已提交
2446
  taosThreadSpinUnlock(&info->params->lock);
wmmhello's avatar
wmmhello 已提交
2447 2448
  // unlock

wmmhello's avatar
wmmhello 已提交
2449
  uDebug("SML:0x%" PRIx64 " insert finished, code: %d, rows: %d, total: %d", info->id, code, rows, info->affectedRows);
wmmhello's avatar
wmmhello 已提交
2450
  Params *pParam = info->params;
X
Xiaoyu Wang 已提交
2451
  bool    isLast = info->isLast;
wmmhello's avatar
wmmhello 已提交
2452 2453 2454
  info->cost.endTime = taosGetTimestampUs();
  info->cost.code = code;
  smlPrintStatisticInfo(info);
wmmhello's avatar
wmmhello 已提交
2455 2456
  smlDestroyInfo(info);

X
Xiaoyu Wang 已提交
2457
  if (isLast) {
wmmhello's avatar
wmmhello 已提交
2458
    tsem_post(&pParam->sem);
wmmhello's avatar
wmmhello 已提交
2459 2460 2461
  }
}

wmmhello's avatar
wmmhello 已提交
2462 2463 2464 2465 2466 2467 2468 2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482
/**
 * 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 已提交
2483
TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision) {
D
dapan1121 已提交
2484
  if (NULL == taos) {
2485 2486 2487
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
  }
D
dapan1121 已提交
2488

2489
  SRequestObj* request = (SRequestObj*)createRequest(*(int64_t*)taos, TSDB_SQL_INSERT);
wmmhello's avatar
wmmhello 已提交
2490
  if(!request){
2491
    uError("SML:taos_schemaless_insert error request is null");
2492
    return NULL;
wmmhello's avatar
wmmhello 已提交
2493 2494
  }

wmmhello's avatar
wmmhello 已提交
2495
  int    batchs = 0;
2496 2497
  STscObj* pTscObj = request->pTscObj;

2498
  pTscObj->schemalessType = 1;
wmmhello's avatar
wmmhello 已提交
2499
  SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
wmmhello's avatar
wmmhello 已提交
2500

wmmhello's avatar
wmmhello 已提交
2501 2502
  Params params;
  params.request = request;
wmmhello's avatar
wmmhello 已提交
2503 2504 2505
  tsem_init(&params.sem, 0, 0);
  taosThreadSpinInit(&(params.lock), 0);

X
Xiaoyu Wang 已提交
2506
  if (request->pDb == NULL) {
wmmhello's avatar
wmmhello 已提交
2507
    request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
wmmhello's avatar
wmmhello 已提交
2508
    smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
wmmhello's avatar
wmmhello 已提交
2509 2510 2511
    goto end;
  }

2512
  if(isSchemalessDb(pTscObj, request) != TSDB_CODE_SUCCESS){
wmmhello's avatar
wmmhello 已提交
2513
    request->code = TSDB_CODE_SML_INVALID_DB_CONF;
wmmhello's avatar
wmmhello 已提交
2514
    smlBuildInvalidDataMsg(&msg, "Cannot write data to a non schemaless database", NULL);
wmmhello's avatar
wmmhello 已提交
2515 2516 2517
    goto end;
  }

2518
  if (!lines) {
2519
    request->code = TSDB_CODE_SML_INVALID_DATA;
wmmhello's avatar
wmmhello 已提交
2520
    smlBuildInvalidDataMsg(&msg, "lines is null", NULL);
2521 2522 2523
    goto end;
  }

X
Xiaoyu Wang 已提交
2524
  if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
2525
    request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
wmmhello's avatar
wmmhello 已提交
2526
    smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
2527
    goto end;
wmmhello's avatar
wmmhello 已提交
2528 2529
  }

X
Xiaoyu Wang 已提交
2530 2531
  if (protocol == TSDB_SML_LINE_PROTOCOL &&
      (precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)) {
2532
    request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
wmmhello's avatar
wmmhello 已提交
2533
    smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
2534 2535 2536
    goto end;
  }

wmmhello's avatar
wmmhello 已提交
2537 2538 2539 2540 2541 2542 2543 2544
  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 已提交
2545 2546
  batchs = ceil(((double)numLines) / LINE_BATCH);
  for (int i = 0; i < batchs; ++i) {
2547
    SRequestObj* req = (SRequestObj*)createRequest(pTscObj->id, TSDB_SQL_INSERT);
wmmhello's avatar
wmmhello 已提交
2548 2549 2550 2551 2552
    if(!req){
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error request is null");
      goto end;
    }
2553
    SSmlHandle* info = smlBuildSmlInfo(pTscObj, req, (SMLProtocolType)protocol, precision);
wmmhello's avatar
wmmhello 已提交
2554 2555 2556 2557 2558 2559
    if(!info){
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error SSmlHandle is null");
      goto end;
    }

wmmhello's avatar
wmmhello 已提交
2560 2561
    int32_t perBatch = LINE_BATCH;

X
Xiaoyu Wang 已提交
2562
    if (numLines > perBatch) {
wmmhello's avatar
wmmhello 已提交
2563 2564
      numLines -= perBatch;
      info->isLast = false;
X
Xiaoyu Wang 已提交
2565
    } else {
wmmhello's avatar
wmmhello 已提交
2566 2567 2568 2569 2570
      perBatch = numLines;
      numLines = 0;
      info->isLast = true;
    }

wmmhello's avatar
wmmhello 已提交
2571
    info->params = &params;
wmmhello's avatar
wmmhello 已提交
2572 2573
    info->affectedRows = perBatch;
    info->pRequest->body.queryFp = smlInsertCallback;
X
Xiaoyu Wang 已提交
2574
    info->pRequest->body.param = info;
wmmhello's avatar
wmmhello 已提交
2575
    int32_t code = smlProcess(info, lines, perBatch);
wmmhello's avatar
wmmhello 已提交
2576
    lines += perBatch;
X
Xiaoyu Wang 已提交
2577
    if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2578 2579 2580 2581
      info->pRequest->body.queryFp(info, req, code);
    }
  }
  tsem_wait(&params.sem);
2582 2583

end:
wmmhello's avatar
wmmhello 已提交
2584 2585
  taosThreadSpinDestroy(&params.lock);
  tsem_destroy(&params.sem);
2586
//  ((STscObj *)taos)->schemalessType = 0;
2587
  pTscObj->schemalessType = 1;
2588
  uDebug("resultend:%s", request->msgBuf);
wmmhello's avatar
wmmhello 已提交
2589
  return (TAOS_RES*)request;
wmmhello's avatar
wmmhello 已提交
2590
}