parInsert.c 39.3 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/>.
 */

X
Xiaoyu Wang 已提交
16 17 18
#include "parInsertData.h"
#include "parInt.h"
#include "parToken.h"
H
refact  
Hongze Cheng 已提交
19
#include "parUtil.h"
20 21 22 23
#include "tglobal.h"
#include "ttime.h"
#include "ttypes.h"

H
refact  
Hongze Cheng 已提交
24 25 26
#define NEXT_TOKEN(pSql, sToken)                \
  do {                                          \
    int32_t index = 0;                          \
27
    sToken = tStrGetToken(pSql, &index, false); \
H
refact  
Hongze Cheng 已提交
28
    pSql += index;                              \
29 30
  } while (0)

H
refact  
Hongze Cheng 已提交
31 32 33
#define NEXT_TOKEN_WITH_PREV(pSql, sToken)     \
  do {                                         \
    int32_t index = 0;                         \
X
Xiaoyu Wang 已提交
34
    sToken = tStrGetToken(pSql, &index, true); \
H
refact  
Hongze Cheng 已提交
35
    pSql += index;                             \
X
Xiaoyu Wang 已提交
36 37
  } while (0)

38
#define NEXT_TOKEN_KEEP_SQL(pSql, sToken, index) \
H
refact  
Hongze Cheng 已提交
39 40
  do {                                           \
    sToken = tStrGetToken(pSql, &index, false);  \
41 42
  } while (0)

H
refact  
Hongze Cheng 已提交
43 44 45
#define CHECK_CODE(expr)             \
  do {                               \
    int32_t code = expr;             \
46
    if (TSDB_CODE_SUCCESS != code) { \
H
refact  
Hongze Cheng 已提交
47 48
      return code;                   \
    }                                \
49 50 51
  } while (0)

typedef struct SInsertParseContext {
H
refact  
Hongze Cheng 已提交
52 53 54 55 56 57 58 59 60 61 62 63 64
  SParseContext*     pComCxt;             // input
  char*              pSql;                // input
  SMsgBuf            msg;                 // input
  STableMeta*        pTableMeta;          // each table
  SParsedDataColInfo tags;                // each table
  SKVRowBuilder      tagsBuilder;         // each table
  SVCreateTbReq      createTblReq;        // each table
  SHashObj*          pVgroupsHashObj;     // global
  SHashObj*          pTableBlockHashObj;  // global
  SHashObj*          pSubTableHashObj;    // global
  SArray*            pTableDataBlocks;    // global
  SArray*            pVgDataBlocks;       // global
  int32_t            totalNum;
X
Xiaoyu Wang 已提交
65
  SVnodeModifOpStmt* pOutput;
66 67
} SInsertParseContext;

H
refact  
Hongze Cheng 已提交
68
typedef int32_t (*_row_append_fn_t)(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param);
X
Xiaoyu Wang 已提交
69 70 71 72

static uint8_t TRUE_VALUE = (uint8_t)TSDB_TRUE;
static uint8_t FALSE_VALUE = (uint8_t)TSDB_FALSE;

73 74 75 76 77 78 79 80 81 82 83 84 85
static int32_t skipInsertInto(SInsertParseContext* pCxt) {
  SToken sToken;
  NEXT_TOKEN(pCxt->pSql, sToken);
  if (TK_INSERT != sToken.type) {
    return buildSyntaxErrMsg(&pCxt->msg, "keyword INSERT is expected", sToken.z);
  }
  NEXT_TOKEN(pCxt->pSql, sToken);
  if (TK_INTO != sToken.type) {
    return buildSyntaxErrMsg(&pCxt->msg, "keyword INTO is expected", sToken.z);
  }
  return TSDB_CODE_SUCCESS;
}

86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147
static int32_t parserValidateIdToken(SToken* pToken) {
  if (pToken == NULL || pToken->z == NULL || pToken->type != TK_NK_ID) {
    return TSDB_CODE_TSC_INVALID_OPERATION;
  }

  // it is a token quoted with escape char '`'
  if (pToken->z[0] == TS_ESCAPE_CHAR && pToken->z[pToken->n - 1] == TS_ESCAPE_CHAR) {
    return TSDB_CODE_SUCCESS;
  }

  char* sep = strnchr(pToken->z, TS_PATH_DELIMITER[0], pToken->n, true);
  if (sep == NULL) {  // It is a single part token, not a complex type
    if (isNumber(pToken)) {
      return TSDB_CODE_TSC_INVALID_OPERATION;
    }

    strntolower(pToken->z, pToken->z, pToken->n);
  } else {  // two part
    int32_t oldLen = pToken->n;
    char*   pStr = pToken->z;

    if (pToken->type == TK_NK_SPACE) {
      pToken->n = (uint32_t)strtrim(pToken->z);
    }

    pToken->n = tGetToken(pToken->z, &pToken->type);
    if (pToken->z[pToken->n] != TS_PATH_DELIMITER[0]) {
      return TSDB_CODE_TSC_INVALID_OPERATION;
    }

    if (pToken->type != TK_NK_ID) {
      return TSDB_CODE_TSC_INVALID_OPERATION;
    }

    int32_t firstPartLen = pToken->n;

    pToken->z = sep + 1;
    pToken->n = (uint32_t)(oldLen - (sep - pStr) - 1);
    int32_t len = tGetToken(pToken->z, &pToken->type);
    if (len != pToken->n || pToken->type != TK_NK_ID) {
      return TSDB_CODE_TSC_INVALID_OPERATION;
    }

    // re-build the whole name string
    if (pStr[firstPartLen] == TS_PATH_DELIMITER[0]) {
      // first part do not have quote do nothing
    } else {
      pStr[firstPartLen] = TS_PATH_DELIMITER[0];
      memmove(&pStr[firstPartLen + 1], pToken->z, pToken->n);
      uint32_t offset = (uint32_t)(pToken->z - (pStr + firstPartLen + 1));
      memset(pToken->z + pToken->n - offset, ' ', offset);
    }

    pToken->n += (firstPartLen + sizeof(TS_PATH_DELIMITER[0]));
    pToken->z = pStr;

    strntolower(pToken->z, pToken->z, pToken->n);
  }

  return TSDB_CODE_SUCCESS;
}

148
static int32_t buildName(SInsertParseContext* pCxt, SToken* pStname, char* fullDbName, char* tableName) {
149
  if (parserValidateIdToken(pStname) != TSDB_CODE_SUCCESS) {
150
    return buildSyntaxErrMsg(&pCxt->msg, "invalid table name", pStname->z);
151 152
  }

153
  char* p = strnchr(pStname->z, TS_PATH_DELIMITER[0], pStname->n, false);
H
refact  
Hongze Cheng 已提交
154
  if (NULL != p) {  // db.table
H
Haojun Liao 已提交
155
    int32_t n = sprintf(fullDbName, "%d.", pCxt->pComCxt->acctId);
H
Haojun Liao 已提交
156
    strncpy(fullDbName + n, pStname->z, p - pStname->z);
157 158
    strncpy(tableName, p + 1, pStname->n - (p - pStname->z) - 1);
  } else {
H
Haojun Liao 已提交
159
    snprintf(fullDbName, TSDB_DB_FNAME_LEN, "%d.%s", pCxt->pComCxt->acctId, pCxt->pComCxt->db);
160 161
    strncpy(tableName, pStname->z, pStname->n);
  }
H
Haojun Liao 已提交
162

163 164 165
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
166 167 168 169 170
static int32_t createSName(SName* pName, SToken* pTableName, SParseContext* pParseCtx, SMsgBuf* pMsgBuf) {
  const char* msg1 = "name too long";
  const char* msg2 = "invalid database name";
  const char* msg3 = "db is not specified";

H
refact  
Hongze Cheng 已提交
171 172
  int32_t code = TSDB_CODE_SUCCESS;
  char*   p = strnchr(pTableName->z, TS_PATH_DELIMITER[0], pTableName->n, true);
X
Xiaoyu Wang 已提交
173

H
refact  
Hongze Cheng 已提交
174
  if (p != NULL) {  // db has been specified in sql string so we ignore current db path
X
Xiaoyu Wang 已提交
175 176 177
    assert(*p == TS_PATH_DELIMITER[0]);

    int32_t dbLen = p - pTableName->z;
H
refact  
Hongze Cheng 已提交
178
    char    name[TSDB_DB_FNAME_LEN] = {0};
X
Xiaoyu Wang 已提交
179 180 181 182 183 184 185 186 187
    strncpy(name, pTableName->z, dbLen);
    dbLen = strdequote(name);

    code = tNameSetDbName(pName, pParseCtx->acctId, name, dbLen);
    if (code != TSDB_CODE_SUCCESS) {
      return buildInvalidOperationMsg(pMsgBuf, msg1);
    }

    int32_t tbLen = pTableName->n - dbLen - 1;
H
refact  
Hongze Cheng 已提交
188
    char    tbname[TSDB_TABLE_FNAME_LEN] = {0};
X
Xiaoyu Wang 已提交
189
    strncpy(tbname, p + 1, tbLen);
H
refact  
Hongze Cheng 已提交
190
    /*tbLen = */ strdequote(tbname);
X
Xiaoyu Wang 已提交
191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225

    code = tNameFromString(pName, tbname, T_NAME_TABLE);
    if (code != 0) {
      return buildInvalidOperationMsg(pMsgBuf, msg1);
    }
  } else {  // get current DB name first, and then set it into path
    if (pTableName->n >= TSDB_TABLE_NAME_LEN) {
      return buildInvalidOperationMsg(pMsgBuf, msg1);
    }

    assert(pTableName->n < TSDB_TABLE_FNAME_LEN);

    char name[TSDB_TABLE_FNAME_LEN] = {0};
    strncpy(name, pTableName->z, pTableName->n);
    strdequote(name);

    if (pParseCtx->db == NULL) {
      return buildInvalidOperationMsg(pMsgBuf, msg3);
    }

    code = tNameSetDbName(pName, pParseCtx->acctId, pParseCtx->db, strlen(pParseCtx->db));
    if (code != TSDB_CODE_SUCCESS) {
      code = buildInvalidOperationMsg(pMsgBuf, msg2);
      return code;
    }

    code = tNameFromString(pName, name, T_NAME_TABLE);
    if (code != 0) {
      code = buildInvalidOperationMsg(pMsgBuf, msg1);
    }
  }

  return code;
}

D
stmt  
dapan1121 已提交
226
static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SToken* pTname, bool isStb) {
H
Haojun Liao 已提交
227
  SParseContext* pBasicCtx = pCxt->pComCxt;
H
refact  
Hongze Cheng 已提交
228 229
  SName          name = {0};
  createSName(&name, pTname, pBasicCtx, &pCxt->msg);
D
stmt  
dapan1121 已提交
230
  if (isStb) {
H
refact  
Hongze Cheng 已提交
231 232
    CHECK_CODE(catalogGetSTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name,
                                    &pCxt->pTableMeta));
D
stmt  
dapan1121 已提交
233
  } else {
H
refact  
Hongze Cheng 已提交
234 235
    CHECK_CODE(catalogGetTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name,
                                   &pCxt->pTableMeta));
D
stmt  
dapan1121 已提交
236
  }
X
Xiaoyu Wang 已提交
237
  SVgroupInfo vg;
H
refact  
Hongze Cheng 已提交
238 239
  CHECK_CODE(
      catalogGetTableHashVgroup(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, &name, &vg));
X
Xiaoyu Wang 已提交
240
  CHECK_CODE(taosHashPut(pCxt->pVgroupsHashObj, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg)));
241

H
refact  
Hongze Cheng 已提交
242
  return TSDB_CODE_SUCCESS;
D
stmt  
dapan1121 已提交
243 244
}

H
refact  
Hongze Cheng 已提交
245
static int32_t getTableMeta(SInsertParseContext* pCxt, SToken* pTname) { return getTableMetaImpl(pCxt, pTname, false); }
D
stmt  
dapan1121 已提交
246

H
refact  
Hongze Cheng 已提交
247
static int32_t getSTableMeta(SInsertParseContext* pCxt, SToken* pTname) { return getTableMetaImpl(pCxt, pTname, true); }
D
stmt  
dapan1121 已提交
248

249 250 251 252 253 254 255 256 257 258
static int32_t findCol(SToken* pColname, int32_t start, int32_t end, SSchema* pSchema) {
  while (start < end) {
    if (strlen(pSchema[start].name) == pColname->n && strncmp(pColname->z, pSchema[start].name, pColname->n) == 0) {
      return start;
    }
    ++start;
  }
  return -1;
}

D
dapan1121 已提交
259
static void buildMsgHeader(STableDataBlocks* src, SVgDataBlocks* blocks) {
H
refact  
Hongze Cheng 已提交
260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277
  SSubmitReq* submit = (SSubmitReq*)blocks->pData;
  submit->header.vgId = htonl(blocks->vg.vgId);
  submit->header.contLen = htonl(blocks->size);
  submit->length = submit->header.contLen;
  submit->numOfBlocks = htonl(blocks->numOfTables);
  SSubmitBlk* blk = (SSubmitBlk*)(submit + 1);
  int32_t     numOfBlocks = blocks->numOfTables;
  while (numOfBlocks--) {
    int32_t dataLen = blk->dataLen;
    blk->uid = htobe64(blk->uid);
    blk->suid = htobe64(blk->suid);
    blk->padding = htonl(blk->padding);
    blk->sversion = htonl(blk->sversion);
    blk->dataLen = htonl(blk->dataLen);
    blk->schemaLen = htonl(blk->schemaLen);
    blk->numOfRows = htons(blk->numOfRows);
    blk = (SSubmitBlk*)(blk->data + dataLen);
  }
278 279 280 281 282 283 284 285 286 287
}

static int32_t buildOutput(SInsertParseContext* pCxt) {
  size_t numOfVg = taosArrayGetSize(pCxt->pVgDataBlocks);
  pCxt->pOutput->pDataBlocks = taosArrayInit(numOfVg, POINTER_BYTES);
  if (NULL == pCxt->pOutput->pDataBlocks) {
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
  for (size_t i = 0; i < numOfVg; ++i) {
    STableDataBlocks* src = taosArrayGetP(pCxt->pVgDataBlocks, i);
H
refact  
Hongze Cheng 已提交
288
    SVgDataBlocks*    dst = taosMemoryCalloc(1, sizeof(SVgDataBlocks));
289 290 291
    if (NULL == dst) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
H
Haojun Liao 已提交
292
    taosHashGetDup(pCxt->pVgroupsHashObj, (const char*)&src->vgId, sizeof(src->vgId), &dst->vg);
293 294
    dst->numOfTables = src->numOfTables;
    dst->size = src->size;
dengyihao's avatar
dengyihao 已提交
295
    TSWAP(dst->pData, src->pData, char*);
D
dapan1121 已提交
296
    buildMsgHeader(src, dst);
297 298 299 300 301
    taosArrayPush(pCxt->pOutput->pDataBlocks, &dst);
  }
  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
302
static int32_t checkTimestamp(STableDataBlocks* pDataBlocks, const char* start) {
303 304 305 306 307
  // once the data block is disordered, we do NOT keep previous timestamp any more
  if (!pDataBlocks->ordered) {
    return TSDB_CODE_SUCCESS;
  }

H
refact  
Hongze Cheng 已提交
308
  TSKEY k = *(TSKEY*)start;
309
  if (k <= pDataBlocks->prevTS) {
310 311 312 313 314 315 316
    pDataBlocks->ordered = false;
  }

  pDataBlocks->prevTS = k;
  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
317 318 319 320 321 322
static int parseTime(char** end, SToken* pToken, int16_t timePrec, int64_t* time, SMsgBuf* pMsgBuf) {
  int32_t index = 0;
  SToken  sToken;
  int64_t interval;
  int64_t ts = 0;
  char*   pTokenEnd = *end;
323 324

  if (pToken->type == TK_NOW) {
325
    ts = taosGetTimestamp(timePrec);
326 327
  } else if (pToken->type == TK_TODAY) {
    ts = taosGetTimestampToday(timePrec);
328
  } else if (pToken->type == TK_NK_INTEGER) {
329 330
    bool isSigned = false;
    toInteger(pToken->z, pToken->n, 10, &ts, &isSigned);
H
refact  
Hongze Cheng 已提交
331
  } else {  // parse the RFC-3339/ISO-8601 timestamp format string
S
os env  
Shengliang Guan 已提交
332
    if (taosParseTime(pToken->z, time, pToken->n, timePrec, tsDaylight) != TSDB_CODE_SUCCESS) {
333
      return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
334 335 336 337 338 339 340
    }

    return TSDB_CODE_SUCCESS;
  }

  for (int k = pToken->n; pToken->z[k] != '\0'; k++) {
    if (pToken->z[k] == ' ' || pToken->z[k] == '\t') continue;
H
refact  
Hongze Cheng 已提交
341
    if (pToken->z[k] == '(' && pToken->z[k + 1] == ')') {  // for insert NOW()/TODAY()
342 343 344 345
      *end = pTokenEnd = &pToken->z[k + 2];
      k++;
      continue;
    }
346
    if (pToken->z[k] == ',') {
347 348
      *end = pTokenEnd;
      *time = ts;
349 350 351 352 353 354 355 356 357 358 359 360 361 362 363
      return 0;
    }

    break;
  }

  /*
   * time expression:
   * e.g., now+12a, now-5h
   */
  SToken valueToken;
  index = 0;
  sToken = tStrGetToken(pTokenEnd, &index, false);
  pTokenEnd += index;

X
Xiaoyu Wang 已提交
364
  if (sToken.type == TK_NK_MINUS || sToken.type == TK_NK_PLUS) {
365 366 367 368 369
    index = 0;
    valueToken = tStrGetToken(pTokenEnd, &index, false);
    pTokenEnd += index;

    if (valueToken.n < 2) {
370
      return buildSyntaxErrMsg(pMsgBuf, "value expected in timestamp", sToken.z);
371 372 373 374 375 376 377
    }

    char unit = 0;
    if (parseAbsoluteDuration(valueToken.z, valueToken.n, &interval, &unit, timePrec) != TSDB_CODE_SUCCESS) {
      return TSDB_CODE_TSC_INVALID_OPERATION;
    }

378
    if (sToken.type == TK_NK_PLUS) {
379
      ts += interval;
380
    } else {
381
      ts = ts - interval;
382 383
    }

384
    *end = pTokenEnd;
385 386
  }

387
  *time = ts;
388 389
  return TSDB_CODE_SUCCESS;
}
390

X
Xiaoyu Wang 已提交
391
static FORCE_INLINE int32_t checkAndTrimValue(SToken* pToken, uint32_t type, char* tmpTokenBuf, SMsgBuf* pMsgBuf) {
H
refact  
Hongze Cheng 已提交
392 393 394 395
  if ((pToken->type != TK_NOW && pToken->type != TK_TODAY && pToken->type != TK_NK_INTEGER &&
       pToken->type != TK_NK_STRING && pToken->type != TK_NK_FLOAT && pToken->type != TK_NK_BOOL &&
       pToken->type != TK_NULL && pToken->type != TK_NK_HEX && pToken->type != TK_NK_OCT &&
       pToken->type != TK_NK_BIN) ||
396
      (pToken->n == 0) || (pToken->type == TK_NK_RP)) {
X
Xiaoyu Wang 已提交
397 398 399 400 401 402 403 404
    return buildSyntaxErrMsg(pMsgBuf, "invalid data or symbol", pToken->z);
  }

  if (IS_NUMERIC_TYPE(type) && pToken->n == 0) {
    return buildSyntaxErrMsg(pMsgBuf, "invalid numeric data", pToken->z);
  }

  // Remove quotation marks
X
Xiaoyu Wang 已提交
405
  if (TK_NK_STRING == pToken->type) {
X
Xiaoyu Wang 已提交
406 407 408 409 410
    if (pToken->n >= TSDB_MAX_BYTES_PER_ROW) {
      return buildSyntaxErrMsg(pMsgBuf, "too long string", pToken->z);
    }

    // delete escape character: \\, \', \"
H
refact  
Hongze Cheng 已提交
411
    char    delim = pToken->z[0];
X
Xiaoyu Wang 已提交
412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433
    int32_t cnt = 0;
    int32_t j = 0;
    for (uint32_t k = 1; k < pToken->n - 1; ++k) {
      if (pToken->z[k] == '\\' || (pToken->z[k] == delim && pToken->z[k + 1] == delim)) {
        tmpTokenBuf[j] = pToken->z[k + 1];
        cnt++;
        j++;
        k++;
        continue;
      }
      tmpTokenBuf[j] = pToken->z[k];
      j++;
    }

    tmpTokenBuf[j] = 0;
    pToken->z = tmpTokenBuf;
    pToken->n -= 2 + cnt;
  }

  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
434
static bool isNullStr(SToken* pToken) {
435
  return (pToken->type == TK_NULL) || ((pToken->type == TK_NK_STRING) && (pToken->n != 0) &&
X
Xiaoyu Wang 已提交
436 437 438
                                       (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0));
}

H
refact  
Hongze Cheng 已提交
439
static FORCE_INLINE int32_t toDouble(SToken* pToken, double* value, char** endPtr) {
X
Xiaoyu Wang 已提交
440 441 442 443 444
  errno = 0;
  *value = strtold(pToken->z, endPtr);

  // not a valid integer number, return error
  if ((*endPtr - pToken->z) != pToken->n) {
445
    return TK_NK_ILLEGAL;
X
Xiaoyu Wang 已提交
446 447 448 449 450
  }

  return pToken->type;
}

H
refact  
Hongze Cheng 已提交
451 452
static int32_t parseValueToken(char** end, SToken* pToken, SSchema* pSchema, int16_t timePrec, char* tmpTokenBuf,
                               _row_append_fn_t func, void* param, SMsgBuf* pMsgBuf) {
X
Xiaoyu Wang 已提交
453
  int64_t iv;
H
refact  
Hongze Cheng 已提交
454
  char*   endptr = NULL;
X
Xiaoyu Wang 已提交
455 456 457 458 459 460 461 462 463
  bool    isSigned = false;

  int32_t code = checkAndTrimValue(pToken, pSchema->type, tmpTokenBuf, pMsgBuf);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  if (isNullStr(pToken)) {
    if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
464
      return buildSyntaxErrMsg(pMsgBuf, "primary timestamp can not be null", pToken->z);
X
Xiaoyu Wang 已提交
465 466
    }

X
Xiaoyu Wang 已提交
467
    return func(pMsgBuf, NULL, 0, param);
X
Xiaoyu Wang 已提交
468 469 470 471
  }

  switch (pSchema->type) {
    case TSDB_DATA_TYPE_BOOL: {
472
      if ((pToken->type == TK_NK_BOOL || pToken->type == TK_NK_STRING) && (pToken->n != 0)) {
X
Xiaoyu Wang 已提交
473
        if (strncmp(pToken->z, "true", pToken->n) == 0) {
X
Xiaoyu Wang 已提交
474
          return func(pMsgBuf, &TRUE_VALUE, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
475
        } else if (strncmp(pToken->z, "false", pToken->n) == 0) {
X
Xiaoyu Wang 已提交
476
          return func(pMsgBuf, &FALSE_VALUE, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
477 478 479
        } else {
          return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
        }
480
      } else if (pToken->type == TK_NK_INTEGER) {
X
Xiaoyu Wang 已提交
481
        return func(pMsgBuf, ((strtoll(pToken->z, NULL, 10) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes, param);
482
      } else if (pToken->type == TK_NK_FLOAT) {
X
Xiaoyu Wang 已提交
483
        return func(pMsgBuf, ((strtod(pToken->z, NULL) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes, param);
X
Xiaoyu Wang 已提交
484 485 486 487 488 489 490 491 492 493 494 495 496
      } else {
        return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
      }
    }

    case TSDB_DATA_TYPE_TINYINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid tinyint data", pToken->z);
      } else if (!IS_VALID_TINYINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "tinyint data overflow", pToken->z);
      }

      uint8_t tmpVal = (uint8_t)iv;
X
Xiaoyu Wang 已提交
497
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
498 499
    }

H
refact  
Hongze Cheng 已提交
500
    case TSDB_DATA_TYPE_UTINYINT: {
X
Xiaoyu Wang 已提交
501 502 503 504 505 506
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned tinyint data", pToken->z);
      } else if (!IS_VALID_UTINYINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned tinyint data overflow", pToken->z);
      }
      uint8_t tmpVal = (uint8_t)iv;
X
Xiaoyu Wang 已提交
507
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
508 509 510 511 512 513 514 515 516
    }

    case TSDB_DATA_TYPE_SMALLINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid smallint data", pToken->z);
      } else if (!IS_VALID_SMALLINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "smallint data overflow", pToken->z);
      }
      int16_t tmpVal = (int16_t)iv;
X
Xiaoyu Wang 已提交
517
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
518 519 520 521 522 523 524 525 526
    }

    case TSDB_DATA_TYPE_USMALLINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned smallint data", pToken->z);
      } else if (!IS_VALID_USMALLINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned smallint data overflow", pToken->z);
      }
      uint16_t tmpVal = (uint16_t)iv;
X
Xiaoyu Wang 已提交
527
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
528 529 530 531 532 533 534 535 536
    }

    case TSDB_DATA_TYPE_INT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid int data", pToken->z);
      } else if (!IS_VALID_INT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "int data overflow", pToken->z);
      }
      int32_t tmpVal = (int32_t)iv;
X
Xiaoyu Wang 已提交
537
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
538 539 540 541 542 543 544 545 546
    }

    case TSDB_DATA_TYPE_UINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned int data", pToken->z);
      } else if (!IS_VALID_UINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned int data overflow", pToken->z);
      }
      uint32_t tmpVal = (uint32_t)iv;
X
Xiaoyu Wang 已提交
547
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
548 549 550 551 552 553 554 555
    }

    case TSDB_DATA_TYPE_BIGINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid bigint data", pToken->z);
      } else if (!IS_VALID_BIGINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "bigint data overflow", pToken->z);
      }
X
Xiaoyu Wang 已提交
556
      return func(pMsgBuf, &iv, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
557 558 559 560 561 562 563 564 565
    }

    case TSDB_DATA_TYPE_UBIGINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv, &isSigned)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned bigint data", pToken->z);
      } else if (!IS_VALID_UBIGINT((uint64_t)iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned bigint data overflow", pToken->z);
      }
      uint64_t tmpVal = (uint64_t)iv;
X
Xiaoyu Wang 已提交
566
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
567 568 569 570
    }

    case TSDB_DATA_TYPE_FLOAT: {
      double dv;
571
      if (TK_NK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
X
Xiaoyu Wang 已提交
572 573
        return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
      }
H
refact  
Hongze Cheng 已提交
574 575
      if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || dv > FLT_MAX || dv < -FLT_MAX || isinf(dv) ||
          isnan(dv)) {
X
Xiaoyu Wang 已提交
576 577 578
        return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
      }
      float tmpVal = (float)dv;
X
Xiaoyu Wang 已提交
579
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
580 581 582 583
    }

    case TSDB_DATA_TYPE_DOUBLE: {
      double dv;
584
      if (TK_NK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
X
Xiaoyu Wang 已提交
585 586 587 588 589
        return buildSyntaxErrMsg(pMsgBuf, "illegal double data", pToken->z);
      }
      if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || isinf(dv) || isnan(dv)) {
        return buildSyntaxErrMsg(pMsgBuf, "illegal double data", pToken->z);
      }
X
Xiaoyu Wang 已提交
590
      return func(pMsgBuf, &dv, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
591 592 593 594 595 596 597 598
    }

    case TSDB_DATA_TYPE_BINARY: {
      // Too long values will raise the invalid sql error message
      if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) {
        return buildSyntaxErrMsg(pMsgBuf, "string data overflow", pToken->z);
      }

X
Xiaoyu Wang 已提交
599
      return func(pMsgBuf, pToken->z, pToken->n, param);
X
Xiaoyu Wang 已提交
600 601 602
    }

    case TSDB_DATA_TYPE_NCHAR: {
X
Xiaoyu Wang 已提交
603
      return func(pMsgBuf, pToken->z, pToken->n, param);
X
Xiaoyu Wang 已提交
604 605 606 607 608 609 610 611
    }

    case TSDB_DATA_TYPE_TIMESTAMP: {
      int64_t tmpVal;
      if (parseTime(end, pToken, timePrec, &tmpVal, pMsgBuf) != TSDB_CODE_SUCCESS) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp", pToken->z);
      }

X
Xiaoyu Wang 已提交
612
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
613 614 615 616 617 618
    }
  }

  return TSDB_CODE_FAILED;
}

619
typedef struct SMemParam {
C
Cary Xu 已提交
620
  SRowBuilder* rb;
621 622 623
  SSchema*     schema;
  int32_t      toffset;
  col_id_t     colIdx;
624 625
} SMemParam;

X
Xiaoyu Wang 已提交
626
static FORCE_INLINE int32_t MemRowAppend(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param) {
C
Cary Xu 已提交
627 628
  SMemParam*   pa = (SMemParam*)param;
  SRowBuilder* rb = pa->rb;
629 630 631 632 633 634

  if (value == NULL) {  // it is a null data
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NULL, value, false, pa->toffset, pa->colIdx);
    return TSDB_CODE_SUCCESS;
  }

635
  if (TSDB_DATA_TYPE_BINARY == pa->schema->type) {
C
Cary Xu 已提交
636
    const char* rowEnd = tdRowEnd(rb->pBuf);
637
    STR_WITH_SIZE_TO_VARSTR(rowEnd, value, len);
C
Cary Xu 已提交
638
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, true, pa->toffset, pa->colIdx);
639 640
  } else if (TSDB_DATA_TYPE_NCHAR == pa->schema->type) {
    // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
C
Cary Xu 已提交
641 642
    int32_t     output = 0;
    const char* rowEnd = tdRowEnd(rb->pBuf);
wafwerar's avatar
wafwerar 已提交
643
    if (!taosMbsToUcs4(value, len, (TdUcs4*)varDataVal(rowEnd), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
X
Xiaoyu Wang 已提交
644 645 646
      char buf[512] = {0};
      snprintf(buf, tListLen(buf), "%s", strerror(errno));
      return buildSyntaxErrMsg(pMsgBuf, buf, value);
647
    }
648
    varDataSetLen(rowEnd, output);
C
Cary Xu 已提交
649
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, false, pa->toffset, pa->colIdx);
650
  } else {
651
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, value, false, pa->toffset, pa->colIdx);
652
  }
653

654 655
  return TSDB_CODE_SUCCESS;
}
656 657 658

// pSql -> tag1_name, ...)
static int32_t parseBoundColumns(SInsertParseContext* pCxt, SParsedDataColInfo* pColList, SSchema* pSchema) {
C
Cary Xu 已提交
659
  col_id_t nCols = pColList->numOfCols;
660

H
refact  
Hongze Cheng 已提交
661
  pColList->numOfBound = 0;
C
Cary Xu 已提交
662
  pColList->boundNullLen = 0;
C
Cary Xu 已提交
663
  memset(pColList->boundColumns, 0, sizeof(col_id_t) * nCols);
C
Cary Xu 已提交
664
  for (col_id_t i = 0; i < nCols; ++i) {
665 666 667
    pColList->cols[i].valStat = VAL_STAT_NONE;
  }

H
refact  
Hongze Cheng 已提交
668 669
  SToken   sToken;
  bool     isOrdered = true;
C
Cary Xu 已提交
670
  col_id_t lastColIdx = -1;  // last column found
671 672 673
  while (1) {
    NEXT_TOKEN(pCxt->pSql, sToken);

674
    if (TK_NK_RP == sToken.type) {
675 676 677
      break;
    }

C
Cary Xu 已提交
678 679
    col_id_t t = lastColIdx + 1;
    col_id_t index = findCol(&sToken, t, nCols, pSchema);
680 681 682 683 684 685 686 687 688 689 690 691
    if (index < 0 && t > 0) {
      index = findCol(&sToken, 0, t, pSchema);
      isOrdered = false;
    }
    if (index < 0) {
      return buildSyntaxErrMsg(&pCxt->msg, "invalid column/tag name", sToken.z);
    }
    if (pColList->cols[index].valStat == VAL_STAT_HAS) {
      return buildSyntaxErrMsg(&pCxt->msg, "duplicated column name", sToken.z);
    }
    lastColIdx = index;
    pColList->cols[index].valStat = VAL_STAT_HAS;
692
    pColList->boundColumns[pColList->numOfBound] = index + PRIMARYKEY_TIMESTAMP_COL_ID;
693
    ++pColList->numOfBound;
C
Cary Xu 已提交
694 695 696 697 698 699 700 701 702 703 704
    switch (pSchema[t].type) {
      case TSDB_DATA_TYPE_BINARY:
        pColList->boundNullLen += (sizeof(VarDataOffsetT) + VARSTR_HEADER_SIZE + CHAR_BYTES);
        break;
      case TSDB_DATA_TYPE_NCHAR:
        pColList->boundNullLen += (sizeof(VarDataOffsetT) + VARSTR_HEADER_SIZE + TSDB_NCHAR_SIZE);
        break;
      default:
        pColList->boundNullLen += TYPE_BYTES[pSchema[t].type];
        break;
    }
705 706 707 708 709
  }

  pColList->orderStatus = isOrdered ? ORDER_STATUS_ORDERED : ORDER_STATUS_DISORDERED;

  if (!isOrdered) {
wafwerar's avatar
wafwerar 已提交
710
    pColList->colIdxInfo = taosMemoryCalloc(pColList->numOfBound, sizeof(SBoundIdxInfo));
711 712 713 714
    if (NULL == pColList->colIdxInfo) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    SBoundIdxInfo* pColIdx = pColList->colIdxInfo;
C
Cary Xu 已提交
715
    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
716
      pColIdx[i].schemaColIdx = pColList->boundColumns[i];
717 718 719
      pColIdx[i].boundIdx = i;
    }
    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar);
C
Cary Xu 已提交
720
    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
721 722 723 724 725
      pColIdx[i].finalIdx = i;
    }
    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar);
  }

726
  memset(&pColList->boundColumns[pColList->numOfBound], 0,
C
Cary Xu 已提交
727
         sizeof(col_id_t) * (pColList->numOfCols - pColList->numOfBound));
728 729 730 731

  return TSDB_CODE_SUCCESS;
}

732
typedef struct SKvParam {
H
refact  
Hongze Cheng 已提交
733 734
  SKVRowBuilder* builder;
  SSchema*       schema;
735 736 737
  char           buf[TSDB_MAX_TAGS_LEN];
} SKvParam;

H
refact  
Hongze Cheng 已提交
738 739
static int32_t KvRowAppend(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param) {
  SKvParam* pa = (SKvParam*)param;
740

741 742
  int8_t  type = pa->schema->type;
  int16_t colId = pa->schema->colId;
743 744 745 746 747 748 749

  if (TSDB_DATA_TYPE_BINARY == type) {
    STR_WITH_SIZE_TO_VARSTR(pa->buf, value, len);
    tdAddColToKVRow(pa->builder, colId, type, pa->buf);
  } else if (TSDB_DATA_TYPE_NCHAR == type) {
    // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
    int32_t output = 0;
wafwerar's avatar
wafwerar 已提交
750
    if (!taosMbsToUcs4(value, len, (TdUcs4*)varDataVal(pa->buf), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
X
Xiaoyu Wang 已提交
751 752
      char buf[512] = {0};
      snprintf(buf, tListLen(buf), "%s", strerror(errno));
H
refact  
Hongze Cheng 已提交
753 754
      return buildSyntaxErrMsg(pMsgBuf, buf, value);
      ;
755 756 757 758 759 760 761 762 763 764 765
    }

    varDataSetLen(pa->buf, output);
    tdAddColToKVRow(pa->builder, colId, type, pa->buf);
  } else {
    tdAddColToKVRow(pa->builder, colId, type, value);
  }

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
766 767 768 769 770 771 772 773 774 775 776
static int32_t buildCreateTbReq(SInsertParseContext* pCxt, const SName* pName, SKVRow row) {
  char dbFName[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(pName, dbFName);
  pCxt->createTblReq.type = TD_CHILD_TABLE;
  pCxt->createTblReq.name = strdup(pName->tname);
  pCxt->createTblReq.ctbCfg.suid = pCxt->pTableMeta->suid;
  pCxt->createTblReq.ctbCfg.pTag = row;

  return TSDB_CODE_SUCCESS;
}

777
// pSql -> tag1_value, ...)
X
Xiaoyu Wang 已提交
778
static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const SName* pName) {
779
  if (tdInitKVRowBuilder(&pCxt->tagsBuilder) < 0) {
780 781 782
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

783
  SKvParam param = {.builder = &pCxt->tagsBuilder};
H
refact  
Hongze Cheng 已提交
784 785
  SToken   sToken;
  char     tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0};  // used for deleting Escape character: \\, \', \"
786
  for (int i = 0; i < pCxt->tags.numOfBound; ++i) {
X
Xiaoyu Wang 已提交
787
    NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken);
H
refact  
Hongze Cheng 已提交
788
    SSchema* pTagSchema = &pSchema[pCxt->tags.boundColumns[i] - 1];  // colId starts with 1
X
Xiaoyu Wang 已提交
789
    param.schema = pTagSchema;
H
refact  
Hongze Cheng 已提交
790 791
    CHECK_CODE(
        parseValueToken(&pCxt->pSql, &sToken, pTagSchema, precision, tmpTokenBuf, KvRowAppend, &param, &pCxt->msg));
792 793
  }

794
  SKVRow row = tdGetKVRowFromBuilder(&pCxt->tagsBuilder);
795 796 797 798 799
  if (NULL == row) {
    return buildInvalidOperationMsg(&pCxt->msg, "tag value expected");
  }
  tdSortKVRowByColIdx(row);

X
Xiaoyu Wang 已提交
800 801
  return buildCreateTbReq(pCxt, pName, row);
}
802

X
Xiaoyu Wang 已提交
803 804 805 806 807 808 809 810
static int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) {
  *pDst = taosMemoryMalloc(TABLE_META_SIZE(pSrc));
  if (NULL == *pDst) {
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
  memcpy(*pDst, pSrc, TABLE_META_SIZE(pSrc));
  return TSDB_CODE_SUCCESS;
}
811

X
Xiaoyu Wang 已提交
812 813 814 815 816
static int32_t storeTableMeta(SHashObj* pHash, const char* pName, int32_t len, STableMeta* pMeta) {
  STableMeta* pBackup = NULL;
  if (TSDB_CODE_SUCCESS != cloneTableMeta(pMeta, &pBackup)) {
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
817
  pBackup->uid = tGenIdPI64();
X
Xiaoyu Wang 已提交
818
  return taosHashPut(pHash, pName, len, &pBackup, POINTER_BYTES);
819 820 821 822
}

// pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
static int32_t parseUsingClause(SInsertParseContext* pCxt, SToken* pTbnameToken) {
X
Xiaoyu Wang 已提交
823 824 825 826
  SName name;
  createSName(&name, pTbnameToken, pCxt->pComCxt, &pCxt->msg);
  char tbFName[TSDB_TABLE_FNAME_LEN];
  tNameExtractFullName(&name, tbFName);
H
refact  
Hongze Cheng 已提交
827
  int32_t      len = strlen(tbFName);
X
Xiaoyu Wang 已提交
828 829 830 831
  STableMeta** pMeta = taosHashGet(pCxt->pSubTableHashObj, tbFName, len);
  if (NULL != pMeta) {
    return cloneTableMeta(*pMeta, &pCxt->pTableMeta);
  }
832

X
Xiaoyu Wang 已提交
833
  SToken sToken;
834 835
  // pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
  NEXT_TOKEN(pCxt->pSql, sToken);
D
stmt  
dapan1121 已提交
836
  CHECK_CODE(getSTableMeta(pCxt, &sToken));
837 838 839
  if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) {
    return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed");
  }
X
Xiaoyu Wang 已提交
840
  CHECK_CODE(storeTableMeta(pCxt->pSubTableHashObj, tbFName, len, pCxt->pTableMeta));
841 842

  SSchema* pTagsSchema = getTableTagSchema(pCxt->pTableMeta);
843
  setBoundColumnInfo(&pCxt->tags, pTagsSchema, getNumOfTags(pCxt->pTableMeta));
844 845 846

  // pSql -> [(tag1_name, ...)] TAGS (tag1_value, ...)
  NEXT_TOKEN(pCxt->pSql, sToken);
847
  if (TK_NK_LP == sToken.type) {
848
    CHECK_CODE(parseBoundColumns(pCxt, &pCxt->tags, pTagsSchema));
849 850 851 852 853 854 855 856
    NEXT_TOKEN(pCxt->pSql, sToken);
  }

  if (TK_TAGS != sToken.type) {
    return buildSyntaxErrMsg(&pCxt->msg, "TAGS is expected", sToken.z);
  }
  // pSql -> (tag1_value, ...)
  NEXT_TOKEN(pCxt->pSql, sToken);
857
  if (TK_NK_LP != sToken.type) {
858 859
    return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
  }
X
Xiaoyu Wang 已提交
860 861 862 863 864
  CHECK_CODE(parseTagsClause(pCxt, pCxt->pTableMeta->schema, getTableInfo(pCxt->pTableMeta).precision, &name));
  NEXT_TOKEN(pCxt->pSql, sToken);
  if (TK_NK_RP != sToken.type) {
    return buildSyntaxErrMsg(&pCxt->msg, ") is expected", sToken.z);
  }
865 866 867 868

  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
869 870
static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks, int16_t timePrec, int32_t* len,
                       char* tmpTokenBuf) {
871
  SParsedDataColInfo* spd = &pDataBlocks->boundColumnInfo;
C
Cary Xu 已提交
872 873 874 875
  SRowBuilder*        pBuilder = &pDataBlocks->rowBuilder;
  STSRow*             row = (STSRow*)(pDataBlocks->pData + pDataBlocks->size);  // skip the SSubmitBlk header

  tdSRowResetBuf(pBuilder, row);
876

H
refact  
Hongze Cheng 已提交
877 878
  bool      isParseBindParam = false;
  SSchema*  schema = getTableColumnSchema(pDataBlocks->pTableMeta);
C
Cary Xu 已提交
879
  SMemParam param = {.rb = pBuilder};
H
refact  
Hongze Cheng 已提交
880
  SToken    sToken = {0};
881 882
  // 1. set the parsed value from sql string
  for (int i = 0; i < spd->numOfBound; ++i) {
X
Xiaoyu Wang 已提交
883
    NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken);
884
    SSchema* pSchema = &schema[spd->boundColumns[i] - 1];
885
    param.schema = pSchema;
886
    getSTSRowAppendInfo(schema, pBuilder->rowType, spd, i, &param.toffset, &param.colIdx);
887
    CHECK_CODE(parseValueToken(&pCxt->pSql, &sToken, pSchema, timePrec, tmpTokenBuf, MemRowAppend, &param, &pCxt->msg));
888 889

    if (PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
C
Cary Xu 已提交
890
      TSKEY tsKey = TD_ROW_KEY(row);
H
refact  
Hongze Cheng 已提交
891
      if (checkTimestamp(pDataBlocks, (const char*)&tsKey) != TSDB_CODE_SUCCESS) {
892 893 894 895 896 897 898
        buildSyntaxErrMsg(&pCxt->msg, "client time/server time can not be mixed up", sToken.z);
        return TSDB_CODE_TSC_INVALID_TIME_STAMP;
      }
    }
  }

  if (!isParseBindParam) {
C
Cary Xu 已提交
899
    // set the null value for the columns that do not assign values
C
Cary Xu 已提交
900
    if ((spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
901
      for (int32_t i = 0; i < spd->numOfCols; ++i) {
C
Cary Xu 已提交
902 903 904
        if (spd->cols[i].valStat == VAL_STAT_NONE) {  // the primary TS key is not VAL_STAT_NONE
          tdAppendColValToTpRow(pBuilder, TD_VTYPE_NONE, getNullValue(schema[i].type), true, schema[i].type, i,
                                spd->cols[i].toffset);
905 906 907 908 909
        }
      }
    }
  }

C
Cary Xu 已提交
910
  // *len = pBuilder->extendedRowSize;
911 912 913 914
  return TSDB_CODE_SUCCESS;
}

// pSql -> (field1_value, ...) [(field1_value2, ...) ...]
915
static int32_t parseValues(SInsertParseContext* pCxt, STableDataBlocks* pDataBlock, int maxRows, int32_t* numOfRows) {
916
  STableComInfo tinfo = getTableInfo(pDataBlock->pTableMeta);
H
refact  
Hongze Cheng 已提交
917
  int32_t       extendedRowSize = getExtendedRowSize(pDataBlock);
C
Cary Xu 已提交
918
  CHECK_CODE(initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo));
919 920

  (*numOfRows) = 0;
H
refact  
Hongze Cheng 已提交
921
  char   tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0};  // used for deleting Escape character: \\, \', \"
922 923
  SToken sToken;
  while (1) {
924 925
    int32_t index = 0;
    NEXT_TOKEN_KEEP_SQL(pCxt->pSql, sToken, index);
926
    if (TK_NK_LP != sToken.type) {
927 928
      break;
    }
929
    pCxt->pSql += index;
930 931 932 933 934 935 936 937 938 939

    if ((*numOfRows) >= maxRows || pDataBlock->size + extendedRowSize >= pDataBlock->nAllocSize) {
      int32_t tSize;
      CHECK_CODE(allocateMemIfNeed(pDataBlock, extendedRowSize, &tSize));
      ASSERT(tSize >= maxRows);
      maxRows = tSize;
    }

    int32_t len = 0;
    CHECK_CODE(parseOneRow(pCxt, pDataBlock, tinfo.precision, &len, tmpTokenBuf));
H
refact  
Hongze Cheng 已提交
940
    pDataBlock->size += extendedRowSize;  // len;
941 942

    NEXT_TOKEN(pCxt->pSql, sToken);
943
    if (TK_NK_RP != sToken.type) {
944 945 946 947 948 949 950
      return buildSyntaxErrMsg(&pCxt->msg, ") expected", sToken.z);
    }

    (*numOfRows)++;
  }

  if (0 == (*numOfRows)) {
H
refact  
Hongze Cheng 已提交
951
    return buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL);
952 953 954 955
  }
  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
956
static int32_t parseValuesClause(SInsertParseContext* pCxt, STableDataBlocks* dataBuf) {
957 958 959 960
  int32_t maxNumOfRows;
  CHECK_CODE(allocateMemIfNeed(dataBuf, getExtendedRowSize(dataBuf), &maxNumOfRows));

  int32_t numOfRows = 0;
961
  CHECK_CODE(parseValues(pCxt, dataBuf, maxNumOfRows, &numOfRows));
962

H
refact  
Hongze Cheng 已提交
963
  SSubmitBlk* pBlocks = (SSubmitBlk*)(dataBuf->pData);
D
dapan1121 已提交
964
  if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, dataBuf, numOfRows)) {
965 966 967 968 969 970 971 972
    return buildInvalidOperationMsg(&pCxt->msg, "too many rows in sql, total number of rows should be less than 32767");
  }

  dataBuf->numOfTables = 1;
  pCxt->totalNum += numOfRows;
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
973 974 975 976 977
static void destroyCreateSubTbReq(SVCreateTbReq* pReq) {
  taosMemoryFreeClear(pReq->name);
  taosMemoryFreeClear(pReq->ctbCfg.pTag);
}

X
Xiaoyu Wang 已提交
978
static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) {
wafwerar's avatar
wafwerar 已提交
979
  taosMemoryFreeClear(pCxt->pTableMeta);
X
Xiaoyu Wang 已提交
980 981
  destroyBoundColumnInfo(&pCxt->tags);
  tdDestroyKVRowBuilder(&pCxt->tagsBuilder);
X
Xiaoyu Wang 已提交
982
  destroyCreateSubTbReq(&pCxt->createTblReq);
X
Xiaoyu Wang 已提交
983 984
}

985 986 987 988 989
static void destroyDataBlock(STableDataBlocks* pDataBlock) {
  if (pDataBlock == NULL) {
    return;
  }

wafwerar's avatar
wafwerar 已提交
990
  taosMemoryFreeClear(pDataBlock->pData);
991 992 993
  if (!pDataBlock->cloned) {
    // free the refcount for metermeta
    if (pDataBlock->pTableMeta != NULL) {
wafwerar's avatar
wafwerar 已提交
994
      taosMemoryFreeClear(pDataBlock->pTableMeta);
995 996 997 998
    }

    destroyBoundColumnInfo(&pDataBlock->boundColumnInfo);
  }
wafwerar's avatar
wafwerar 已提交
999
  taosMemoryFreeClear(pDataBlock);
1000 1001
}

X
Xiaoyu Wang 已提交
1002 1003 1004
static void destroyInsertParseContext(SInsertParseContext* pCxt) {
  destroyInsertParseContextForTable(pCxt);
  taosHashCleanup(pCxt->pVgroupsHashObj);
1005 1006

  destroyBlockHashmap(pCxt->pTableBlockHashObj);
X
Xiaoyu Wang 已提交
1007 1008 1009 1010
  destroyBlockArrayList(pCxt->pTableDataBlocks);
  destroyBlockArrayList(pCxt->pVgDataBlocks);
}

1011 1012 1013 1014 1015 1016
//   tb_name
//       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
//       [(field1_name, ...)]
//       VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
//   [...];
static int32_t parseInsertBody(SInsertParseContext* pCxt) {
X
Xiaoyu Wang 已提交
1017
  // for each table
1018
  while (1) {
X
Xiaoyu Wang 已提交
1019 1020
    destroyInsertParseContextForTable(pCxt);

1021 1022 1023 1024 1025 1026 1027
    SToken sToken;
    // pSql -> tb_name ...
    NEXT_TOKEN(pCxt->pSql, sToken);

    // no data in the sql string anymore.
    if (sToken.n == 0) {
      if (0 == pCxt->totalNum) {
H
refact  
Hongze Cheng 已提交
1028 1029
        return buildInvalidOperationMsg(&pCxt->msg, "no data in sql");
        ;
1030 1031 1032 1033 1034 1035 1036
      }
      break;
    }

    SToken tbnameToken = sToken;
    NEXT_TOKEN(pCxt->pSql, sToken);

H
refact  
Hongze Cheng 已提交
1037
    // USING cluase
1038 1039 1040 1041
    if (TK_USING == sToken.type) {
      CHECK_CODE(parseUsingClause(pCxt, &tbnameToken));
      NEXT_TOKEN(pCxt->pSql, sToken);
    } else {
1042
      CHECK_CODE(getTableMeta(pCxt, &tbnameToken));
1043 1044
    }

H
refact  
Hongze Cheng 已提交
1045
    STableDataBlocks* dataBuf = NULL;
1046
    CHECK_CODE(getDataBlockFromList(pCxt->pTableBlockHashObj, pCxt->pTableMeta->uid, TSDB_DEFAULT_PAYLOAD_SIZE,
H
refact  
Hongze Cheng 已提交
1047 1048 1049
                                    sizeof(SSubmitBlk), getTableInfo(pCxt->pTableMeta).rowSize, pCxt->pTableMeta,
                                    &dataBuf, NULL, &pCxt->createTblReq));

1050
    if (TK_NK_LP == sToken.type) {
1051
      // pSql -> field1_name, ...)
1052
      CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pCxt->pTableMeta)));
1053 1054 1055 1056 1057 1058
      NEXT_TOKEN(pCxt->pSql, sToken);
    }

    if (TK_VALUES == sToken.type) {
      // pSql -> (field1_value, ...) [(field1_value2, ...) ...]
      CHECK_CODE(parseValuesClause(pCxt, dataBuf));
1059
      pCxt->pOutput->insertType = TSDB_QUERY_TYPE_INSERT;
1060 1061 1062 1063
      continue;
    }

    // FILE csv_file_path
1064
    if (TK_NK_FILE == sToken.type) {
1065 1066
      // pSql -> csv_file_path
      NEXT_TOKEN(pCxt->pSql, sToken);
1067
      if (0 == sToken.n || (TK_NK_STRING != sToken.type && TK_NK_ID != sToken.type)) {
1068 1069 1070
        return buildSyntaxErrMsg(&pCxt->msg, "file path is required following keyword FILE", sToken.z);
      }
      // todo
1071
      pCxt->pOutput->insertType = TSDB_QUERY_TYPE_FILE_INSERT;
1072 1073 1074 1075 1076 1077
      continue;
    }

    return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is expected", sToken.z);
  }
  // merge according to vgId
H
refact  
Hongze Cheng 已提交
1078 1079
  if (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT) &&
      taosHashGetSize(pCxt->pTableBlockHashObj) > 0) {
X
Xiaoyu Wang 已提交
1080
    CHECK_CODE(mergeTableDataBlocks(pCxt->pTableBlockHashObj, pCxt->pOutput->payloadType, &pCxt->pVgDataBlocks));
1081
  }
1082
  return buildOutput(pCxt);
1083 1084 1085 1086 1087 1088 1089 1090
}

// INSERT INTO
//   tb_name
//       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
//       [(field1_name, ...)]
//       VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
//   [...];
X
Xiaoyu Wang 已提交
1091
int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
1092
  SInsertParseContext context = {
H
refact  
Hongze Cheng 已提交
1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104
      .pComCxt = pContext,
      .pSql = (char*)pContext->pSql,
      .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
      .pTableMeta = NULL,
      .pVgroupsHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, false),
      .pTableBlockHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false),
      .pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, false),
      .totalNum = 0,
      .pOutput = (SVnodeModifOpStmt*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT)};

  if (NULL == context.pVgroupsHashObj || NULL == context.pTableBlockHashObj || NULL == context.pSubTableHashObj ||
      NULL == context.pOutput) {
X
Xiaoyu Wang 已提交
1105
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
1106 1107
  }

wafwerar's avatar
wafwerar 已提交
1108
  *pQuery = taosMemoryCalloc(1, sizeof(SQuery));
1109 1110 1111
  if (NULL == *pQuery) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
X
Xiaoyu Wang 已提交
1112
  (*pQuery)->execMode = QUERY_EXEC_MODE_SCHEDULE;
1113 1114
  (*pQuery)->haveResultSet = false;
  (*pQuery)->msgType = TDMT_VND_SUBMIT;
X
Xiaoyu Wang 已提交
1115
  (*pQuery)->pRoot = (SNode*)context.pOutput;
1116
  context.pOutput->payloadType = PAYLOAD_TYPE_KV;
1117

1118 1119 1120 1121 1122
  int32_t code = skipInsertInto(&context);
  if (TSDB_CODE_SUCCESS == code) {
    code = parseInsertBody(&context);
  }
  destroyInsertParseContext(&context);
X
Xiaoyu Wang 已提交
1123
  return code;
1124
}