clientSml.c 68.0 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

wmmhello's avatar
wmmhello 已提交
16 17 18 19 20
#include <ctype.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

21 22 23 24 25 26
#include "clientSml.h"

int64_t smlToMilli[3] = {3600000LL, 60000LL, 1000LL};
int64_t smlFactorNS[3] = {NANOSECOND_PER_MSEC, NANOSECOND_PER_USEC, 1};
int64_t smlFactorS[3] = {1000LL, 1000000LL, 1000000000LL};

wmmhello's avatar
wmmhello 已提交
27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99
//void *nodeListGet(NodeList *list, const void *key, int32_t len, _equal_fn_sml fn) {
//  NodeList *tmp = list;
//  while (tmp) {
//    if (fn == NULL) {
//      if (tmp->data.used && tmp->data.keyLen == len && memcmp(tmp->data.key, key, len) == 0) {
//        return tmp->data.value;
//      }
//    } else {
//      if (tmp->data.used && fn(tmp->data.key, key) == 0) {
//        return tmp->data.value;
//      }
//    }
//
//    tmp = tmp->next;
//  }
//  return NULL;
//}
//
//int nodeListSet(NodeList **list, const void *key, int32_t len, void *value, _equal_fn_sml fn) {
//  NodeList *tmp = *list;
//  while (tmp) {
//    if (!tmp->data.used) break;
//    if (fn == NULL) {
//      if (tmp->data.keyLen == len && memcmp(tmp->data.key, key, len) == 0) {
//        return -1;
//      }
//    } else {
//      if (tmp->data.keyLen == len && fn(tmp->data.key, key) == 0) {
//        return -1;
//      }
//    }
//
//    tmp = tmp->next;
//  }
//  if (tmp) {
//    tmp->data.key = key;
//    tmp->data.keyLen = len;
//    tmp->data.value = value;
//    tmp->data.used = true;
//  } else {
//    NodeList *newNode = (NodeList *)taosMemoryCalloc(1, sizeof(NodeList));
//    if (newNode == NULL) {
//      return -1;
//    }
//    newNode->data.key = key;
//    newNode->data.keyLen = len;
//    newNode->data.value = value;
//    newNode->data.used = true;
//    newNode->next = *list;
//    *list = newNode;
//  }
//  return 0;
//}
//
//int nodeListSize(NodeList *list) {
//  int cnt = 0;
//  while (list) {
//    if (list->data.used)
//      cnt++;
//    else
//      break;
//    list = list->next;
//  }
//  return cnt;
//}

static int32_t smlCheckAuth(SSmlHandle *info,  SRequestConnInfo* conn, const char* pTabName, AUTH_TYPE type){
  SUserAuthInfo pAuth = {0};
  snprintf(pAuth.user, sizeof(pAuth.user), "%s", info->taos->user);
  if (NULL == pTabName) {
    tNameSetDbName(&pAuth.tbName, info->taos->acctId, info->pRequest->pDb, strlen(info->pRequest->pDb));
  } else {
    toName(info->taos->acctId, info->pRequest->pDb, pTabName, &pAuth.tbName);
wmmhello's avatar
wmmhello 已提交
100
  }
wmmhello's avatar
wmmhello 已提交
101
  pAuth.type = type;
wmmhello's avatar
wmmhello 已提交
102

wmmhello's avatar
wmmhello 已提交
103 104
  int32_t      code = TSDB_CODE_SUCCESS;
  SUserAuthRes authRes = {0};
105

wmmhello's avatar
wmmhello 已提交
106
  code = catalogChkAuth(info->pCatalog, conn, &pAuth, &authRes);
wmmhello's avatar
wmmhello 已提交
107

wmmhello's avatar
wmmhello 已提交
108

wmmhello's avatar
wmmhello 已提交
109
  return (code == TSDB_CODE_SUCCESS) ? (authRes.pass ? TSDB_CODE_SUCCESS : TSDB_CODE_PAR_PERMISSION_DENIED) : code;
wmmhello's avatar
wmmhello 已提交
110 111

}
112
inline bool smlDoubleToInt64OverFlow(double num) {
X
Xiaoyu Wang 已提交
113
  if (num >= (double)INT64_MAX || num <= (double)INT64_MIN) return true;
114 115 116
  return false;
}

117 118 119 120 121 122 123 124
void smlStrReplace(char* src, int32_t len){
  for(int i = 0; i < len; i++){
    if(src[i] == '.'){
      src[i] = '_';
    }
  }
}

125
int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2) {
126
  if (pBuf->buf) {
127 128 129 130 131 132 133
    memset(pBuf->buf, 0, pBuf->len);
    if (msg1) strncat(pBuf->buf, msg1, pBuf->len);
    int32_t left = pBuf->len - strlen(pBuf->buf);
    if (left > 2 && msg2) {
      strncat(pBuf->buf, ":", left - 1);
      strncat(pBuf->buf, msg2, left - 2);
    }
wmmhello's avatar
wmmhello 已提交
134
  }
wmmhello's avatar
wmmhello 已提交
135 136 137
  return TSDB_CODE_SML_INVALID_DATA;
}

138 139 140 141 142
int64_t smlGetTimeValue(const char *value, int32_t len, uint8_t fromPrecision, uint8_t toPrecision) {
  char   *endPtr = NULL;
  int64_t tsInt64 = taosStr2Int64(value, &endPtr, 10);
  if (unlikely(value + len != endPtr)) {
    return -1;
143
  }
wmmhello's avatar
wmmhello 已提交
144

X
Xiaoyu Wang 已提交
145
  if (unlikely(fromPrecision >= TSDB_TIME_PRECISION_HOURS)) {
146
    int64_t unit = smlToMilli[fromPrecision - TSDB_TIME_PRECISION_HOURS];
D
dapan1121 已提交
147
    if (tsInt64 != 0 && unit > INT64_MAX / tsInt64) {
148 149 150 151
      return -1;
    }
    tsInt64 *= unit;
    fromPrecision = TSDB_TIME_PRECISION_MILLI;
wmmhello's avatar
wmmhello 已提交
152
  }
wmmhello's avatar
wmmhello 已提交
153

154
  return convertTimePrecision(tsInt64, fromPrecision, toPrecision);
wmmhello's avatar
wmmhello 已提交
155 156
}

157 158 159 160 161 162 163
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) {
    return TSDB_TIME_PRECISION_MILLI;
  } else {
    return -1;
164 165 166
  }
}

X
Xiaoyu Wang 已提交
167
SSmlTableInfo *smlBuildTableInfo(int numRows, const char *measure, int32_t measureLen) {
168 169 170
  SSmlTableInfo *tag = (SSmlTableInfo *)taosMemoryCalloc(sizeof(SSmlTableInfo), 1);
  if (!tag) {
    return NULL;
171 172
  }

173 174
  tag->sTableName = measure;
  tag->sTableNameLen = measureLen;
175

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

X
Xiaoyu Wang 已提交
182 183 184 185 186
  //  tag->tags = taosArrayInit(16, sizeof(SSmlKv));
  //  if (tag->tags == NULL) {
  //    uError("SML:smlBuildTableInfo failed to allocate memory");
  //    goto cleanup;
  //  }
187 188
  return tag;

X
Xiaoyu Wang 已提交
189
cleanup:
190 191
  taosMemoryFree(tag);
  return NULL;
192 193
}

194
static int32_t smlParseTableName(SArray *tags, char *childTableName) {
X
Xiaoyu Wang 已提交
195
  size_t childTableNameLen = strlen(tsSmlChildTableName);
196 197
  if (childTableNameLen <= 0) return TSDB_CODE_SUCCESS;

X
Xiaoyu Wang 已提交
198
  for (int i = 0; i < taosArrayGetSize(tags); i++) {
199 200 201 202 203
    SSmlKv *tag = (SSmlKv *)taosArrayGet(tags, i);
    // handle child table name
    if (childTableNameLen == tag->keyLen && strncmp(tag->key, tsSmlChildTableName, tag->keyLen) == 0) {
      memset(childTableName, 0, TSDB_TABLE_NAME_LEN);
      strncpy(childTableName, tag->value, (tag->length < TSDB_TABLE_NAME_LEN ? tag->length : TSDB_TABLE_NAME_LEN));
wmmhello's avatar
wmmhello 已提交
204
      taosArrayRemove(tags, i);
205
      break;
wmmhello's avatar
wmmhello 已提交
206 207
    }
  }
208

wmmhello's avatar
wmmhello 已提交
209 210 211
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
212
int32_t smlSetCTableName(SSmlTableInfo *oneTable) {
213
  smlParseTableName(oneTable->tags, oneTable->childTableName);
214

215
  if (strlen(oneTable->childTableName) == 0) {
X
Xiaoyu Wang 已提交
216
    SArray       *dst = taosArrayDup(oneTable->tags, NULL);
wmmhello's avatar
wmmhello 已提交
217
    RandTableName rName = {dst, oneTable->sTableName, (uint8_t)oneTable->sTableNameLen, oneTable->childTableName};
wmmhello's avatar
wmmhello 已提交
218

219 220
    buildChildTableName(&rName);
    taosArrayDestroy(dst);
221
  }
222 223
  return TSDB_CODE_SUCCESS;
}
224

225 226
void getTableUid(SSmlHandle *info, SSmlLineInfo *currElement, SSmlTableInfo *tinfo) {
  char   key[TSDB_TABLE_NAME_LEN * 2 + 1] = {0};
227 228 229
  size_t nLen = strlen(tinfo->childTableName);
  memcpy(key, currElement->measure, currElement->measureLen);
  memcpy(key + currElement->measureLen + 1, tinfo->childTableName, nLen);
230 231 232
  void *uid =
      taosHashGet(info->tableUids, key,
                  currElement->measureLen + 1 + nLen);  // use \0 as separator for stable name and child table name
233 234 235
  if (uid == NULL) {
    tinfo->uid = info->uid++;
    taosHashPut(info->tableUids, key, currElement->measureLen + 1 + nLen, &tinfo->uid, sizeof(uint64_t));
236 237
  } else {
    tinfo->uid = *(uint64_t *)uid;
238 239 240
  }
}

wmmhello's avatar
wmmhello 已提交
241 242 243 244 245 246
static void smlDestroySTableMeta(void *para) {
  SSmlSTableMeta *meta = *(SSmlSTableMeta**)para;
  taosHashCleanup(meta->tagHash);
  taosHashCleanup(meta->colHash);
  taosArrayDestroy(meta->tags);
  taosArrayDestroy(meta->cols);
wmmhello's avatar
wmmhello 已提交
247
  taosMemoryFreeClear(meta->tableMeta);
wmmhello's avatar
wmmhello 已提交
248 249 250
  taosMemoryFree(meta);
}

251 252 253 254
SSmlSTableMeta *smlBuildSTableMeta(bool isDataFormat) {
  SSmlSTableMeta *meta = (SSmlSTableMeta *)taosMemoryCalloc(sizeof(SSmlSTableMeta), 1);
  if (!meta) {
    return NULL;
255 256
  }

X
Xiaoyu Wang 已提交
257
  if (unlikely(!isDataFormat)) {
258 259 260 261 262
    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 已提交
263

264 265 266 267 268
    meta->colHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
    if (meta->colHash == NULL) {
      uError("SML:smlBuildSTableMeta failed to allocate memory");
      goto cleanup;
    }
wmmhello's avatar
wmmhello 已提交
269 270
  }

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

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

X
Xiaoyu Wang 已提交
284
cleanup:
wmmhello's avatar
wmmhello 已提交
285
  smlDestroySTableMeta(meta);
286
  return NULL;
wmmhello's avatar
wmmhello 已提交
287 288
}

X
Xiaoyu Wang 已提交
289 290 291 292 293 294 295
// uint16_t smlCalTypeSum(char* endptr, int32_t left){
//   uint16_t sum = 0;
//   for(int i = 0; i < left; i++){
//     sum += endptr[i];
//   }
//   return sum;
// }
wmmhello's avatar
wmmhello 已提交
296

X
Xiaoyu Wang 已提交
297 298 299
#define RETURN_FALSE                                 \
  smlBuildInvalidDataMsg(msg, "invalid data", pVal); \
  return false;
wmmhello's avatar
wmmhello 已提交
300

X
Xiaoyu Wang 已提交
301 302 303 304 305 306 307 308 309 310
#define SET_DOUBLE                     \
  kvVal->type = TSDB_DATA_TYPE_DOUBLE; \
  kvVal->d = result;

#define SET_FLOAT                                                                              \
  if (!IS_VALID_FLOAT(result)) {                                                               \
    smlBuildInvalidDataMsg(msg, "float out of range[-3.402823466e+38,3.402823466e+38]", pVal); \
    return false;                                                                              \
  }                                                                                            \
  kvVal->type = TSDB_DATA_TYPE_FLOAT;                                                          \
wmmhello's avatar
wmmhello 已提交
311 312
  kvVal->f = (float)result;

X
Xiaoyu Wang 已提交
313
#define SET_BIGINT                                                                                       \
wmmhello's avatar
wmmhello 已提交
314 315 316 317 318 319 320 321
  errno = 0;                                                                                             \
  int64_t tmp = taosStr2Int64(pVal, &endptr, 10);                                                        \
  if (errno == ERANGE) {                                                                                 \
    smlBuildInvalidDataMsg(msg, "big int out of range[-9223372036854775808,9223372036854775807]", pVal); \
    return false;                                                                                        \
  }                                                                                                      \
  kvVal->type = TSDB_DATA_TYPE_BIGINT;                                                                   \
  kvVal->i = tmp;
wmmhello's avatar
wmmhello 已提交
322

X
Xiaoyu Wang 已提交
323 324 325 326 327 328
#define SET_INT                                                                    \
  if (!IS_VALID_INT(result)) {                                                     \
    smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", pVal); \
    return false;                                                                  \
  }                                                                                \
  kvVal->type = TSDB_DATA_TYPE_INT;                                                \
wmmhello's avatar
wmmhello 已提交
329 330
  kvVal->i = result;

X
Xiaoyu Wang 已提交
331 332 333 334 335 336
#define SET_SMALL_INT                                                          \
  if (!IS_VALID_SMALLINT(result)) {                                            \
    smlBuildInvalidDataMsg(msg, "small int our of range[-32768,32767]", pVal); \
    return false;                                                              \
  }                                                                            \
  kvVal->type = TSDB_DATA_TYPE_SMALLINT;                                       \
wmmhello's avatar
wmmhello 已提交
337 338
  kvVal->i = result;

X
Xiaoyu Wang 已提交
339
#define SET_UBIGINT                                                                             \
wmmhello's avatar
wmmhello 已提交
340 341 342 343 344 345 346 347
  errno = 0;                                                                                    \
  uint64_t tmp = taosStr2UInt64(pVal, &endptr, 10);                                             \
  if (errno == ERANGE || result < 0) {                                                          \
    smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", pVal); \
    return false;                                                                               \
  }                                                                                             \
  kvVal->type = TSDB_DATA_TYPE_UBIGINT;                                                         \
  kvVal->u = tmp;
wmmhello's avatar
wmmhello 已提交
348

X
Xiaoyu Wang 已提交
349 350 351 352 353 354
#define SET_UINT                                                                  \
  if (!IS_VALID_UINT(result)) {                                                   \
    smlBuildInvalidDataMsg(msg, "unsigned int out of range[0,4294967295]", pVal); \
    return false;                                                                 \
  }                                                                               \
  kvVal->type = TSDB_DATA_TYPE_UINT;                                              \
wmmhello's avatar
wmmhello 已提交
355 356
  kvVal->u = result;

X
Xiaoyu Wang 已提交
357 358 359 360 361 362
#define SET_USMALL_INT                                                            \
  if (!IS_VALID_USMALLINT(result)) {                                              \
    smlBuildInvalidDataMsg(msg, "unsigned small int out of rang[0,65535]", pVal); \
    return false;                                                                 \
  }                                                                               \
  kvVal->type = TSDB_DATA_TYPE_USMALLINT;                                         \
wmmhello's avatar
wmmhello 已提交
363 364
  kvVal->u = result;

X
Xiaoyu Wang 已提交
365 366 367 368 369 370
#define SET_TINYINT                                                       \
  if (!IS_VALID_TINYINT(result)) {                                        \
    smlBuildInvalidDataMsg(msg, "tiny int out of range[-128,127]", pVal); \
    return false;                                                         \
  }                                                                       \
  kvVal->type = TSDB_DATA_TYPE_TINYINT;                                   \
wmmhello's avatar
wmmhello 已提交
371 372
  kvVal->i = result;

X
Xiaoyu Wang 已提交
373 374 375 376 377 378
#define SET_UTINYINT                                                            \
  if (!IS_VALID_UTINYINT(result)) {                                             \
    smlBuildInvalidDataMsg(msg, "unsigned tiny int out of range[0,255]", pVal); \
    return false;                                                               \
  }                                                                             \
  kvVal->type = TSDB_DATA_TYPE_UTINYINT;                                        \
wmmhello's avatar
wmmhello 已提交
379 380
  kvVal->u = result;

381
bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
382 383
  const char *pVal = kvVal->value;
  int32_t     len = kvVal->length;
X
Xiaoyu Wang 已提交
384
  char       *endptr = NULL;
wmmhello's avatar
wmmhello 已提交
385 386 387 388 389 390 391 392 393
  double      result = taosStr2Double(pVal, &endptr);
  if (pVal == endptr) {
    RETURN_FALSE
  }

  int32_t left = len - (endptr - pVal);
  if (left == 0) {
    SET_DOUBLE
  } else if (left == 3) {
X
Xiaoyu Wang 已提交
394 395
    if (endptr[0] == 'f' || endptr[0] == 'F') {
      if (endptr[1] == '6' && endptr[2] == '4') {
wmmhello's avatar
wmmhello 已提交
396
        SET_DOUBLE
X
Xiaoyu Wang 已提交
397
      } else if (endptr[1] == '3' && endptr[2] == '2') {
wmmhello's avatar
wmmhello 已提交
398
        SET_FLOAT
X
Xiaoyu Wang 已提交
399
      } else {
wmmhello's avatar
wmmhello 已提交
400 401
        RETURN_FALSE
      }
X
Xiaoyu Wang 已提交
402 403
    } else if (endptr[0] == 'i' || endptr[0] == 'I') {
      if (endptr[1] == '6' && endptr[2] == '4') {
wmmhello's avatar
wmmhello 已提交
404
        SET_BIGINT
X
Xiaoyu Wang 已提交
405
      } else if (endptr[1] == '3' && endptr[2] == '2') {
wmmhello's avatar
wmmhello 已提交
406
        SET_INT
X
Xiaoyu Wang 已提交
407
      } else if (endptr[1] == '1' && endptr[2] == '6') {
wmmhello's avatar
wmmhello 已提交
408
        SET_SMALL_INT
X
Xiaoyu Wang 已提交
409
      } else {
wmmhello's avatar
wmmhello 已提交
410 411
        RETURN_FALSE
      }
X
Xiaoyu Wang 已提交
412 413
    } else if (endptr[0] == 'u' || endptr[0] == 'U') {
      if (endptr[1] == '6' && endptr[2] == '4') {
wmmhello's avatar
wmmhello 已提交
414
        SET_UBIGINT
X
Xiaoyu Wang 已提交
415
      } else if (endptr[1] == '3' && endptr[2] == '2') {
wmmhello's avatar
wmmhello 已提交
416
        SET_UINT
X
Xiaoyu Wang 已提交
417
      } else if (endptr[1] == '1' && endptr[2] == '6') {
wmmhello's avatar
wmmhello 已提交
418
        SET_USMALL_INT
X
Xiaoyu Wang 已提交
419
      } else {
wmmhello's avatar
wmmhello 已提交
420 421
        RETURN_FALSE
      }
wmmhello's avatar
wmmhello 已提交
422
    } else {
wmmhello's avatar
wmmhello 已提交
423 424
      RETURN_FALSE
    }
X
Xiaoyu Wang 已提交
425 426 427
  } else if (left == 2) {
    if (endptr[0] == 'i' || endptr[0] == 'I') {
      if (endptr[1] == '8') {
wmmhello's avatar
wmmhello 已提交
428
        SET_TINYINT
X
Xiaoyu Wang 已提交
429
      } else {
wmmhello's avatar
wmmhello 已提交
430 431
        RETURN_FALSE
      }
X
Xiaoyu Wang 已提交
432
    } else if (endptr[0] == 'u' || endptr[0] == 'U') {
wmmhello's avatar
wmmhello 已提交
433 434 435 436 437
      if (endptr[1] == '8') {
        SET_UTINYINT
      } else {
        RETURN_FALSE
      }
X
Xiaoyu Wang 已提交
438
    } else {
wmmhello's avatar
wmmhello 已提交
439 440
      RETURN_FALSE
    }
X
Xiaoyu Wang 已提交
441 442
  } else if (left == 1) {
    if (endptr[0] == 'i' || endptr[0] == 'I') {
wmmhello's avatar
wmmhello 已提交
443
      SET_BIGINT
X
Xiaoyu Wang 已提交
444
    } else if (endptr[0] == 'u' || endptr[0] == 'U') {
wmmhello's avatar
wmmhello 已提交
445
      SET_UBIGINT
X
Xiaoyu Wang 已提交
446
    } else {
wmmhello's avatar
wmmhello 已提交
447 448 449 450 451 452 453 454 455
      RETURN_FALSE
    }
  } else {
    RETURN_FALSE;
  }
  return true;
}

bool smlParseNumberOld(SSmlKv *kvVal, SSmlMsgBuf *msg) {
wmmhello's avatar
wmmhello 已提交
456
  const char *pVal = kvVal->value;
X
Xiaoyu Wang 已提交
457 458 459 460
  int32_t     len = kvVal->length;
  char       *endptr = NULL;
  double      result = taosStr2Double(pVal, &endptr);
  if (pVal == endptr) {
461
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
462 463 464
    return false;
  }

465
  int32_t left = len - (endptr - pVal);
X
Xiaoyu Wang 已提交
466
  if (left == 0 || (left == 3 && strncasecmp(endptr, "f64", left) == 0)) {
467 468
    kvVal->type = TSDB_DATA_TYPE_DOUBLE;
    kvVal->d = result;
X
Xiaoyu Wang 已提交
469 470
  } else if ((left == 3 && strncasecmp(endptr, "f32", left) == 0)) {
    if (!IS_VALID_FLOAT(result)) {
471 472
      smlBuildInvalidDataMsg(msg, "float out of range[-3.402823466e+38,3.402823466e+38]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
473
    }
474 475
    kvVal->type = TSDB_DATA_TYPE_FLOAT;
    kvVal->f = (float)result;
X
Xiaoyu Wang 已提交
476 477
  } else if ((left == 1 && *endptr == 'i') || (left == 3 && strncasecmp(endptr, "i64", left) == 0)) {
    if (smlDoubleToInt64OverFlow(result)) {
wmmhello's avatar
wmmhello 已提交
478 479
      errno = 0;
      int64_t tmp = taosStr2Int64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
480
      if (errno == ERANGE) {
wmmhello's avatar
wmmhello 已提交
481 482 483 484 485 486
        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 已提交
487
    }
488
    kvVal->type = TSDB_DATA_TYPE_BIGINT;
wmmhello's avatar
wmmhello 已提交
489
    kvVal->i = (int64_t)result;
wmmhello's avatar
wmmhello 已提交
490
  } else if ((left == 1 && *endptr == 'u') || (left == 3 && strncasecmp(endptr, "u64", left) == 0)) {
X
Xiaoyu Wang 已提交
491
    if (result >= (double)UINT64_MAX || result < 0) {
wmmhello's avatar
wmmhello 已提交
492 493
      errno = 0;
      uint64_t tmp = taosStr2UInt64(pVal, &endptr, 10);
X
Xiaoyu Wang 已提交
494
      if (errno == ERANGE || result < 0) {
wmmhello's avatar
wmmhello 已提交
495 496 497 498 499 500
        smlBuildInvalidDataMsg(msg, "unsigned big int out of range[0,18446744073709551615]", pVal);
        return false;
      }
      kvVal->type = TSDB_DATA_TYPE_UBIGINT;
      kvVal->u = tmp;
      return true;
501
    }
502
    kvVal->type = TSDB_DATA_TYPE_UBIGINT;
wmmhello's avatar
wmmhello 已提交
503
    kvVal->u = result;
X
Xiaoyu Wang 已提交
504 505
  } else if (left == 3 && strncasecmp(endptr, "i32", left) == 0) {
    if (!IS_VALID_INT(result)) {
506 507
      smlBuildInvalidDataMsg(msg, "int out of range[-2147483648,2147483647]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
508
    }
509 510
    kvVal->type = TSDB_DATA_TYPE_INT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
511 512
  } else if (left == 3 && strncasecmp(endptr, "u32", left) == 0) {
    if (!IS_VALID_UINT(result)) {
513 514
      smlBuildInvalidDataMsg(msg, "unsigned int out of range[0,4294967295]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
515
    }
516 517
    kvVal->type = TSDB_DATA_TYPE_UINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
518 519
  } else if (left == 3 && strncasecmp(endptr, "i16", left) == 0) {
    if (!IS_VALID_SMALLINT(result)) {
520 521
      smlBuildInvalidDataMsg(msg, "small int our of range[-32768,32767]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
522
    }
523 524
    kvVal->type = TSDB_DATA_TYPE_SMALLINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
525 526
  } else if (left == 3 && strncasecmp(endptr, "u16", left) == 0) {
    if (!IS_VALID_USMALLINT(result)) {
527 528
      smlBuildInvalidDataMsg(msg, "unsigned small int out of rang[0,65535]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
529
    }
530 531
    kvVal->type = TSDB_DATA_TYPE_USMALLINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
532 533
  } else if (left == 2 && strncasecmp(endptr, "i8", left) == 0) {
    if (!IS_VALID_TINYINT(result)) {
534 535
      smlBuildInvalidDataMsg(msg, "tiny int out of range[-128,127]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
536
    }
537 538
    kvVal->type = TSDB_DATA_TYPE_TINYINT;
    kvVal->i = result;
X
Xiaoyu Wang 已提交
539 540
  } else if (left == 2 && strncasecmp(endptr, "u8", left) == 0) {
    if (!IS_VALID_UTINYINT(result)) {
541 542
      smlBuildInvalidDataMsg(msg, "unsigned tiny int out of range[0,255]", pVal);
      return false;
wmmhello's avatar
wmmhello 已提交
543
    }
544 545
    kvVal->type = TSDB_DATA_TYPE_UTINYINT;
    kvVal->u = result;
X
Xiaoyu Wang 已提交
546
  } else {
547
    smlBuildInvalidDataMsg(msg, "invalid data", pVal);
wmmhello's avatar
wmmhello 已提交
548 549
    return false;
  }
550
  return true;
wmmhello's avatar
wmmhello 已提交
551 552
}

X
Xiaoyu Wang 已提交
553
STableMeta *smlGetMeta(SSmlHandle *info, const void *measure, int32_t measureLen) {
554
  STableMeta *pTableMeta = NULL;
wmmhello's avatar
wmmhello 已提交
555

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

559 560 561 562 563 564 565
  SRequestConnInfo conn = {0};
  conn.pTrans = info->taos->pAppInfo->pTransporter;
  conn.requestId = info->pRequest->requestId;
  conn.requestObjRefId = info->pRequest->self;
  conn.mgmtEps = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
  memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
  memcpy(pName.tname, measure, measureLen);
wmmhello's avatar
wmmhello 已提交
566

567
  int32_t code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
X
Xiaoyu Wang 已提交
568
  if (code != TSDB_CODE_SUCCESS) {
569 570
    return NULL;
  }
571
  return pTableMeta;
wmmhello's avatar
wmmhello 已提交
572
}
wmmhello's avatar
wmmhello 已提交
573

574 575
static int64_t smlGenId() {
  static volatile int64_t linesSmlHandleId = 0;
wmmhello's avatar
wmmhello 已提交
576

577 578 579 580
  int64_t id = 0;
  do {
    id = atomic_add_fetch_64(&linesSmlHandleId, 1);
  } while (id == 0);
581

582
  return id;
583 584
}

585 586 587 588 589
static int32_t smlGenerateSchemaAction(SSchema *colField, SHashObj *colHash, SSmlKv *kv, bool isTag,
                                       ESchemaAction *action, SSmlHandle *info) {
  uint16_t *index = colHash ? (uint16_t *)taosHashGet(colHash, kv->key, kv->keyLen) : NULL;
  if (index) {
    if (colField[*index].type != kv->type) {
590 591
      uError("SML:0x%" PRIx64 " point type and db type mismatch. db type: %d, point type: %d, key: %s", info->id,
             colField[*index].type, kv->type, kv->key);
592
      return TSDB_CODE_SML_INVALID_DATA;
593 594
    }

D
Dingle Zhang 已提交
595
    if (((colField[*index].type == TSDB_DATA_TYPE_VARCHAR || colField[*index].type == TSDB_DATA_TYPE_GEOMETRY) &&
596 597 598 599 600 601 602 603 604 605 606 607 608 609 610
         (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))) {
      if (isTag) {
        *action = SCHEMA_ACTION_CHANGE_TAG_SIZE;
      } else {
        *action = SCHEMA_ACTION_CHANGE_COLUMN_SIZE;
      }
    }
  } else {
    if (isTag) {
      *action = SCHEMA_ACTION_ADD_TAG;
    } else {
      *action = SCHEMA_ACTION_ADD_COLUMN;
    }
wmmhello's avatar
wmmhello 已提交
611
  }
612
  return 0;
613 614
}

615
#define BOUNDARY 1024
616 617
static int32_t smlFindNearestPowerOf2(int32_t length, uint8_t type) {
  int32_t result = 1;
618
  if (length >= BOUNDARY) {
619
    result = length;
620
  } else {
621
    while (result <= length) {
K
kailixu 已提交
622
      result <<= 1;
623
    }
624
  }
K
kailixu 已提交
625

D
Dingle Zhang 已提交
626
  if ((type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_GEOMETRY) && result > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) {
627
    result = TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE;
K
kailixu 已提交
628 629
  } else if (type == TSDB_DATA_TYPE_NCHAR && result > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
    result = (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
630
  }
631 632 633

  if (type == TSDB_DATA_TYPE_NCHAR) {
    result = result * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
D
Dingle Zhang 已提交
634
  } else if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_GEOMETRY) {
635
    result = result + VARSTR_HEADER_SIZE;
636
  }
637
  return result;
638 639
}

640 641 642 643 644 645 646 647 648 649
static int32_t smlProcessSchemaAction(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
                                      ESchemaAction *action, bool isTag) {
  int32_t code = TSDB_CODE_SUCCESS;
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
    if (j == 0 && !isTag) continue;
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, j);
    code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, action, info);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
650
  }
651
  return TSDB_CODE_SUCCESS;
652 653
}

654 655 656 657 658
static int32_t smlCheckMeta(SSchema *schema, int32_t length, SArray *cols, bool isTag) {
  SHashObj *hashTmp = taosHashInit(length, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
  int32_t   i = 0;
  for (; i < length; i++) {
    taosHashPut(hashTmp, schema[i].name, strlen(schema[i].name), &i, SHORT_BYTES);
659 660
  }

661 662 663 664
  if (isTag) {
    i = 0;
  } else {
    i = 1;
665
  }
666 667 668 669 670 671
  for (; i < taosArrayGetSize(cols); i++) {
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
    if (taosHashGet(hashTmp, kv->key, kv->keyLen) == NULL) {
      taosHashCleanup(hashTmp);
      return -1;
    }
672
  }
673 674 675
  taosHashCleanup(hashTmp);
  return 0;
}
676

677
static int32_t getBytes(uint8_t type, int32_t length) {
D
Dingle Zhang 已提交
678
  if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_GEOMETRY) {
679 680 681 682
    return smlFindNearestPowerOf2(length, type);
  } else {
    return tDataTypes[type].bytes;
  }
683 684
}

685 686 687 688 689
static int32_t smlBuildFieldsList(SSmlHandle *info, SSchema *schemaField, SHashObj *schemaHash, SArray *cols,
                                  SArray *results, int32_t numOfCols, bool isTag) {
  for (int j = 0; j < taosArrayGetSize(cols); ++j) {
    SSmlKv       *kv = (SSmlKv *)taosArrayGet(cols, j);
    ESchemaAction action = SCHEMA_ACTION_NULL;
X
Xiaoyu Wang 已提交
690 691
    int           code = smlGenerateSchemaAction(schemaField, schemaHash, kv, isTag, &action, info);
    if (code != 0) {
D
dapan1121 已提交
692 693
      return code;
    }
694 695 696 697 698 699 700 701
    if (action == SCHEMA_ACTION_ADD_COLUMN || action == SCHEMA_ACTION_ADD_TAG) {
      SField field = {0};
      field.type = kv->type;
      field.bytes = getBytes(kv->type, kv->length);
      memcpy(field.name, kv->key, kv->keyLen);
      taosArrayPush(results, &field);
    } else if (action == SCHEMA_ACTION_CHANGE_COLUMN_SIZE || action == SCHEMA_ACTION_CHANGE_TAG_SIZE) {
      uint16_t *index = (uint16_t *)taosHashGet(schemaHash, kv->key, kv->keyLen);
X
Xiaoyu Wang 已提交
702
      if (index == NULL) {
D
dapan1121 已提交
703 704 705
        uError("smlBuildFieldsList get error, key:%s", kv->key);
        return TSDB_CODE_SML_INVALID_DATA;
      }
X
Xiaoyu Wang 已提交
706
      uint16_t newIndex = *index;
707 708 709
      if (isTag) newIndex -= numOfCols;
      SField *field = (SField *)taosArrayGet(results, newIndex);
      field->bytes = getBytes(kv->type, kv->length);
710 711
    }
  }
712 713 714 715 716 717 718 719

  int32_t maxLen = isTag ? TSDB_MAX_TAGS_LEN : TSDB_MAX_BYTES_PER_ROW;
  int32_t len = 0;
  for (int j = 0; j < taosArrayGetSize(results); ++j) {
    SField *field = taosArrayGet(results, j);
    len += field->bytes;
  }
  if(len > maxLen){
720
    return isTag ? TSDB_CODE_PAR_INVALID_TAGS_LENGTH : TSDB_CODE_PAR_INVALID_ROW_LENGTH;
721 722
  }

723 724 725
  return TSDB_CODE_SUCCESS;
}

726 727 728 729 730 731 732 733
// static int32_t smlSendMetaMsg(SSmlHandle *info, SName *pName, SSmlSTableMeta *sTableData,
//                               int32_t colVer, int32_t tagVer, int8_t source, uint64_t suid){
static int32_t smlSendMetaMsg(SSmlHandle *info, SName *pName, SArray *pColumns, SArray *pTags, STableMeta *pTableMeta,
                              ESchemaAction action) {
  SRequestObj   *pRequest = NULL;
  SMCreateStbReq pReq = {0};
  int32_t        code = TSDB_CODE_SUCCESS;
  SCmdMsgInfo    pCmdMsg = {0};
734
  char          *pSql = NULL;
735 736 737 738 739 740 741 742 743 744 745 746

  // put front for free
  pReq.numOfColumns = taosArrayGetSize(pColumns);
  pReq.pColumns = pColumns;
  pReq.numOfTags = taosArrayGetSize(pTags);
  pReq.pTags = pTags;

  if (action == SCHEMA_ACTION_CREATE_STABLE) {
    pReq.colVer = 1;
    pReq.tagVer = 1;
    pReq.suid = 0;
    pReq.source = TD_REQ_FROM_APP;
747
    pSql = "sml_create_stable";
748 749 750 751 752
  } else if (action == SCHEMA_ACTION_ADD_TAG || action == SCHEMA_ACTION_CHANGE_TAG_SIZE) {
    pReq.colVer = pTableMeta->sversion;
    pReq.tagVer = pTableMeta->tversion + 1;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
753
    pSql = (action == SCHEMA_ACTION_ADD_TAG) ? "sml_add_tag" : "sml_modify_tag_size";
754 755 756 757 758
  } else if (action == SCHEMA_ACTION_ADD_COLUMN || action == SCHEMA_ACTION_CHANGE_COLUMN_SIZE) {
    pReq.colVer = pTableMeta->sversion + 1;
    pReq.tagVer = pTableMeta->tversion;
    pReq.suid = pTableMeta->uid;
    pReq.source = TD_REQ_FROM_TAOX;
759
    pSql = (action == SCHEMA_ACTION_ADD_COLUMN) ? "sml_add_column" : "sml_modify_column_size";
wmmhello's avatar
wmmhello 已提交
760 761 762
  } else{
    uError("SML:0x%" PRIx64 " invalid action:%d", info->id, action);
    goto end;
763 764 765 766 767 768 769 770 771 772 773
  }

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

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

776 777 778 779
  if (pReq.numOfTags == 0) {
    pReq.numOfTags = 1;
    SField field = {0};
    field.type = TSDB_DATA_TYPE_NCHAR;
780
    field.bytes = TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE;
781 782
    strcpy(field.name, tsSmlTagName);
    taosArrayPush(pReq.pTags, &field);
783
  }
wmmhello's avatar
wmmhello 已提交
784

785 786 787
  pReq.commentLen = -1;
  pReq.igExists = true;
  tNameExtractFullName(pName, pReq.name);
788

789 790 791 792 793 794 795 796 797
  pCmdMsg.epSet = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
  pCmdMsg.msgType = TDMT_MND_CREATE_STB;
  pCmdMsg.msgLen = tSerializeSMCreateStbReq(NULL, 0, &pReq);
  pCmdMsg.pMsg = taosMemoryMalloc(pCmdMsg.msgLen);
  if (NULL == pCmdMsg.pMsg) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto end;
  }
  tSerializeSMCreateStbReq(pCmdMsg.pMsg, pCmdMsg.msgLen, &pReq);
798

799 800 801 802 803 804
  SQuery pQuery;
  memset(&pQuery, 0, sizeof(pQuery));
  pQuery.execMode = QUERY_EXEC_MODE_RPC;
  pQuery.pCmdMsg = &pCmdMsg;
  pQuery.msgType = pQuery.pCmdMsg->msgType;
  pQuery.stableQuery = true;
805

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

808 809
  if (pRequest->code == TSDB_CODE_SUCCESS) {
    catalogRemoveTableMeta(info->pCatalog, pName);
810
  }
811 812
  code = pRequest->code;
  taosMemoryFree(pCmdMsg.pMsg);
813

X
Xiaoyu Wang 已提交
814
end:
815 816 817 818
  destroyRequest(pRequest);
  tFreeSMCreateStbReq(&pReq);
  return code;
}
819

820
static int32_t smlModifyDBSchemas(SSmlHandle *info) {
X
Xiaoyu Wang 已提交
821 822
  uDebug("SML:0x%" PRIx64 " smlModifyDBSchemas start, format:%d, needModifySchema:%d", info->id, info->dataFormat,
         info->needModifySchema);
X
Xiaoyu Wang 已提交
823
  if (info->dataFormat && !info->needModifySchema) {
824
    return TSDB_CODE_SUCCESS;
825
  }
826 827 828
  int32_t     code = 0;
  SHashObj   *hashTmp = NULL;
  STableMeta *pTableMeta = NULL;
829 830 831 832 833 834 835 836 837

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

  SRequestConnInfo conn = {0};
  conn.pTrans = info->taos->pAppInfo->pTransporter;
  conn.requestId = info->pRequest->requestId;
  conn.requestObjRefId = info->pRequest->self;
  conn.mgmtEps = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
838

839
  SSmlSTableMeta **tmp = (SSmlSTableMeta **)taosHashIterate(info->superTables, NULL);
840
  while (tmp) {
841
    SSmlSTableMeta *sTableData = *tmp;
842
    bool            needCheckMeta = false;  // for multi thread
843

844 845
    size_t superTableLen = 0;
    void  *superTable = taosHashGetKey(tmp, &superTableLen);
846
    char  *measure = taosMemoryMalloc(superTableLen);
wmmhello's avatar
wmmhello 已提交
847 848
    memcpy(measure, superTable, superTableLen);
    PROCESS_SLASH_IN_MEASUREMENT(measure, superTableLen);
849
    smlStrReplace(measure, superTableLen);
850
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
wmmhello's avatar
wmmhello 已提交
851 852
    memcpy(pName.tname, measure, superTableLen);
    taosMemoryFree(measure);
853

854
    code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
855

856
    if (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_MND_STB_NOT_EXIST) {
wmmhello's avatar
wmmhello 已提交
857 858 859 860
      code = smlCheckAuth(info, &conn, NULL, AUTH_TYPE_WRITE);
      if(code != TSDB_CODE_SUCCESS){
        goto end;
      }
D
dapan1121 已提交
861
      uDebug("SML:0x%" PRIx64 " smlModifyDBSchemas create table:%s", info->id, pName.tname);
862 863
      SArray *pColumns = taosArrayInit(taosArrayGetSize(sTableData->cols), sizeof(SField));
      SArray *pTags = taosArrayInit(taosArrayGetSize(sTableData->tags), sizeof(SField));
D
dapan1121 已提交
864 865 866
      code = smlBuildFieldsList(info, NULL, NULL, sTableData->tags, pTags, 0, true);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " smlBuildFieldsList tag1 failed. %s", info->id, pName.tname);
867 868
        taosArrayDestroy(pColumns);
        taosArrayDestroy(pTags);
D
dapan1121 已提交
869 870 871 872 873
        goto end;
      }
      code = smlBuildFieldsList(info, NULL, NULL, sTableData->cols, pColumns, 0, false);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " smlBuildFieldsList col1 failed. %s", info->id, pName.tname);
874 875
        taosArrayDestroy(pColumns);
        taosArrayDestroy(pTags);
D
dapan1121 已提交
876 877
        goto end;
      }
878 879 880 881
      code = smlSendMetaMsg(info, &pName, pColumns, pTags, NULL, SCHEMA_ACTION_CREATE_STABLE);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, pName.tname);
        goto end;
882
      }
883 884
      info->cost.numOfCreateSTables++;
      taosMemoryFreeClear(pTableMeta);
885

886 887 888 889
      code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " catalogGetSTableMeta failed. super table name %s", info->id, pName.tname);
        goto end;
890
      }
891 892 893 894 895 896
    } else if (code == TSDB_CODE_SUCCESS) {
      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++) {
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
897
      }
898

899 900 901 902
      ESchemaAction action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->tags, &action, true);
      if (code != TSDB_CODE_SUCCESS) {
        goto end;
903
      }
904
      if (action != SCHEMA_ACTION_NULL) {
wmmhello's avatar
wmmhello 已提交
905 906 907 908
        code = smlCheckAuth(info, &conn, pName.tname, AUTH_TYPE_WRITE);
        if(code != TSDB_CODE_SUCCESS){
          goto end;
        }
X
Xiaoyu Wang 已提交
909 910
        uDebug("SML:0x%" PRIx64 " smlModifyDBSchemas change table tag, table:%s, action:%d", info->id, pName.tname,
               action);
911 912 913 914
        SArray *pColumns =
            taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
        SArray *pTags =
            taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
915

916 917 918 919 920 921 922 923 924
        for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
          SField field = {0};
          field.type = pTableMeta->schema[i].type;
          field.bytes = pTableMeta->schema[i].bytes;
          strcpy(field.name, pTableMeta->schema[i].name);
          if (i < pTableMeta->tableInfo.numOfColumns) {
            taosArrayPush(pColumns, &field);
          } else {
            taosArrayPush(pTags, &field);
925 926
          }
        }
D
dapan1121 已提交
927
        code = smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->tags, pTags,
X
Xiaoyu Wang 已提交
928
                                  pTableMeta->tableInfo.numOfColumns, true);
D
dapan1121 已提交
929 930
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " smlBuildFieldsList tag2 failed. %s", info->id, pName.tname);
931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947
          taosArrayDestroy(pColumns);
          taosArrayDestroy(pTags);
          goto end;
        }

        if (taosArrayGetSize(pTags) + pTableMeta->tableInfo.numOfColumns > TSDB_MAX_COLUMNS) {
          uError("SML:0x%" PRIx64 " too many columns than 4096", info->id);
          code = TSDB_CODE_PAR_TOO_MANY_COLUMNS;
          taosArrayDestroy(pColumns);
          taosArrayDestroy(pTags);
          goto end;
        }
        if (taosArrayGetSize(pTags) > TSDB_MAX_TAGS) {
          uError("SML:0x%" PRIx64 " too many tags than 128", info->id);
          code = TSDB_CODE_PAR_INVALID_TAGS_NUM;
          taosArrayDestroy(pColumns);
          taosArrayDestroy(pTags);
D
dapan1121 已提交
948 949
          goto end;
        }
950

951 952 953 954
        code = smlSendMetaMsg(info, &pName, pColumns, pTags, pTableMeta, action);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, pName.tname);
          goto end;
955
        }
956

957 958 959 960 961 962 963 964 965
        info->cost.numOfAlterTagSTables++;
        taosMemoryFreeClear(pTableMeta);
        code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
        if (code != TSDB_CODE_SUCCESS) {
          goto end;
        }
        code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
        if (code != TSDB_CODE_SUCCESS) {
          goto end;
wmmhello's avatar
wmmhello 已提交
966
        }
967 968
      }

969 970 971
      taosHashClear(hashTmp);
      for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns; i++) {
        taosHashPut(hashTmp, pTableMeta->schema[i].name, strlen(pTableMeta->schema[i].name), &i, SHORT_BYTES);
972
      }
973 974 975 976
      action = SCHEMA_ACTION_NULL;
      code = smlProcessSchemaAction(info, pTableMeta->schema, hashTmp, sTableData->cols, &action, false);
      if (code != TSDB_CODE_SUCCESS) {
        goto end;
977
      }
978
      if (action != SCHEMA_ACTION_NULL) {
wmmhello's avatar
wmmhello 已提交
979 980 981 982
        code = smlCheckAuth(info, &conn, pName.tname, AUTH_TYPE_WRITE);
        if(code != TSDB_CODE_SUCCESS){
          goto end;
        }
X
Xiaoyu Wang 已提交
983 984
        uDebug("SML:0x%" PRIx64 " smlModifyDBSchemas change table col, table:%s, action:%d", info->id, pName.tname,
               action);
985 986 987 988
        SArray *pColumns =
            taosArrayInit(taosArrayGetSize(sTableData->cols) + pTableMeta->tableInfo.numOfColumns, sizeof(SField));
        SArray *pTags =
            taosArrayInit(taosArrayGetSize(sTableData->tags) + pTableMeta->tableInfo.numOfTags, sizeof(SField));
989

990 991 992 993 994 995 996 997 998 999
        for (uint16_t i = 0; i < pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags; i++) {
          SField field = {0};
          field.type = pTableMeta->schema[i].type;
          field.bytes = pTableMeta->schema[i].bytes;
          strcpy(field.name, pTableMeta->schema[i].name);
          if (i < pTableMeta->tableInfo.numOfColumns) {
            taosArrayPush(pColumns, &field);
          } else {
            taosArrayPush(pTags, &field);
          }
1000 1001
        }

D
dapan1121 已提交
1002
        code = smlBuildFieldsList(info, pTableMeta->schema, hashTmp, sTableData->cols, pColumns,
X
Xiaoyu Wang 已提交
1003
                                  pTableMeta->tableInfo.numOfColumns, false);
D
dapan1121 已提交
1004 1005
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " smlBuildFieldsList col2 failed. %s", info->id, pName.tname);
1006 1007 1008 1009 1010 1011 1012 1013 1014 1015
          taosArrayDestroy(pColumns);
          taosArrayDestroy(pTags);
          goto end;
        }

        if (taosArrayGetSize(pColumns) + pTableMeta->tableInfo.numOfTags > TSDB_MAX_COLUMNS) {
          uError("SML:0x%" PRIx64 " too many columns than 4096", info->id);
          code = TSDB_CODE_PAR_TOO_MANY_COLUMNS;
          taosArrayDestroy(pColumns);
          taosArrayDestroy(pTags);
D
dapan1121 已提交
1016 1017
          goto end;
        }
1018

1019 1020 1021 1022
        code = smlSendMetaMsg(info, &pName, pColumns, pTags, pTableMeta, action);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " smlSendMetaMsg failed. can not create %s", info->id, pName.tname);
          goto end;
1023
        }
1024

1025 1026 1027 1028 1029 1030 1031 1032 1033 1034
        info->cost.numOfAlterColSTables++;
        taosMemoryFreeClear(pTableMeta);
        code = catalogRefreshTableMeta(info->pCatalog, &conn, &pName, -1);
        if (code != TSDB_CODE_SUCCESS) {
          goto end;
        }
        code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
        if (code != TSDB_CODE_SUCCESS) {
          uError("SML:0x%" PRIx64 " catalogGetSTableMeta failed. super table name %s", info->id, pName.tname);
          goto end;
1035 1036
        }
      }
wmmhello's avatar
wmmhello 已提交
1037

1038 1039 1040
      needCheckMeta = true;
      taosHashCleanup(hashTmp);
      hashTmp = NULL;
X
Xiaoyu Wang 已提交
1041
    } else {
1042 1043
      uError("SML:0x%" PRIx64 " load table meta error: %s", info->id, tstrerror(code));
      goto end;
wmmhello's avatar
wmmhello 已提交
1044
    }
1045

1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056
    if (needCheckMeta) {
      code = smlCheckMeta(&(pTableMeta->schema[pTableMeta->tableInfo.numOfColumns]), pTableMeta->tableInfo.numOfTags,
                          sTableData->tags, true);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " check tag failed. super table name %s", info->id, pName.tname);
        goto end;
      }
      code = smlCheckMeta(&(pTableMeta->schema[0]), pTableMeta->tableInfo.numOfColumns, sTableData->cols, false);
      if (code != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " check cols failed. super table name %s", info->id, pName.tname);
        goto end;
1057 1058
      }
    }
1059

wmmhello's avatar
wmmhello 已提交
1060
    taosMemoryFreeClear(sTableData->tableMeta);
1061
    sTableData->tableMeta = pTableMeta;
X
Xiaoyu Wang 已提交
1062
    uDebug("SML:0x%" PRIx64 "modify schema uid:%" PRIu64 ", sversion:%d, tversion:%d", info->id, pTableMeta->uid,
1063 1064
           pTableMeta->sversion, pTableMeta->tversion);
    tmp = (SSmlSTableMeta **)taosHashIterate(info->superTables, tmp);
wmmhello's avatar
wmmhello 已提交
1065
  }
X
Xiaoyu Wang 已提交
1066 1067
  uDebug("SML:0x%" PRIx64 " smlModifyDBSchemas end success, format:%d, needModifySchema:%d", info->id, info->dataFormat,
         info->needModifySchema);
D
dapan1121 已提交
1068

1069
  return 0;
1070

X
Xiaoyu Wang 已提交
1071
end:
1072 1073
  taosHashCleanup(hashTmp);
  taosMemoryFreeClear(pTableMeta);
D
dapan1121 已提交
1074
  catalogRefreshTableMeta(info->pCatalog, &conn, &pName, 1);
X
Xiaoyu Wang 已提交
1075 1076
  uError("SML:0x%" PRIx64 " smlModifyDBSchemas end failed:%d:%s, format:%d, needModifySchema:%d", info->id, code,
         tstrerror(code), info->dataFormat, info->needModifySchema);
D
dapan1121 已提交
1077

1078
  return code;
wmmhello's avatar
wmmhello 已提交
1079 1080
}

1081 1082 1083 1084 1085 1086 1087
/*
static int32_t smlCheckDupUnit(SHashObj *dumplicateKey, SArray *tags, SSmlMsgBuf *msg){
  for(int i = 0; i < taosArrayGetSize(tags); i++) {
    SSmlKv *tag = taosArrayGet(tags, i);
    if (smlCheckDuplicateKey(tag->key, tag->keyLen, dumplicateKey)) {
      smlBuildInvalidDataMsg(msg, "dumplicate key", tag->key);
      return TSDB_CODE_TSC_DUP_NAMES;
1088
    }
wmmhello's avatar
wmmhello 已提交
1089
  }
1090 1091 1092
  return TSDB_CODE_SUCCESS;
}

1093 1094 1095 1096 1097
static int32_t smlJudgeDupColName(SArray *cols, SArray *tags, SSmlMsgBuf *msg) {
  SHashObj *dumplicateKey = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  int ret = smlCheckDupUnit(dumplicateKey, cols, msg);
  if(ret != TSDB_CODE_SUCCESS){
    goto end;
1098
  }
1099 1100 1101
  ret = smlCheckDupUnit(dumplicateKey, tags, msg);
  if(ret != TSDB_CODE_SUCCESS){
    goto end;
1102
  }
1103

1104 1105 1106
  end:
  taosHashCleanup(dumplicateKey);
  return ret;
wmmhello's avatar
wmmhello 已提交
1107
}
1108
*/
wmmhello's avatar
wmmhello 已提交
1109

X
Xiaoyu Wang 已提交
1110
static void smlInsertMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols) {
1111 1112
  for (int16_t i = 0; i < taosArrayGetSize(cols); ++i) {
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
X
Xiaoyu Wang 已提交
1113 1114
    int     ret = taosHashPut(metaHash, kv->key, kv->keyLen, &i, SHORT_BYTES);
    if (ret == 0) {
1115 1116
      taosArrayPush(metaArray, kv);
    }
1117
  }
1118
}
wmmhello's avatar
wmmhello 已提交
1119

1120 1121 1122 1123 1124 1125 1126
static int32_t smlUpdateMeta(SHashObj *metaHash, SArray *metaArray, SArray *cols, bool isTag, SSmlMsgBuf *msg) {
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);

    int16_t *index = (int16_t *)taosHashGet(metaHash, kv->key, kv->keyLen);
    if (index) {
      SSmlKv *value = (SSmlKv *)taosArrayGet(metaArray, *index);
X
Xiaoyu Wang 已提交
1127
      if (isTag) {
1128 1129
        if (kv->length > value->length) {
          value->length = kv->length;
1130
        }
1131 1132 1133 1134 1135
        continue;
      }
      if (kv->type != value->type) {
        smlBuildInvalidDataMsg(msg, "the type is not the same like before", kv->key);
        return TSDB_CODE_SML_NOT_SAME_TYPE;
1136 1137
      }

1138 1139 1140 1141 1142
      if (IS_VAR_DATA_TYPE(kv->type) && (kv->length > value->length)) {  // update string len, if bigger
        value->length = kv->length;
      }
    } else {
      size_t tmp = taosArrayGetSize(metaArray);
X
Xiaoyu Wang 已提交
1143
      if (tmp > INT16_MAX) {
D
dapan1121 已提交
1144
        smlBuildInvalidDataMsg(msg, "too many cols or tags", kv->key);
1145
        uError("too many cols or tags");
D
dapan1121 已提交
1146
        return TSDB_CODE_SML_INVALID_DATA;
1147
      }
1148
      int16_t size = tmp;
X
Xiaoyu Wang 已提交
1149 1150
      int     ret = taosHashPut(metaHash, kv->key, kv->keyLen, &size, SHORT_BYTES);
      if (ret == 0) {
1151
        taosArrayPush(metaArray, kv);
1152 1153 1154 1155
      }
    }
  }

1156 1157
  return TSDB_CODE_SUCCESS;
}
1158

wmmhello's avatar
wmmhello 已提交
1159 1160
void smlDestroyTableInfo(void *para) {
  SSmlTableInfo *tag = *(SSmlTableInfo**)para;
1161 1162 1163 1164
  for (size_t i = 0; i < taosArrayGetSize(tag->cols); i++) {
    SHashObj *kvHash = (SHashObj *)taosArrayGetP(tag->cols, i);
    taosHashCleanup(kvHash);
  }
1165

X
Xiaoyu Wang 已提交
1166 1167 1168 1169 1170
  //  if (info->parseJsonByLib) {
  //    SSmlLineInfo *key = (SSmlLineInfo *)(tag->key);
  //    if (key != NULL) taosMemoryFree(key->tags);
  //  }
  //  taosMemoryFree(tag->key);
1171
  taosArrayDestroy(tag->cols);
wmmhello's avatar
wmmhello 已提交
1172
  taosArrayDestroyEx(tag->tags, freeSSmlKv);
1173 1174
  taosMemoryFree(tag);
}
1175

X
Xiaoyu Wang 已提交
1176
void clearColValArray(SArray *pCols) {
wmmhello's avatar
wmmhello 已提交
1177 1178
  int32_t num = taosArrayGetSize(pCols);
  for (int32_t i = 0; i < num; ++i) {
X
Xiaoyu Wang 已提交
1179
    SColVal *pCol = taosArrayGet(pCols, i);
wmmhello's avatar
wmmhello 已提交
1180 1181 1182 1183 1184 1185
    if (TSDB_DATA_TYPE_NCHAR == pCol->type) {
      taosMemoryFreeClear(pCol->value.pData);
    }
  }
}

1186 1187 1188 1189
void freeSSmlKv(void *data) {
  SSmlKv *kv = (SSmlKv *)data;
  if (kv->keyEscaped) taosMemoryFree((void *)(kv->key));
  if (kv->valueEscaped) taosMemoryFree((void *)(kv->value));
wmmhello's avatar
wmmhello 已提交
1190 1191
}

wmmhello's avatar
wmmhello 已提交
1192
void smlDestroyInfo(SSmlHandle *info) {
1193 1194
  if (!info) return;
  qDestroyQuery(info->pQuery);
1195

1196
  // destroy info->childTables
wmmhello's avatar
wmmhello 已提交
1197 1198 1199 1200 1201
//  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL);
//  while (oneTable) {
//    smlDestroyTableInfo(oneTable);
//    oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, oneTable);
//  }
1202

1203
  // destroy info->superTables
wmmhello's avatar
wmmhello 已提交
1204 1205 1206 1207 1208
//  SSmlSTableMeta **oneSTable = (SSmlSTableMeta **)taosHashIterate(info->superTables, NULL);
//  while (oneSTable) {
//    smlDestroySTableMeta(*oneSTable);
//    oneSTable = (SSmlSTableMeta **)taosHashIterate(info->superTables, oneSTable);
//  }
1209

1210 1211
  // destroy info->pVgHash
  taosHashCleanup(info->pVgHash);
1212 1213
  taosHashCleanup(info->childTables);
  taosHashCleanup(info->superTables);
1214
  taosHashCleanup(info->tableUids);
wmmhello's avatar
wmmhello 已提交
1215

X
Xiaoyu Wang 已提交
1216
  for (int i = 0; i < taosArrayGetSize(info->tagJsonArray); i++) {
wmmhello's avatar
wmmhello 已提交
1217 1218 1219 1220 1221
    cJSON *tags = (cJSON *)taosArrayGetP(info->tagJsonArray, i);
    cJSON_Delete(tags);
  }
  taosArrayDestroy(info->tagJsonArray);

1222 1223 1224 1225 1226 1227
  for (int i = 0; i < taosArrayGetSize(info->valueJsonArray); i++) {
    cJSON *value = (cJSON *)taosArrayGetP(info->valueJsonArray, i);
    cJSON_Delete(value);
  }
  taosArrayDestroy(info->valueJsonArray);

wmmhello's avatar
wmmhello 已提交
1228
  taosArrayDestroyEx(info->preLineTagKV, freeSSmlKv);
wmmhello's avatar
wmmhello 已提交
1229

X
Xiaoyu Wang 已提交
1230 1231
  if (!info->dataFormat) {
    for (int i = 0; i < info->lineNum; i++) {
wmmhello's avatar
wmmhello 已提交
1232
      taosArrayDestroyEx(info->lines[i].colArray, freeSSmlKv);
X
Xiaoyu Wang 已提交
1233
      if (info->parseJsonByLib) {
wmmhello's avatar
wmmhello 已提交
1234 1235
        taosMemoryFree(info->lines[i].tags);
      }
X
Xiaoyu Wang 已提交
1236
      if (info->lines[i].measureTagsLen != 0) taosMemoryFree(info->lines[i].measureTag);
1237 1238
    }
    taosMemoryFree(info->lines);
wmmhello's avatar
wmmhello 已提交
1239
  }
1240

wmmhello's avatar
wmmhello 已提交
1241
  cJSON_Delete(info->root);
1242 1243
  taosMemoryFreeClear(info);
}
wmmhello's avatar
wmmhello 已提交
1244

wmmhello's avatar
wmmhello 已提交
1245
SSmlHandle *smlBuildSmlInfo(TAOS *taos) {
1246 1247 1248 1249
  int32_t     code = TSDB_CODE_SUCCESS;
  SSmlHandle *info = (SSmlHandle *)taosMemoryCalloc(1, sizeof(SSmlHandle));
  if (NULL == info) {
    return NULL;
1250
  }
X
Xiaoyu Wang 已提交
1251
  if (taos != NULL) {
1252
    info->taos = acquireTscObj(*(int64_t *)taos);
X
Xiaoyu Wang 已提交
1253
    if (info->taos == NULL) {
D
dapan1121 已提交
1254 1255
      goto cleanup;
    }
1256 1257 1258 1259 1260
    code = catalogGetHandle(info->taos->pAppInfo->clusterId, &info->pCatalog);
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " get catalog error %d", info->id, code);
      goto cleanup;
    }
1261
  }
wmmhello's avatar
wmmhello 已提交
1262

1263
  info->pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
1264
  info->childTables = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
1265
  info->tableUids = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
1266
  info->superTables = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
wmmhello's avatar
wmmhello 已提交
1267 1268
  taosHashSetFreeFp(info->superTables, smlDestroySTableMeta);
  taosHashSetFreeFp(info->childTables, smlDestroyTableInfo);
1269

1270 1271 1272
  info->id = smlGenId();
  info->pQuery = smlInitHandle();
  info->dataFormat = true;
1273

wmmhello's avatar
wmmhello 已提交
1274
  info->tagJsonArray = taosArrayInit(8, POINTER_BYTES);
1275
  info->valueJsonArray = taosArrayInit(8, POINTER_BYTES);
1276 1277
  info->preLineTagKV = taosArrayInit(8, sizeof(SSmlKv));

1278
  if (NULL == info->pVgHash || NULL == info->childTables || NULL == info->superTables || NULL == info->tableUids) {
1279 1280
    uError("create SSmlHandle failed");
    goto cleanup;
1281 1282
  }

1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297
  return info;

cleanup:
  smlDestroyInfo(info);
  return NULL;
}

static int32_t smlPushCols(SArray *colsArray, SArray *cols) {
  SHashObj *kvHash = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (!kvHash) {
    uError("SML:smlDealCols failed to allocate memory");
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
    SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
1298
    terrno = 0;
1299
    taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
X
Xiaoyu Wang 已提交
1300
    if (terrno == TSDB_CODE_DUP_KEY) {
1301
      taosHashCleanup(kvHash);
X
Xiaoyu Wang 已提交
1302 1303
      return terrno;
    }
1304 1305
  }

1306
  taosArrayPush(colsArray, &kvHash);
1307
  return TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1308
}
1309

1310
static int32_t smlParseLineBottom(SSmlHandle *info) {
X
Xiaoyu Wang 已提交
1311 1312
  uDebug("SML:0x%" PRIx64 " smlParseLineBottom start, format:%d, linenum:%d", info->id, info->dataFormat,
         info->lineNum);
X
Xiaoyu Wang 已提交
1313
  if (info->dataFormat) return TSDB_CODE_SUCCESS;
1314

X
Xiaoyu Wang 已提交
1315 1316
  for (int32_t i = 0; i < info->lineNum; i++) {
    SSmlLineInfo  *elements = info->lines + i;
1317
    SSmlTableInfo *tinfo = NULL;
X
Xiaoyu Wang 已提交
1318
    if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
X
Xiaoyu Wang 已提交
1319 1320 1321
      SSmlTableInfo **tmp =
          (SSmlTableInfo **)taosHashGet(info->childTables, elements->measure, elements->measureTagsLen);
      if (tmp) tinfo = *tmp;
X
Xiaoyu Wang 已提交
1322
    } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
X
Xiaoyu Wang 已提交
1323
      SSmlTableInfo **tmp = (SSmlTableInfo **)taosHashGet(info->childTables, elements->measureTag,
wmmhello's avatar
wmmhello 已提交
1324
                                                          elements->measureLen + elements->tagsLen);
X
Xiaoyu Wang 已提交
1325
      if (tmp) tinfo = *tmp;
X
Xiaoyu Wang 已提交
1326
    } else {
X
Xiaoyu Wang 已提交
1327
      SSmlTableInfo **tmp = (SSmlTableInfo **)taosHashGet(info->childTables, elements->measureTag,
wmmhello's avatar
wmmhello 已提交
1328
                                                          elements->measureLen + elements->tagsLen);
X
Xiaoyu Wang 已提交
1329
      if (tmp) tinfo = *tmp;
1330 1331
    }

X
Xiaoyu Wang 已提交
1332
    if (tinfo == NULL) {
1333 1334 1335
      uError("SML:0x%" PRIx64 "get oneTable failed, line num:%d", info->id, i);
      smlBuildInvalidDataMsg(&info->msgBuf, "get oneTable failed", elements->measure);
      return TSDB_CODE_SML_INVALID_DATA;
1336
    }
wmmhello's avatar
wmmhello 已提交
1337

1338 1339 1340 1341 1342
    if (taosArrayGetSize(tinfo->tags) > TSDB_MAX_TAGS) {
      smlBuildInvalidDataMsg(&info->msgBuf, "too many tags than 128", NULL);
      return TSDB_CODE_PAR_INVALID_TAGS_NUM;
    }

1343 1344 1345
    if (taosArrayGetSize(elements->colArray) + taosArrayGetSize(tinfo->tags) > TSDB_MAX_COLUMNS) {
      smlBuildInvalidDataMsg(&info->msgBuf, "too many columns than 4096", NULL);
      return TSDB_CODE_PAR_TOO_MANY_COLUMNS;
wmmhello's avatar
wmmhello 已提交
1346
    }
wmmhello's avatar
wmmhello 已提交
1347

1348
    int ret = smlPushCols(tinfo->cols, elements->colArray);
X
Xiaoyu Wang 已提交
1349
    if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1350 1351 1352
      return ret;
    }

1353 1354
    SSmlSTableMeta **tableMeta =
        (SSmlSTableMeta **)taosHashGet(info->superTables, elements->measure, elements->measureLen);
1355
    if (tableMeta) {  // update meta
X
Xiaoyu Wang 已提交
1356 1357
      uDebug("SML:0x%" PRIx64 " smlParseLineBottom update meta, format:%d, linenum:%d", info->id, info->dataFormat,
             info->lineNum);
1358
      ret = smlUpdateMeta((*tableMeta)->colHash, (*tableMeta)->cols, elements->colArray, false, &info->msgBuf);
1359
      if (ret == TSDB_CODE_SUCCESS) {
1360
        ret = smlUpdateMeta((*tableMeta)->tagHash, (*tableMeta)->tags, tinfo->tags, true, &info->msgBuf);
1361 1362 1363 1364 1365
      }
      if (ret != TSDB_CODE_SUCCESS) {
        uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
        return ret;
      }
X
Xiaoyu Wang 已提交
1366
    } else {
X
Xiaoyu Wang 已提交
1367 1368 1369 1370 1371
      //      ret = smlJudgeDupColName(elements->colArray, tinfo->tags, &info->msgBuf);
      //      if (ret != TSDB_CODE_SUCCESS) {
      //        uError("SML:0x%" PRIx64 " smlUpdateMeta failed", info->id);
      //        return ret;
      //      }
X
Xiaoyu Wang 已提交
1372 1373
      uDebug("SML:0x%" PRIx64 " smlParseLineBottom add meta, format:%d, linenum:%d", info->id, info->dataFormat,
             info->lineNum);
1374
      SSmlSTableMeta *meta = smlBuildSTableMeta(info->dataFormat);
wmmhello's avatar
wmmhello 已提交
1375 1376 1377
      if(meta == NULL){
        return TSDB_CODE_OUT_OF_MEMORY;
      }
1378
      taosHashPut(info->superTables, elements->measure, elements->measureLen, &meta, POINTER_BYTES);
1379
      terrno = 0;
1380
      smlInsertMeta(meta->tagHash, meta->tags, tinfo->tags);
X
Xiaoyu Wang 已提交
1381 1382 1383
      if (terrno == TSDB_CODE_DUP_KEY) {
        return terrno;
      }
1384
      smlInsertMeta(meta->colHash, meta->cols, elements->colArray);
wmmhello's avatar
wmmhello 已提交
1385
    }
wmmhello's avatar
wmmhello 已提交
1386
  }
D
dapan1121 已提交
1387
  uDebug("SML:0x%" PRIx64 " smlParseLineBottom end, format:%d, linenum:%d", info->id, info->dataFormat, info->lineNum);
1388

wmmhello's avatar
wmmhello 已提交
1389 1390 1391
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1392
static int32_t smlInsertData(SSmlHandle *info) {
wmmhello's avatar
wmmhello 已提交
1393
  int32_t code = TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1394
  uDebug("SML:0x%" PRIx64 " smlInsertData start, format:%d", info->id, info->dataFormat);
wmmhello's avatar
wmmhello 已提交
1395

X
Xiaoyu Wang 已提交
1396
  if (info->pRequest->dbList == NULL) {
wmmhello's avatar
wmmhello 已提交
1397 1398
    info->pRequest->dbList = taosArrayInit(1, TSDB_DB_FNAME_LEN);
  }
1399
  char *data = (char *)taosArrayReserve(info->pRequest->dbList, 1);
1400 1401 1402
  SName pName = {TSDB_TABLE_NAME_T, info->taos->acctId, {0}, {0}};
  tstrncpy(pName.dbname, info->pRequest->pDb, sizeof(pName.dbname));
  tNameGetFullDbName(&pName, data);
wmmhello's avatar
wmmhello 已提交
1403

1404 1405 1406
  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL);
  while (oneTable) {
    SSmlTableInfo *tableData = *oneTable;
1407 1408 1409 1410 1411 1412

    int   measureLen = tableData->sTableNameLen;
    char *measure = (char *)taosMemoryMalloc(tableData->sTableNameLen);
    memcpy(measure, tableData->sTableName, tableData->sTableNameLen);
    PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen);
    smlStrReplace(measure, measureLen);
wmmhello's avatar
wmmhello 已提交
1413 1414
    memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
    memcpy(pName.tname, measure, measureLen);
D
dapan1121 已提交
1415

X
Xiaoyu Wang 已提交
1416
    if (info->pRequest->tableList == NULL) {
wmmhello's avatar
wmmhello 已提交
1417 1418 1419 1420
      info->pRequest->tableList = taosArrayInit(1, sizeof(SName));
    }
    taosArrayPush(info->pRequest->tableList, &pName);

wmmhello's avatar
wmmhello 已提交
1421
    strcpy(pName.tname, tableData->childTableName);
1422

D
dapan1121 已提交
1423 1424 1425 1426 1427
    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 已提交
1428

wmmhello's avatar
wmmhello 已提交
1429 1430
    code = smlCheckAuth(info, &conn, pName.tname, AUTH_TYPE_WRITE);
    if(code != TSDB_CODE_SUCCESS){
1431
      taosMemoryFree(measure);
wmmhello's avatar
wmmhello 已提交
1432 1433 1434
      return code;
    }

wmmhello's avatar
wmmhello 已提交
1435
    SVgroupInfo vg;
D
dapan1121 已提交
1436
    code = catalogGetTableHashVgroup(info->pCatalog, &conn, &pName, &vg);
1437
    if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
1438
      uError("SML:0x%" PRIx64 " catalogGetTableHashVgroup failed. table name: %s", info->id, tableData->childTableName);
1439
      taosMemoryFree(measure);
wmmhello's avatar
wmmhello 已提交
1440 1441
      return code;
    }
X
Xiaoyu Wang 已提交
1442
    taosHashPut(info->pVgHash, (const char *)&vg.vgId, sizeof(vg.vgId), (char *)&vg, sizeof(vg));
wmmhello's avatar
wmmhello 已提交
1443

1444 1445 1446
    SSmlSTableMeta **pMeta =
        (SSmlSTableMeta **)taosHashGet(info->superTables, tableData->sTableName, tableData->sTableNameLen);
    if (unlikely(NULL == pMeta || NULL == (*pMeta)->tableMeta)) {
1447
      uError("SML:0x%" PRIx64 " NULL == pMeta. table name: %s", info->id, tableData->childTableName);
1448
      taosMemoryFree(measure);
1449 1450
      return TSDB_CODE_SML_INTERNAL_ERROR;
    }
wmmhello's avatar
wmmhello 已提交
1451

1452
    // use tablemeta of stable to save vgid and uid of child table
1453 1454
    (*pMeta)->tableMeta->vgId = vg.vgId;
    (*pMeta)->tableMeta->uid = tableData->uid;  // one table merge data block together according uid
X
Xiaoyu Wang 已提交
1455 1456
    uDebug("SML:0x%" PRIx64 " smlInsertData table:%s, uid:%" PRIu64 ", format:%d", info->id, pName.tname,
           tableData->uid, info->dataFormat);
wmmhello's avatar
wmmhello 已提交
1457

X
Xiaoyu Wang 已提交
1458
    code = smlBindData(info->pQuery, info->dataFormat, tableData->tags, (*pMeta)->cols, tableData->cols,
1459 1460
                       (*pMeta)->tableMeta, tableData->childTableName, measure, measureLen, info->ttl, info->msgBuf.buf,
                       info->msgBuf.len);
1461
    taosMemoryFree(measure);
X
Xiaoyu Wang 已提交
1462 1463
    if (code != TSDB_CODE_SUCCESS) {
      uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
wmmhello's avatar
wmmhello 已提交
1464 1465
      return code;
    }
1466
    oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, oneTable);
wmmhello's avatar
wmmhello 已提交
1467
  }
wmmhello's avatar
wmmhello 已提交
1468

wmmhello's avatar
wmmhello 已提交
1469
  code = smlBuildOutput(info->pQuery, info->pVgHash);
1470
  if (code != TSDB_CODE_SUCCESS) {
X
Xiaoyu Wang 已提交
1471
    uError("SML:0x%" PRIx64 " smlBuildOutput failed", info->id);
1472 1473
    return code;
  }
1474 1475
  info->cost.insertRpcTime = taosGetTimestampUs();

1476 1477 1478
  SAppClusterSummary *pActivity = &info->taos->pAppInfo->summary;
  atomic_add_fetch_64((int64_t *)&pActivity->numOfInsertsReq, 1);

wmmhello's avatar
wmmhello 已提交
1479
  launchQueryImpl(info->pRequest, info->pQuery, true, NULL);
X
Xiaoyu Wang 已提交
1480 1481
  uDebug("SML:0x%" PRIx64 " smlInsertData end, format:%d, code:%d,%s", info->id, info->dataFormat, info->pRequest->code,
         tstrerror(info->pRequest->code));
D
dapan1121 已提交
1482

wmmhello's avatar
wmmhello 已提交
1483
  return info->pRequest->code;
wmmhello's avatar
wmmhello 已提交
1484 1485
}

X
Xiaoyu Wang 已提交
1486
static void smlPrintStatisticInfo(SSmlHandle *info) {
1487
  uDebug(
X
Xiaoyu Wang 已提交
1488
      "SML:0x%" PRIx64
D
dapan1121 已提交
1489
      " smlInsertLines result, code:%d, msg:%s, lineNum:%d,stable num:%d,ctable num:%d,create stable num:%d,alter stable tag num:%d,alter stable col num:%d \
X
Xiaoyu Wang 已提交
1490
        parse cost:%" PRId64 ",schema cost:%" PRId64 ",bind cost:%" PRId64 ",rpc cost:%" PRId64 ",total cost:%" PRId64
X
Xiaoyu Wang 已提交
1491
      "",
X
Xiaoyu Wang 已提交
1492 1493 1494 1495 1496
      info->id, info->cost.code, tstrerror(info->cost.code), info->cost.lineNum, info->cost.numOfSTables,
      info->cost.numOfCTables, info->cost.numOfCreateSTables, info->cost.numOfAlterTagSTables,
      info->cost.numOfAlterColSTables, 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);
1497 1498
}

X
Xiaoyu Wang 已提交
1499
int32_t smlClearForRerun(SSmlHandle *info) {
1500 1501
  info->reRun = false;
  // clear info->childTables
wmmhello's avatar
wmmhello 已提交
1502 1503 1504 1505 1506
//  SSmlTableInfo **oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, NULL);
//  while (oneTable) {
//    smlDestroyTableInfo(info, *oneTable);
//    oneTable = (SSmlTableInfo **)taosHashIterate(info->childTables, oneTable);
//  }
1507 1508

  // clear info->superTables
wmmhello's avatar
wmmhello 已提交
1509 1510 1511 1512 1513
//  SSmlSTableMeta **oneSTable = (SSmlSTableMeta **)taosHashIterate(info->superTables, NULL);
//  while (oneSTable) {
//    smlDestroySTableMeta(*oneSTable);
//    oneSTable = (SSmlSTableMeta **)taosHashIterate(info->superTables, oneSTable);
//  }
1514

wmmhello's avatar
wmmhello 已提交
1515 1516
  taosHashClear(info->childTables);
  taosHashClear(info->superTables);
1517
  taosHashClear(info->tableUids);
wmmhello's avatar
wmmhello 已提交
1518

X
Xiaoyu Wang 已提交
1519
  if (!info->dataFormat) {
wmmhello's avatar
wmmhello 已提交
1520 1521 1522 1523 1524
    if (unlikely(info->lines != NULL)) {
      uError("SML:0x%" PRIx64 " info->lines != NULL", info->id);
      return TSDB_CODE_SML_INVALID_DATA;
    }
    info->lines = (SSmlLineInfo *)taosMemoryCalloc(info->lineNum, sizeof(SSmlLineInfo));
1525 1526 1527 1528 1529 1530
  }

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

X
Xiaoyu Wang 已提交
1531
  SVnodeModifyOpStmt *stmt = (SVnodeModifyOpStmt *)(info->pQuery->pRoot);
1532 1533 1534 1535 1536
  stmt->freeHashFunc(stmt->pTableBlockHashObj);
  stmt->pTableBlockHashObj = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
1537
static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char *rawLineEnd, int numLines) {
D
dapan1121 已提交
1538
  uDebug("SML:0x%" PRIx64 " smlParseLine start", info->id);
wmmhello's avatar
wmmhello 已提交
1539
  int32_t code = TSDB_CODE_SUCCESS;
1540
  if (info->protocol == TSDB_SML_JSON_PROTOCOL) {
dengyihao's avatar
dengyihao 已提交
1541
    if (lines) {
wmmhello's avatar
wmmhello 已提交
1542
      code = smlParseJSON(info, *lines);
dengyihao's avatar
dengyihao 已提交
1543
    } else if (rawLine) {
wmmhello's avatar
wmmhello 已提交
1544 1545
      code = smlParseJSON(info, rawLine);
    }
1546
    if (code != TSDB_CODE_SUCCESS) {
dengyihao's avatar
dengyihao 已提交
1547
      uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, lines ? *lines : rawLine);
1548 1549
      return code;
    }
wmmhello's avatar
wmmhello 已提交
1550
    return code;
wmmhello's avatar
wmmhello 已提交
1551
  }
wmmhello's avatar
wmmhello 已提交
1552

X
Xiaoyu Wang 已提交
1553
  char   *oldRaw = rawLine;
1554 1555
  int32_t i = 0;
  while (i < numLines) {
wmmhello's avatar
wmmhello 已提交
1556
    char *tmp = NULL;
dengyihao's avatar
dengyihao 已提交
1557 1558
    int   len = 0;
    if (lines) {
wmmhello's avatar
wmmhello 已提交
1559 1560
      tmp = lines[i];
      len = strlen(tmp);
dengyihao's avatar
dengyihao 已提交
1561
    } else if (rawLine) {
wmmhello's avatar
wmmhello 已提交
1562
      tmp = rawLine;
dengyihao's avatar
dengyihao 已提交
1563 1564
      while (rawLine < rawLineEnd) {
        if (*(rawLine++) == '\n') {
wmmhello's avatar
wmmhello 已提交
1565 1566 1567 1568
          break;
        }
        len++;
      }
dengyihao's avatar
dengyihao 已提交
1569
      if (info->protocol == TSDB_SML_LINE_PROTOCOL && tmp[0] == '#') {  // this line is comment
wmmhello's avatar
wmmhello 已提交
1570 1571
        continue;
      }
wmmhello's avatar
wmmhello 已提交
1572 1573
    }

X
Xiaoyu Wang 已提交
1574
    uDebug("SML:0x%" PRIx64 " smlParseLine israw:%d, numLines:%d, protocol:%d, len:%d, sql:%s", info->id,
1575
           info->isRawLine, numLines, info->protocol, len, info->isRawLine ? "rawdata" : tmp);
1576

X
Xiaoyu Wang 已提交
1577
    if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
X
Xiaoyu Wang 已提交
1578
      if (info->dataFormat) {
wmmhello's avatar
wmmhello 已提交
1579 1580
        SSmlLineInfo element = {0};
        code = smlParseInfluxString(info, tmp, tmp + len, &element);
X
Xiaoyu Wang 已提交
1581
      } else {
wmmhello's avatar
wmmhello 已提交
1582 1583
        code = smlParseInfluxString(info, tmp, tmp + len, info->lines + i);
      }
X
Xiaoyu Wang 已提交
1584
    } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) {
X
Xiaoyu Wang 已提交
1585
      if (info->dataFormat) {
1586 1587
        SSmlLineInfo element = {0};
        code = smlParseTelnetString(info, (char *)tmp, (char *)tmp + len, &element);
X
Xiaoyu Wang 已提交
1588
        if (element.measureTagsLen != 0) taosMemoryFree(element.measureTag);
X
Xiaoyu Wang 已提交
1589
      } else {
1590 1591
        code = smlParseTelnetString(info, (char *)tmp, (char *)tmp + len, info->lines + i);
      }
X
Xiaoyu Wang 已提交
1592
    } else {
1593
      code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
1594
    }
wmmhello's avatar
wmmhello 已提交
1595
    if (code != TSDB_CODE_SUCCESS) {
1596
      uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, info->isRawLine ? "rawdata" : tmp);
1597
      return code;
wmmhello's avatar
wmmhello 已提交
1598
    }
X
Xiaoyu Wang 已提交
1599
    if (info->reRun) {
D
dapan1121 已提交
1600
      uDebug("SML:0x%" PRIx64 " smlParseLine re run", info->id);
1601
      i = 0;
1602
      rawLine = oldRaw;
1603
      code = smlClearForRerun(info);
X
Xiaoyu Wang 已提交
1604
      if (code != TSDB_CODE_SUCCESS) {
1605
        return code;
1606
      }
1607
      continue;
1608
    }
1609
    i++;
wmmhello's avatar
wmmhello 已提交
1610
  }
D
dapan1121 已提交
1611
  uDebug("SML:0x%" PRIx64 " smlParseLine end", info->id);
1612

1613 1614 1615
  return code;
}

dengyihao's avatar
dengyihao 已提交
1616
static int smlProcess(SSmlHandle *info, char *lines[], char *rawLine, char *rawLineEnd, int numLines) {
1617
  int32_t code = TSDB_CODE_SUCCESS;
1618 1619
  int32_t retryNum = 0;

1620 1621
  info->cost.parseTime = taosGetTimestampUs();

wmmhello's avatar
wmmhello 已提交
1622
  code = smlParseLine(info, lines, rawLine, rawLineEnd, numLines);
1623
  if (code != 0) {
X
Xiaoyu Wang 已提交
1624
    uError("SML:0x%" PRIx64 " smlParseLine error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
1625
    return code;
1626
  }
1627 1628 1629 1630 1631 1632
  code = smlParseLineBottom(info);
  if (code != 0) {
    uError("SML:0x%" PRIx64 " smlParseLineBottom error : %s", info->id, tstrerror(code));
    return code;
  }

1633
  info->cost.lineNum = info->lineNum;
1634 1635
  info->cost.numOfSTables = taosHashGetSize(info->superTables);
  info->cost.numOfCTables = taosHashGetSize(info->childTables);
1636 1637

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

X
Xiaoyu Wang 已提交
1639
  do {
1640
    code = smlModifyDBSchemas(info);
wmmhello's avatar
wmmhello 已提交
1641
    if (code != TSDB_CODE_TDB_INVALID_TABLE_SCHEMA_VER && code != TSDB_CODE_SDB_OBJ_CREATING && code != TSDB_CODE_MND_TRANS_CONFLICT) {
1642 1643
      break;
    }
1644
    taosMsleep(100);
D
dapan1121 已提交
1645
    uInfo("SML:0x%" PRIx64 " smlModifyDBSchemas retry code:%s, times:%d", info->id, tstrerror(code), retryNum);
1646
  } while (retryNum++ < taosHashGetSize(info->superTables) * MAX_RETRY_TIMES);
1647

wmmhello's avatar
wmmhello 已提交
1648
  if (code != 0) {
X
Xiaoyu Wang 已提交
1649
    uError("SML:0x%" PRIx64 " smlModifyDBSchemas error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
1650
    return code;
wmmhello's avatar
wmmhello 已提交
1651
  }
wmmhello's avatar
wmmhello 已提交
1652

1653
  info->cost.insertBindTime = taosGetTimestampUs();
wmmhello's avatar
wmmhello 已提交
1654 1655
  code = smlInsertData(info);
  if (code != 0) {
X
Xiaoyu Wang 已提交
1656
    uError("SML:0x%" PRIx64 " smlInsertData error : %s", info->id, tstrerror(code));
wmmhello's avatar
wmmhello 已提交
1657
    return code;
wmmhello's avatar
wmmhello 已提交
1658 1659 1660 1661 1662
  }

  return code;
}

1663 1664 1665 1666 1667
void smlSetReqSQL(SRequestObj *request, char *lines[], char *rawLine, char *rawLineEnd) {
  if (tsSlowLogScope & SLOW_LOG_TYPE_INSERT) {
    int32_t len = 0;
    int32_t rlen = 0;
    char* p = NULL;
1668

1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683
    if (lines && lines[0]) {
      len = strlen(lines[0]);
      p = lines[0];
    } else if (rawLine) {
      if (rawLineEnd) {
        len = rawLineEnd - rawLine;
      } else {
        len = strlen(rawLine);
      }
      p = rawLine;
    }

    if (NULL == p) {
      return;
    }
1684

1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696 1697 1698 1699 1700
    rlen = TMIN(len, TSDB_MAX_ALLOWED_SQL_LEN);
    rlen = TMAX(rlen, 0);

    char *sql = taosMemoryMalloc(rlen + 1);
    if (NULL == sql) {
      uError("malloc %d for sml sql failed", rlen + 1);
      return;
    }
    memcpy(sql, p, rlen);
    sql[rlen] = 0;

    request->sqlstr = sql;
    request->sqlLen = rlen;
  }
}

X
Xiaoyu Wang 已提交
1701 1702
TAOS_RES *taos_schemaless_insert_inner(TAOS *taos, char *lines[], char *rawLine, char *rawLineEnd, int numLines,
                                       int protocol, int precision, int32_t ttl, int64_t reqid) {
1703
  int32_t code = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1704 1705 1706
  if (NULL == taos) {
    terrno = TSDB_CODE_TSC_DISCONNECTED;
    return NULL;
1707
  }
wmmhello's avatar
wmmhello 已提交
1708
  SRequestObj *request = NULL;
X
Xiaoyu Wang 已提交
1709 1710 1711
  SSmlHandle  *info = NULL;
  int          cnt = 0;
  while (1) {
wmmhello's avatar
wmmhello 已提交
1712 1713 1714 1715 1716
    request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT, reqid);
    if (request == NULL) {
      uError("SML:taos_schemaless_insert error request is null");
      return NULL;
    }
1717

wmmhello's avatar
wmmhello 已提交
1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732
    info = smlBuildSmlInfo(taos);
    if (info == NULL) {
      request->code = TSDB_CODE_OUT_OF_MEMORY;
      uError("SML:taos_schemaless_insert error SSmlHandle is null");
      return (TAOS_RES *)request;
    }
    info->pRequest = request;
    info->isRawLine = rawLine != NULL;
    info->ttl = ttl;
    info->precision = precision;
    info->protocol = (TSDB_SML_PROTOCOL_TYPE)protocol;
    info->msgBuf.buf = info->pRequest->msgBuf;
    info->msgBuf.len = ERROR_MSG_BUF_DEFAULT_SIZE;
    info->lineNum = numLines;

1733 1734
    smlSetReqSQL(request, lines, rawLine, rawLineEnd);

wmmhello's avatar
wmmhello 已提交
1735 1736 1737 1738 1739 1740
    SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf};
    if (request->pDb == NULL) {
      request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
      smlBuildInvalidDataMsg(&msg, "Database not specified", NULL);
      goto end;
    }
1741

wmmhello's avatar
wmmhello 已提交
1742 1743 1744 1745 1746
    if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) {
      request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
      smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL);
      goto end;
    }
wmmhello's avatar
wmmhello 已提交
1747

wmmhello's avatar
wmmhello 已提交
1748 1749 1750 1751 1752 1753
    if (protocol == TSDB_SML_LINE_PROTOCOL &&
        (precision < TSDB_SML_TIMESTAMP_NOT_CONFIGURED || precision > TSDB_SML_TIMESTAMP_NANO_SECONDS)) {
      request->code = TSDB_CODE_SML_INVALID_PRECISION_TYPE;
      smlBuildInvalidDataMsg(&msg, "precision invalidate for line protocol", NULL);
      goto end;
    }
wmmhello's avatar
wmmhello 已提交
1754

wmmhello's avatar
wmmhello 已提交
1755 1756 1757 1758 1759 1760 1761
    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;
    }
1762

wmmhello's avatar
wmmhello 已提交
1763 1764 1765 1766
    code = smlProcess(info, lines, rawLine, rawLineEnd, numLines);
    request->code = code;
    info->cost.endTime = taosGetTimestampUs();
    info->cost.code = code;
1767 1768
    if (NEED_CLIENT_HANDLE_ERROR(code) || code == TSDB_CODE_SDB_OBJ_CREATING ||
        code == TSDB_CODE_PAR_VALUE_TOO_LONG || code == TSDB_CODE_MND_TRANS_CONFLICT) {
X
Xiaoyu Wang 已提交
1769 1770
      if (cnt++ >= 10) {
        uInfo("SML:%" PRIx64 " retry:%d/10 end code:%d, msg:%s", info->id, cnt, code, tstrerror(code));
D
dapan1121 已提交
1771 1772 1773
        break;
      }
      taosMsleep(100);
wmmhello's avatar
wmmhello 已提交
1774
      refreshMeta(request->pTscObj, request);
X
Xiaoyu Wang 已提交
1775 1776
      uInfo("SML:%" PRIx64 " retry:%d/10,ver is old retry or object is creating code:%d, msg:%s", info->id, cnt, code,
            tstrerror(code));
wmmhello's avatar
wmmhello 已提交
1777 1778 1779 1780 1781 1782
      smlDestroyInfo(info);
      info = NULL;
      taos_free_result(request);
      request = NULL;
      continue;
    }
D
dapan1121 已提交
1783
    smlPrintStatisticInfo(info);
wmmhello's avatar
wmmhello 已提交
1784
    break;
wmmhello's avatar
wmmhello 已提交
1785 1786
  }

wmmhello's avatar
wmmhello 已提交
1787
end:
wmmhello's avatar
wmmhello 已提交
1788
  smlDestroyInfo(info);
1789
  return (TAOS_RES *)request;
wmmhello's avatar
wmmhello 已提交
1790
}
wmmhello's avatar
wmmhello 已提交
1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807

/**
 * taos_schemaless_insert() parse and insert data points into database according to
 * different protocol.
 *
 * @param $lines input array may contain multiple lines, each line indicates a data point.
 *               If protocol=2 is used input array should contain single JSON
 *               string(e.g. char *lines[] = {"$JSON_string"}). If need to insert
 *               multiple data points in JSON format, should include them in $JSON_string
 *               as a JSON array.
 * @param $numLines indicates how many data points in $lines.
 *                  If protocol = 2 is used this param will be ignored as $lines should
 *                  contain single JSON string.
 * @param $protocol indicates which protocol to use for parsing:
 *                  0 - influxDB line protocol
 *                  1 - OpenTSDB telnet line protocol
 *                  2 - OpenTSDB JSON format protocol
dengyihao's avatar
dengyihao 已提交
1808
 * @return TAOS_RES
wmmhello's avatar
wmmhello 已提交
1809 1810
 */

1811 1812
TAOS_RES *taos_schemaless_insert_ttl_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol, int precision,
                                                int32_t ttl, int64_t reqid) {
wmmhello's avatar
wmmhello 已提交
1813
  return taos_schemaless_insert_inner(taos, lines, NULL, NULL, numLines, protocol, precision, ttl, reqid);
dengyihao's avatar
dengyihao 已提交
1814 1815
}

1816 1817 1818
TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, TSDB_DEFAULT_TABLE_TTL, 0);
}
wmmhello's avatar
wmmhello 已提交
1819

X
Xiaoyu Wang 已提交
1820 1821
TAOS_RES *taos_schemaless_insert_ttl(TAOS *taos, char *lines[], int numLines, int protocol, int precision,
                                     int32_t ttl) {
1822 1823
  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, ttl, 0);
}
wmmhello's avatar
wmmhello 已提交
1824

X
Xiaoyu Wang 已提交
1825 1826 1827 1828
TAOS_RES *taos_schemaless_insert_with_reqid(TAOS *taos, char *lines[], int numLines, int protocol, int precision,
                                            int64_t reqid) {
  return taos_schemaless_insert_ttl_with_reqid(taos, lines, numLines, protocol, precision, TSDB_DEFAULT_TABLE_TTL,
                                               reqid);
wmmhello's avatar
wmmhello 已提交
1829 1830
}

1831
TAOS_RES *taos_schemaless_insert_raw_ttl_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
1832
                                                    int precision, int32_t ttl, int64_t reqid) {
dengyihao's avatar
dengyihao 已提交
1833 1834 1835 1836 1837 1838 1839 1840 1841 1842 1843 1844
  int numLines = 0;
  *totalRows = 0;
  char *tmp = lines;
  for (int i = 0; i < len; i++) {
    if (lines[i] == '\n' || i == len - 1) {
      numLines++;
      if (tmp[0] != '#' || protocol != TSDB_SML_LINE_PROTOCOL) {  // ignore comment
        (*totalRows)++;
      }
      tmp = lines + i + 1;
    }
  }
wmmhello's avatar
wmmhello 已提交
1845
  return taos_schemaless_insert_inner(taos, NULL, lines, lines + len, *totalRows, protocol, precision, ttl, reqid);
dengyihao's avatar
dengyihao 已提交
1846 1847
}

X
Xiaoyu Wang 已提交
1848 1849 1850 1851
TAOS_RES *taos_schemaless_insert_raw_with_reqid(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
                                                int precision, int64_t reqid) {
  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision,
                                                   TSDB_DEFAULT_TABLE_TTL, reqid);
1852
}
X
Xiaoyu Wang 已提交
1853 1854
TAOS_RES *taos_schemaless_insert_raw_ttl(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
                                         int precision, int32_t ttl) {
1855 1856
  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision, ttl, 0);
}
wmmhello's avatar
wmmhello 已提交
1857

X
Xiaoyu Wang 已提交
1858 1859 1860 1861
TAOS_RES *taos_schemaless_insert_raw(TAOS *taos, char *lines, int len, int32_t *totalRows, int protocol,
                                     int precision) {
  return taos_schemaless_insert_raw_ttl_with_reqid(taos, lines, len, totalRows, protocol, precision,
                                                   TSDB_DEFAULT_TABLE_TTL, 0);
wmmhello's avatar
wmmhello 已提交
1862
}