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

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

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

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

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

44 45 46 47 48 49 50
#define NEXT_VALID_TOKEN(pSql, sToken)        \
  do {                                        \
    sToken.n = tGetToken(pSql, &sToken.type); \
    sToken.z = pSql;                          \
    pSql += sToken.n;                         \
  } while (TK_NK_SPACE == sToken.type)

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

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

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

D
stmt  
dapan1121 已提交
74
typedef struct SKvParam {
C
Cary Xu 已提交
75
  int16_t  pos;
C
Cary Xu 已提交
76
  SArray*  pTagVals;
C
Cary Xu 已提交
77 78
  SSchema* schema;
  char     buf[TSDB_MAX_TAGS_LEN];
D
stmt  
dapan1121 已提交
79 80 81 82 83 84 85 86 87
} SKvParam;

typedef struct SMemParam {
  SRowBuilder* rb;
  SSchema*     schema;
  int32_t      toffset;
  col_id_t     colIdx;
} SMemParam;

X
Xiaoyu Wang 已提交
88 89 90
#define CHECK_CODE(expr)             \
  do {                               \
    int32_t code = expr;             \
D
stmt  
dapan1121 已提交
91
    if (TSDB_CODE_SUCCESS != code) { \
X
Xiaoyu Wang 已提交
92 93
      return code;                   \
    }                                \
D
stmt  
dapan1121 已提交
94 95
  } while (0)

96
static int32_t skipInsertInto(char** pSql, SMsgBuf* pMsg) {
97
  SToken sToken;
98
  NEXT_TOKEN(*pSql, sToken);
99
  if (TK_INSERT != sToken.type) {
100
    return buildSyntaxErrMsg(pMsg, "keyword INSERT is expected", sToken.z);
101
  }
102
  NEXT_TOKEN(*pSql, sToken);
103
  if (TK_INTO != sToken.type) {
104
    return buildSyntaxErrMsg(pMsg, "keyword INTO is expected", sToken.z);
105 106 107 108
  }
  return TSDB_CODE_SUCCESS;
}

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 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170
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;
}

171
static int32_t buildName(SInsertParseContext* pCxt, SToken* pStname, char* fullDbName, char* tableName) {
172
  if (parserValidateIdToken(pStname) != TSDB_CODE_SUCCESS) {
173
    return buildSyntaxErrMsg(&pCxt->msg, "invalid table name", pStname->z);
174 175
  }

176
  char* p = strnchr(pStname->z, TS_PATH_DELIMITER[0], pStname->n, false);
H
refact  
Hongze Cheng 已提交
177
  if (NULL != p) {  // db.table
H
Haojun Liao 已提交
178
    int32_t n = sprintf(fullDbName, "%d.", pCxt->pComCxt->acctId);
H
Haojun Liao 已提交
179
    strncpy(fullDbName + n, pStname->z, p - pStname->z);
180 181
    strncpy(tableName, p + 1, pStname->n - (p - pStname->z) - 1);
  } else {
H
Haojun Liao 已提交
182
    snprintf(fullDbName, TSDB_DB_FNAME_LEN, "%d.%s", pCxt->pComCxt->acctId, pCxt->pComCxt->db);
183 184
    strncpy(tableName, pStname->z, pStname->n);
  }
H
Haojun Liao 已提交
185

186 187 188
  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
189
static int32_t createSName(SName* pName, SToken* pTableName, int32_t acctId, const char* dbName, SMsgBuf* pMsgBuf) {
X
Xiaoyu Wang 已提交
190 191 192
  const char* msg1 = "name too long";
  const char* msg2 = "invalid database name";
  const char* msg3 = "db is not specified";
193
  const char* msg4 = "invalid table name";
X
Xiaoyu Wang 已提交
194

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

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

    int32_t dbLen = p - pTableName->z;
H
refact  
Hongze Cheng 已提交
202
    char    name[TSDB_DB_FNAME_LEN] = {0};
X
Xiaoyu Wang 已提交
203 204 205
    strncpy(name, pTableName->z, dbLen);
    dbLen = strdequote(name);

D
stmt  
dapan1121 已提交
206
    code = tNameSetDbName(pName, acctId, name, dbLen);
X
Xiaoyu Wang 已提交
207 208 209 210 211
    if (code != TSDB_CODE_SUCCESS) {
      return buildInvalidOperationMsg(pMsgBuf, msg1);
    }

    int32_t tbLen = pTableName->n - dbLen - 1;
212 213 214 215
    if (tbLen <= 0) {
      return buildInvalidOperationMsg(pMsgBuf, msg4);
    }

216
    char tbname[TSDB_TABLE_FNAME_LEN] = {0};
X
Xiaoyu Wang 已提交
217
    strncpy(tbname, p + 1, tbLen);
H
refact  
Hongze Cheng 已提交
218
    /*tbLen = */ strdequote(tbname);
X
Xiaoyu Wang 已提交
219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234

    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);

D
stmt  
dapan1121 已提交
235
    if (dbName == NULL) {
X
Xiaoyu Wang 已提交
236 237 238
      return buildInvalidOperationMsg(pMsgBuf, msg3);
    }

D
stmt  
dapan1121 已提交
239
    code = tNameSetDbName(pName, acctId, dbName, strlen(dbName));
X
Xiaoyu Wang 已提交
240 241 242 243 244 245 246 247 248 249 250 251 252 253
    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;
}

254
static int32_t checkAuth(SInsertParseContext* pCxt, char* pDbFname, bool* pPass) {
H
Haojun Liao 已提交
255
  SParseContext* pBasicCtx = pCxt->pComCxt;
256 257 258 259 260 261
  if (NULL != pCxt->pMetaCache) {
    return getUserAuthFromCache(pCxt->pMetaCache, pBasicCtx->pUser, pDbFname, AUTH_TYPE_WRITE, pPass);
  }
  return catalogChkAuth(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pBasicCtx->pUser, pDbFname,
                        AUTH_TYPE_WRITE, pPass);
}
D
dapan 已提交
262

263 264 265 266 267 268 269 270 271 272 273 274 275
static int32_t getTableSchema(SInsertParseContext* pCxt, SName* pTbName, bool isStb, STableMeta** pTableMeta) {
  SParseContext* pBasicCtx = pCxt->pComCxt;
  if (NULL != pCxt->pMetaCache) {
    return getTableMetaFromCache(pCxt->pMetaCache, pTbName, pTableMeta);
  }
  if (isStb) {
    return catalogGetSTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pTbName,
                                pTableMeta);
  }
  return catalogGetTableMeta(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pTbName, pTableMeta);
}

static int32_t getTableVgroup(SInsertParseContext* pCxt, SName* pTbName, SVgroupInfo* pVg) {
H
Haojun Liao 已提交
276
  SParseContext* pBasicCtx = pCxt->pComCxt;
277 278 279 280 281
  if (NULL != pCxt->pMetaCache) {
    return getTableVgroupFromCache(pCxt->pMetaCache, pTbName, pVg);
  }
  return catalogGetTableHashVgroup(pBasicCtx->pCatalog, pBasicCtx->pTransporter, &pBasicCtx->mgmtEpSet, pTbName, pVg);
}
D
dapan 已提交
282

283
static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SName* name, char* dbFname, bool isStb) {
D
dapan 已提交
284
  bool pass = false;
285
  CHECK_CODE(checkAuth(pCxt, dbFname, &pass));
D
dapan 已提交
286 287 288
  if (!pass) {
    return TSDB_CODE_PAR_PERMISSION_DENIED;
  }
289 290 291

  CHECK_CODE(getTableSchema(pCxt, name, isStb, &pCxt->pTableMeta));
  if (!isStb) {
X
Xiaoyu Wang 已提交
292
    SVgroupInfo vg;
293
    CHECK_CODE(getTableVgroup(pCxt, name, &vg));
X
Xiaoyu Wang 已提交
294
    CHECK_CODE(taosHashPut(pCxt->pVgroupsHashObj, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg)));
D
stmt  
dapan1121 已提交
295
  }
H
refact  
Hongze Cheng 已提交
296
  return TSDB_CODE_SUCCESS;
D
stmt  
dapan1121 已提交
297 298
}

X
Xiaoyu Wang 已提交
299 300 301
static int32_t getTableMeta(SInsertParseContext* pCxt, SName* name, char* dbFname) {
  return getTableMetaImpl(pCxt, name, dbFname, false);
}
D
stmt  
dapan1121 已提交
302

X
Xiaoyu Wang 已提交
303 304 305
static int32_t getSTableMeta(SInsertParseContext* pCxt, SName* name, char* dbFname) {
  return getTableMetaImpl(pCxt, name, dbFname, true);
}
D
stmt  
dapan1121 已提交
306

307 308 309 310 311 312 313 314 315 316
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 已提交
317
static void buildMsgHeader(STableDataBlocks* src, SVgDataBlocks* blocks) {
H
refact  
Hongze Cheng 已提交
318 319 320 321 322 323 324 325 326
  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;
327
    int32_t schemaLen = blk->schemaLen;
H
refact  
Hongze Cheng 已提交
328 329 330 331 332 333 334
    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);
335
    blk = (SSubmitBlk*)(blk->data + schemaLen + dataLen);
H
refact  
Hongze Cheng 已提交
336
  }
337 338 339 340 341 342 343 344 345 346
}

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 已提交
347
    SVgDataBlocks*    dst = taosMemoryCalloc(1, sizeof(SVgDataBlocks));
348 349 350
    if (NULL == dst) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
H
Haojun Liao 已提交
351
    taosHashGetDup(pCxt->pVgroupsHashObj, (const char*)&src->vgId, sizeof(src->vgId), &dst->vg);
352 353
    dst->numOfTables = src->numOfTables;
    dst->size = src->size;
wafwerar's avatar
wafwerar 已提交
354
    TSWAP(dst->pData, src->pData);
D
dapan1121 已提交
355
    buildMsgHeader(src, dst);
356 357 358 359 360
    taosArrayPush(pCxt->pOutput->pDataBlocks, &dst);
  }
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
361
int32_t checkTimestamp(STableDataBlocks* pDataBlocks, const char* start) {
362 363 364 365 366
  // 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 已提交
367
  TSKEY k = *(TSKEY*)start;
368
  if (k <= pDataBlocks->prevTS) {
369 370 371 372 373 374 375
    pDataBlocks->ordered = false;
  }

  pDataBlocks->prevTS = k;
  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
376 377 378 379 380 381
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;
382 383

  if (pToken->type == TK_NOW) {
384
    ts = taosGetTimestamp(timePrec);
385 386
  } else if (pToken->type == TK_TODAY) {
    ts = taosGetTimestampToday(timePrec);
387
  } else if (pToken->type == TK_NK_INTEGER) {
X
Xiaoyu Wang 已提交
388
    toInteger(pToken->z, pToken->n, 10, &ts);
H
refact  
Hongze Cheng 已提交
389
  } else {  // parse the RFC-3339/ISO-8601 timestamp format string
S
os env  
Shengliang Guan 已提交
390
    if (taosParseTime(pToken->z, time, pToken->n, timePrec, tsDaylight) != TSDB_CODE_SUCCESS) {
391
      return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
392 393 394 395 396 397 398
    }

    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 已提交
399
    if (pToken->z[k] == '(' && pToken->z[k + 1] == ')') {  // for insert NOW()/TODAY()
400 401 402 403
      *end = pTokenEnd = &pToken->z[k + 2];
      k++;
      continue;
    }
404
    if (pToken->z[k] == ',') {
405 406
      *end = pTokenEnd;
      *time = ts;
407 408 409 410 411 412 413 414 415 416 417 418 419 420 421
      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 已提交
422
  if (sToken.type == TK_NK_MINUS || sToken.type == TK_NK_PLUS) {
423 424 425 426 427
    index = 0;
    valueToken = tStrGetToken(pTokenEnd, &index, false);
    pTokenEnd += index;

    if (valueToken.n < 2) {
428
      return buildSyntaxErrMsg(pMsgBuf, "value expected in timestamp", sToken.z);
429 430 431 432 433 434 435
    }

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

436
    if (sToken.type == TK_NK_PLUS) {
437
      ts += interval;
438
    } else {
439
      ts = ts - interval;
440 441
    }

442
    *end = pTokenEnd;
443 444
  }

445
  *time = ts;
446 447
  return TSDB_CODE_SUCCESS;
}
448

wmmhello's avatar
wmmhello 已提交
449
static FORCE_INLINE int32_t checkAndTrimValue(SToken* pToken, char* tmpTokenBuf, SMsgBuf* pMsgBuf) {
H
refact  
Hongze Cheng 已提交
450 451 452 453
  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) ||
454
      (pToken->n == 0) || (pToken->type == TK_NK_RP)) {
X
Xiaoyu Wang 已提交
455 456 457 458
    return buildSyntaxErrMsg(pMsgBuf, "invalid data or symbol", pToken->z);
  }

  // Remove quotation marks
X
Xiaoyu Wang 已提交
459
  if (TK_NK_STRING == pToken->type) {
X
Xiaoyu Wang 已提交
460 461 462 463
    if (pToken->n >= TSDB_MAX_BYTES_PER_ROW) {
      return buildSyntaxErrMsg(pMsgBuf, "too long string", pToken->z);
    }

464
    int32_t len = trimString(pToken->z, pToken->n, tmpTokenBuf, TSDB_MAX_BYTES_PER_ROW);
X
Xiaoyu Wang 已提交
465
    pToken->z = tmpTokenBuf;
466
    pToken->n = len;
X
Xiaoyu Wang 已提交
467 468 469 470 471
  }

  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
472
static bool isNullStr(SToken* pToken) {
473
  return (pToken->type == TK_NULL) || ((pToken->type == TK_NK_STRING) && (pToken->n != 0) &&
X
Xiaoyu Wang 已提交
474 475 476
                                       (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0));
}

H
refact  
Hongze Cheng 已提交
477
static FORCE_INLINE int32_t toDouble(SToken* pToken, double* value, char** endPtr) {
X
Xiaoyu Wang 已提交
478
  errno = 0;
wafwerar's avatar
wafwerar 已提交
479
  *value = taosStr2Double(pToken->z, endPtr);
X
Xiaoyu Wang 已提交
480 481 482

  // not a valid integer number, return error
  if ((*endPtr - pToken->z) != pToken->n) {
483
    return TK_NK_ILLEGAL;
X
Xiaoyu Wang 已提交
484 485 486 487 488
  }

  return pToken->type;
}

H
refact  
Hongze Cheng 已提交
489 490
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 已提交
491 492 493
  int64_t  iv;
  uint64_t uv;
  char*    endptr = NULL;
X
Xiaoyu Wang 已提交
494

wmmhello's avatar
wmmhello 已提交
495
  int32_t code = checkAndTrimValue(pToken, tmpTokenBuf, pMsgBuf);
X
Xiaoyu Wang 已提交
496 497 498 499 500 501
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  if (isNullStr(pToken)) {
    if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
D
stmt  
dapan1121 已提交
502
      return buildSyntaxErrMsg(pMsgBuf, "primary timestamp should not be null", pToken->z);
X
Xiaoyu Wang 已提交
503 504
    }

X
Xiaoyu Wang 已提交
505
    return func(pMsgBuf, NULL, 0, param);
X
Xiaoyu Wang 已提交
506 507 508 509
  }

  switch (pSchema->type) {
    case TSDB_DATA_TYPE_BOOL: {
510
      if ((pToken->type == TK_NK_BOOL || pToken->type == TK_NK_STRING) && (pToken->n != 0)) {
X
Xiaoyu Wang 已提交
511
        if (strncmp(pToken->z, "true", pToken->n) == 0) {
X
Xiaoyu Wang 已提交
512
          return func(pMsgBuf, &TRUE_VALUE, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
513
        } else if (strncmp(pToken->z, "false", pToken->n) == 0) {
X
Xiaoyu Wang 已提交
514
          return func(pMsgBuf, &FALSE_VALUE, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
515 516 517
        } else {
          return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
        }
518
      } else if (pToken->type == TK_NK_INTEGER) {
519 520
        return func(pMsgBuf, ((taosStr2Int64(pToken->z, NULL, 10) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes,
                    param);
521
      } else if (pToken->type == TK_NK_FLOAT) {
522 523
        return func(pMsgBuf, ((taosStr2Double(pToken->z, NULL) == 0) ? &FALSE_VALUE : &TRUE_VALUE), pSchema->bytes,
                    param);
X
Xiaoyu Wang 已提交
524 525 526 527 528 529
      } else {
        return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
      }
    }

    case TSDB_DATA_TYPE_TINYINT: {
X
Xiaoyu Wang 已提交
530
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
X
Xiaoyu Wang 已提交
531 532 533 534 535 536
        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 已提交
537
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
538 539
    }

H
refact  
Hongze Cheng 已提交
540
    case TSDB_DATA_TYPE_UTINYINT: {
X
Xiaoyu Wang 已提交
541
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
X
Xiaoyu Wang 已提交
542
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned tinyint data", pToken->z);
X
Xiaoyu Wang 已提交
543
      } else if (!IS_VALID_UTINYINT(uv)) {
X
Xiaoyu Wang 已提交
544 545
        return buildSyntaxErrMsg(pMsgBuf, "unsigned tinyint data overflow", pToken->z);
      }
X
Xiaoyu Wang 已提交
546
      uint8_t tmpVal = (uint8_t)uv;
X
Xiaoyu Wang 已提交
547
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
548 549 550
    }

    case TSDB_DATA_TYPE_SMALLINT: {
X
Xiaoyu Wang 已提交
551
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
X
Xiaoyu Wang 已提交
552 553 554 555 556
        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 已提交
557
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
558 559 560
    }

    case TSDB_DATA_TYPE_USMALLINT: {
X
Xiaoyu Wang 已提交
561
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
X
Xiaoyu Wang 已提交
562
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned smallint data", pToken->z);
X
Xiaoyu Wang 已提交
563
      } else if (!IS_VALID_USMALLINT(uv)) {
X
Xiaoyu Wang 已提交
564 565
        return buildSyntaxErrMsg(pMsgBuf, "unsigned smallint data overflow", pToken->z);
      }
X
Xiaoyu Wang 已提交
566
      uint16_t tmpVal = (uint16_t)uv;
X
Xiaoyu Wang 已提交
567
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
568 569 570
    }

    case TSDB_DATA_TYPE_INT: {
X
Xiaoyu Wang 已提交
571
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
X
Xiaoyu Wang 已提交
572 573 574 575 576
        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 已提交
577
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
578 579 580
    }

    case TSDB_DATA_TYPE_UINT: {
X
Xiaoyu Wang 已提交
581
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
X
Xiaoyu Wang 已提交
582
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned int data", pToken->z);
X
Xiaoyu Wang 已提交
583
      } else if (!IS_VALID_UINT(uv)) {
X
Xiaoyu Wang 已提交
584 585
        return buildSyntaxErrMsg(pMsgBuf, "unsigned int data overflow", pToken->z);
      }
X
Xiaoyu Wang 已提交
586
      uint32_t tmpVal = (uint32_t)uv;
X
Xiaoyu Wang 已提交
587
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
588 589 590
    }

    case TSDB_DATA_TYPE_BIGINT: {
X
Xiaoyu Wang 已提交
591
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
X
Xiaoyu Wang 已提交
592 593 594 595
        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 已提交
596
      return func(pMsgBuf, &iv, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
597 598 599
    }

    case TSDB_DATA_TYPE_UBIGINT: {
X
Xiaoyu Wang 已提交
600
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
X
Xiaoyu Wang 已提交
601
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned bigint data", pToken->z);
X
Xiaoyu Wang 已提交
602
      } else if (!IS_VALID_UBIGINT(uv)) {
X
Xiaoyu Wang 已提交
603 604
        return buildSyntaxErrMsg(pMsgBuf, "unsigned bigint data overflow", pToken->z);
      }
X
Xiaoyu Wang 已提交
605
      return func(pMsgBuf, &uv, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
606 607 608 609
    }

    case TSDB_DATA_TYPE_FLOAT: {
      double dv;
610
      if (TK_NK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
X
Xiaoyu Wang 已提交
611 612
        return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
      }
H
refact  
Hongze Cheng 已提交
613 614
      if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || dv > FLT_MAX || dv < -FLT_MAX || isinf(dv) ||
          isnan(dv)) {
X
Xiaoyu Wang 已提交
615 616 617
        return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
      }
      float tmpVal = (float)dv;
X
Xiaoyu Wang 已提交
618
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
619 620 621 622
    }

    case TSDB_DATA_TYPE_DOUBLE: {
      double dv;
623
      if (TK_NK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
X
Xiaoyu Wang 已提交
624 625 626 627 628
        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 已提交
629
      return func(pMsgBuf, &dv, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
630 631 632 633 634
    }

    case TSDB_DATA_TYPE_BINARY: {
      // Too long values will raise the invalid sql error message
      if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) {
D
dapan1121 已提交
635
        return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
X
Xiaoyu Wang 已提交
636 637
      }

X
Xiaoyu Wang 已提交
638
      return func(pMsgBuf, pToken->z, pToken->n, param);
X
Xiaoyu Wang 已提交
639 640 641
    }

    case TSDB_DATA_TYPE_NCHAR: {
X
Xiaoyu Wang 已提交
642
      return func(pMsgBuf, pToken->z, pToken->n, param);
X
Xiaoyu Wang 已提交
643
    }
644
    case TSDB_DATA_TYPE_JSON: {
X
Xiaoyu Wang 已提交
645
      if (pToken->n > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
646 647 648 649
        return buildSyntaxErrMsg(pMsgBuf, "json string too long than 4095", pToken->z);
      }
      return func(pMsgBuf, pToken->z, pToken->n, param);
    }
X
Xiaoyu Wang 已提交
650 651 652 653 654 655
    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 已提交
656
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
657 658 659 660 661 662
    }
  }

  return TSDB_CODE_FAILED;
}

X
Xiaoyu Wang 已提交
663
static FORCE_INLINE int32_t MemRowAppend(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param) {
C
Cary Xu 已提交
664 665
  SMemParam*   pa = (SMemParam*)param;
  SRowBuilder* rb = pa->rb;
666 667 668 669 670 671

  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;
  }

672
  if (TSDB_DATA_TYPE_BINARY == pa->schema->type) {
C
Cary Xu 已提交
673
    const char* rowEnd = tdRowEnd(rb->pBuf);
674
    STR_WITH_SIZE_TO_VARSTR(rowEnd, value, len);
675
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, false, pa->toffset, pa->colIdx);
676 677
  } 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 已提交
678 679
    int32_t     output = 0;
    const char* rowEnd = tdRowEnd(rb->pBuf);
wafwerar's avatar
wafwerar 已提交
680
    if (!taosMbsToUcs4(value, len, (TdUcs4*)varDataVal(rowEnd), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
D
dapan1121 已提交
681 682 683
      if (errno == E2BIG) {
        return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pa->schema->name);
      }
X
Xiaoyu Wang 已提交
684 685 686
      char buf[512] = {0};
      snprintf(buf, tListLen(buf), "%s", strerror(errno));
      return buildSyntaxErrMsg(pMsgBuf, buf, value);
687
    }
688
    varDataSetLen(rowEnd, output);
C
Cary Xu 已提交
689
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, false, pa->toffset, pa->colIdx);
690
  } else {
691
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, value, false, pa->toffset, pa->colIdx);
692
  }
693

694 695
  return TSDB_CODE_SUCCESS;
}
696 697 698

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

H
refact  
Hongze Cheng 已提交
701
  pColList->numOfBound = 0;
C
Cary Xu 已提交
702
  pColList->boundNullLen = 0;
C
Cary Xu 已提交
703
  memset(pColList->boundColumns, 0, sizeof(col_id_t) * nCols);
C
Cary Xu 已提交
704
  for (col_id_t i = 0; i < nCols; ++i) {
705 706 707
    pColList->cols[i].valStat = VAL_STAT_NONE;
  }

H
refact  
Hongze Cheng 已提交
708 709
  SToken   sToken;
  bool     isOrdered = true;
C
Cary Xu 已提交
710
  col_id_t lastColIdx = -1;  // last column found
711 712 713
  while (1) {
    NEXT_TOKEN(pCxt->pSql, sToken);

714
    if (TK_NK_RP == sToken.type) {
715 716 717
      break;
    }

C
Cary Xu 已提交
718 719
    col_id_t t = lastColIdx + 1;
    col_id_t index = findCol(&sToken, t, nCols, pSchema);
720 721 722 723 724
    if (index < 0 && t > 0) {
      index = findCol(&sToken, 0, t, pSchema);
      isOrdered = false;
    }
    if (index < 0) {
725
      return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_INVALID_COLUMN, sToken.z);
726 727 728 729 730 731
    }
    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;
X
Xiaoyu Wang 已提交
732
    pColList->boundColumns[pColList->numOfBound] = index;
733
    ++pColList->numOfBound;
C
Cary Xu 已提交
734 735 736 737 738 739 740 741 742 743 744
    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;
    }
745 746 747 748 749
  }

  pColList->orderStatus = isOrdered ? ORDER_STATUS_ORDERED : ORDER_STATUS_DISORDERED;

  if (!isOrdered) {
wafwerar's avatar
wafwerar 已提交
750
    pColList->colIdxInfo = taosMemoryCalloc(pColList->numOfBound, sizeof(SBoundIdxInfo));
751 752 753 754
    if (NULL == pColList->colIdxInfo) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    SBoundIdxInfo* pColIdx = pColList->colIdxInfo;
C
Cary Xu 已提交
755
    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
756
      pColIdx[i].schemaColIdx = pColList->boundColumns[i];
757 758 759
      pColIdx[i].boundIdx = i;
    }
    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar);
C
Cary Xu 已提交
760
    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
761 762 763 764 765
      pColIdx[i].finalIdx = i;
    }
    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar);
  }

X
Xiaoyu Wang 已提交
766
  if (pColList->numOfCols > pColList->numOfBound) {
767 768 769
    memset(&pColList->boundColumns[pColList->numOfBound], 0,
           sizeof(col_id_t) * (pColList->numOfCols - pColList->numOfBound));
  }
770 771 772 773

  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
774 775 776 777 778
static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid) {
  pTbReq->type = TD_CHILD_TABLE;
  pTbReq->name = strdup(tname);
  pTbReq->ctb.suid = suid;
  pTbReq->ctb.pTag = (uint8_t*)pTag;
779

wmmhello's avatar
wmmhello 已提交
780 781
  return;
}
782

783 784
static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema, int16_t timePrec, STagVal* val,
                             SMsgBuf* pMsgBuf) {
wmmhello's avatar
wmmhello 已提交
785 786 787 788 789 790 791 792
  int64_t  iv;
  uint64_t uv;
  char*    endptr = NULL;

  if (isNullStr(pToken)) {
    if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type && PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
      return buildSyntaxErrMsg(pMsgBuf, "primary timestamp should not be null", pToken->z);
    }
793 794 795 796

    return TSDB_CODE_SUCCESS;
  }

wmmhello's avatar
wmmhello 已提交
797
  val->cid = pSchema->colId;
wmmhello's avatar
wmmhello 已提交
798
  val->type = pSchema->type;
C
Cary Xu 已提交
799

wmmhello's avatar
wmmhello 已提交
800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815
  switch (pSchema->type) {
    case TSDB_DATA_TYPE_BOOL: {
      if ((pToken->type == TK_NK_BOOL || pToken->type == TK_NK_STRING) && (pToken->n != 0)) {
        if (strncmp(pToken->z, "true", pToken->n) == 0) {
          *(int8_t*)(&val->i64) = TRUE_VALUE;
        } else if (strncmp(pToken->z, "false", pToken->n) == 0) {
          *(int8_t*)(&val->i64) = FALSE_VALUE;
        } else {
          return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
        }
      } else if (pToken->type == TK_NK_INTEGER) {
        *(int8_t*)(&val->i64) = ((taosStr2Int64(pToken->z, NULL, 10) == 0) ? FALSE_VALUE : TRUE_VALUE);
      } else if (pToken->type == TK_NK_FLOAT) {
        *(int8_t*)(&val->i64) = ((taosStr2Double(pToken->z, NULL) == 0) ? FALSE_VALUE : TRUE_VALUE);
      } else {
        return buildSyntaxErrMsg(pMsgBuf, "invalid bool data", pToken->z);
D
dapan1121 已提交
816
      }
wmmhello's avatar
wmmhello 已提交
817 818
      break;
    }
819

wmmhello's avatar
wmmhello 已提交
820 821 822 823 824
    case TSDB_DATA_TYPE_TINYINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid tinyint data", pToken->z);
      } else if (!IS_VALID_TINYINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "tinyint data overflow", pToken->z);
D
dapan1121 已提交
825
      }
wmmhello's avatar
wmmhello 已提交
826 827 828

      *(int8_t*)(&val->i64) = iv;
      break;
829 830
    }

wmmhello's avatar
wmmhello 已提交
831 832 833 834 835 836 837 838 839
    case TSDB_DATA_TYPE_UTINYINT: {
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned tinyint data", pToken->z);
      } else if (!IS_VALID_UTINYINT(uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned tinyint data overflow", pToken->z);
      }
      *(uint8_t*)(&val->i64) = uv;
      break;
    }
840

wmmhello's avatar
wmmhello 已提交
841 842 843 844 845 846 847 848 849
    case TSDB_DATA_TYPE_SMALLINT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid smallint data", pToken->z);
      } else if (!IS_VALID_SMALLINT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "smallint data overflow", pToken->z);
      }
      *(int16_t*)(&val->i64) = iv;
      break;
    }
850

wmmhello's avatar
wmmhello 已提交
851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939
    case TSDB_DATA_TYPE_USMALLINT: {
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned smallint data", pToken->z);
      } else if (!IS_VALID_USMALLINT(uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned smallint data overflow", pToken->z);
      }
      *(uint16_t*)(&val->i64) = uv;
      break;
    }

    case TSDB_DATA_TYPE_INT: {
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid int data", pToken->z);
      } else if (!IS_VALID_INT(iv)) {
        return buildSyntaxErrMsg(pMsgBuf, "int data overflow", pToken->z);
      }
      *(int32_t*)(&val->i64) = iv;
      break;
    }

    case TSDB_DATA_TYPE_UINT: {
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned int data", pToken->z);
      } else if (!IS_VALID_UINT(uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned int data overflow", pToken->z);
      }
      *(uint32_t*)(&val->i64) = uv;
      break;
    }

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

      val->i64 = iv;
      break;
    }

    case TSDB_DATA_TYPE_UBIGINT: {
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned bigint data", pToken->z);
      } else if (!IS_VALID_UBIGINT(uv)) {
        return buildSyntaxErrMsg(pMsgBuf, "unsigned bigint data overflow", pToken->z);
      }
      *(uint64_t*)(&val->i64) = uv;
      break;
    }

    case TSDB_DATA_TYPE_FLOAT: {
      double dv;
      if (TK_NK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
        return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
      }
      if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || dv > FLT_MAX || dv < -FLT_MAX || isinf(dv) ||
          isnan(dv)) {
        return buildSyntaxErrMsg(pMsgBuf, "illegal float data", pToken->z);
      }
      *(float*)(&val->i64) = dv;
      break;
    }

    case TSDB_DATA_TYPE_DOUBLE: {
      double dv;
      if (TK_NK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
        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);
      }

      *(double*)(&val->i64) = dv;
      break;
    }

    case TSDB_DATA_TYPE_BINARY: {
      // Too long values will raise the invalid sql error message
      if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) {
        return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
      }
      val->pData = pToken->z;
      val->nData = pToken->n;
      break;
    }

    case TSDB_DATA_TYPE_NCHAR: {
      int32_t output = 0;
940 941
      void*   p = taosMemoryCalloc(1, pToken->n * TSDB_NCHAR_SIZE);
      if (p == NULL) {
wmmhello's avatar
wmmhello 已提交
942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966
        return TSDB_CODE_OUT_OF_MEMORY;
      }
      if (!taosMbsToUcs4(pToken->z, pToken->n, (TdUcs4*)(p), pSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
        if (errno == E2BIG) {
          taosMemoryFree(p);
          return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
        }
        char buf[512] = {0};
        snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
        taosMemoryFree(p);
        return buildSyntaxErrMsg(pMsgBuf, buf, pToken->z);
      }
      val->pData = p;
      val->nData = output;
      break;
    }
    case TSDB_DATA_TYPE_TIMESTAMP: {
      if (parseTime(end, pToken, timePrec, &iv, pMsgBuf) != TSDB_CODE_SUCCESS) {
        return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp", pToken->z);
      }

      val->i64 = iv;
      break;
    }
  }
X
Xiaoyu Wang 已提交
967 968 969 970

  return TSDB_CODE_SUCCESS;
}

971
// pSql -> tag1_value, ...)
wmmhello's avatar
wmmhello 已提交
972
static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const char* tName) {
wmmhello's avatar
wmmhello 已提交
973
  int32_t code = TSDB_CODE_SUCCESS;
974 975 976 977 978
  SArray* pTagVals = taosArrayInit(pCxt->tags.numOfBound, sizeof(STagVal));
  SToken  sToken;
  bool    isParseBindParam = false;
  bool    isJson = false;
  STag*   pTag = NULL;
979
  for (int i = 0; i < pCxt->tags.numOfBound; ++i) {
X
Xiaoyu Wang 已提交
980
    NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken);
D
stmt  
dapan1121 已提交
981 982 983 984

    if (sToken.type == TK_NK_QUESTION) {
      isParseBindParam = true;
      if (NULL == pCxt->pStmtCb) {
wmmhello's avatar
wmmhello 已提交
985 986
        code = buildSyntaxErrMsg(&pCxt->msg, "? only used in stmt", sToken.z);
        goto end;
D
stmt  
dapan1121 已提交
987 988 989 990 991 992
      }

      continue;
    }

    if (isParseBindParam) {
wmmhello's avatar
wmmhello 已提交
993 994
      code = buildInvalidOperationMsg(&pCxt->msg, "no mix usage for ? and tag values");
      goto end;
D
stmt  
dapan1121 已提交
995
    }
X
Xiaoyu Wang 已提交
996

X
Xiaoyu Wang 已提交
997
    SSchema* pTagSchema = &pSchema[pCxt->tags.boundColumns[i]];
998
    char*    tmpTokenBuf = taosMemoryCalloc(1, sToken.n);  // this can be optimize with parse column
wmmhello's avatar
wmmhello 已提交
999 1000 1001 1002 1003
    code = checkAndTrimValue(&sToken, tmpTokenBuf, &pCxt->msg);
    if (code != TSDB_CODE_SUCCESS) {
      taosMemoryFree(tmpTokenBuf);
      goto end;
    }
1004
    if (pTagSchema->type == TSDB_DATA_TYPE_JSON) {
wmmhello's avatar
wmmhello 已提交
1005 1006 1007 1008 1009
      if (sToken.n > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
        code = buildSyntaxErrMsg(&pCxt->msg, "json string too long than 4095", sToken.z);
        taosMemoryFree(tmpTokenBuf);
        goto end;
      }
wmmhello's avatar
wmmhello 已提交
1010
      code = parseJsontoTagData(sToken.z, pTagVals, &pTag, &pCxt->msg);
wmmhello's avatar
wmmhello 已提交
1011
      taosMemoryFree(tmpTokenBuf);
1012
      if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1013 1014
        goto end;
      }
wmmhello's avatar
wmmhello 已提交
1015
      isJson = true;
1016
    } else {
wmmhello's avatar
wmmhello 已提交
1017
      STagVal val = {0};
wmmhello's avatar
wmmhello 已提交
1018
      code = parseTagToken(&pCxt->pSql, &sToken, pTagSchema, precision, &val, &pCxt->msg);
wmmhello's avatar
wmmhello 已提交
1019 1020 1021 1022
      if (TSDB_CODE_SUCCESS != code) {
        taosMemoryFree(tmpTokenBuf);
        goto end;
      }
1023
      if (pTagSchema->type != TSDB_DATA_TYPE_BINARY) {
wmmhello's avatar
wmmhello 已提交
1024 1025 1026 1027
        taosMemoryFree(tmpTokenBuf);
      }
      taosArrayPush(pTagVals, &val);
    }
1028 1029
  }

D
stmt  
dapan1121 已提交
1030
  if (isParseBindParam) {
wmmhello's avatar
wmmhello 已提交
1031 1032
    code = TSDB_CODE_SUCCESS;
    goto end;
D
stmt  
dapan1121 已提交
1033 1034
  }

1035
  if (!isJson && (code = tTagNew(pTagVals, 1, false, &pTag)) != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1036
    goto end;
1037 1038
  }

wmmhello's avatar
wmmhello 已提交
1039 1040 1041 1042
  buildCreateTbReq(&pCxt->createTblReq, tName, pTag, pCxt->pTableMeta->suid);

end:
  for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) {
1043 1044
    STagVal* p = (STagVal*)taosArrayGet(pTagVals, i);
    if (IS_VAR_DATA_TYPE(p->type)) {
wmmhello's avatar
wmmhello 已提交
1045 1046 1047 1048 1049
      taosMemoryFree(p->pData);
    }
  }
  taosArrayDestroy(pTagVals);
  return code;
X
Xiaoyu Wang 已提交
1050
}
1051

X
Xiaoyu Wang 已提交
1052 1053 1054 1055 1056 1057 1058 1059
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;
}
1060

X
Xiaoyu Wang 已提交
1061 1062
static int32_t storeTableMeta(SInsertParseContext* pCxt, SHashObj* pHash, SName* pTableName, const char* pName,
                              int32_t len, STableMeta* pMeta) {
1063 1064
  SVgroupInfo vg;
  CHECK_CODE(getTableVgroup(pCxt, pTableName, &vg));
X
Xiaoyu Wang 已提交
1065 1066
  CHECK_CODE(taosHashPut(pCxt->pVgroupsHashObj, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg)));

D
dapan 已提交
1067
  pMeta->uid = 0;
X
Xiaoyu Wang 已提交
1068
  pMeta->vgId = vg.vgId;
D
dapan 已提交
1069
  pMeta->tableType = TSDB_CHILD_TABLE;
X
Xiaoyu Wang 已提交
1070

X
Xiaoyu Wang 已提交
1071 1072 1073 1074 1075
  STableMeta* pBackup = NULL;
  if (TSDB_CODE_SUCCESS != cloneTableMeta(pMeta, &pBackup)) {
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
  return taosHashPut(pHash, pName, len, &pBackup, POINTER_BYTES);
1076 1077 1078
}

// pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
D
dapan 已提交
1079
static int32_t parseUsingClause(SInsertParseContext* pCxt, SName* name, char* tbFName) {
H
refact  
Hongze Cheng 已提交
1080
  int32_t      len = strlen(tbFName);
X
Xiaoyu Wang 已提交
1081 1082 1083 1084
  STableMeta** pMeta = taosHashGet(pCxt->pSubTableHashObj, tbFName, len);
  if (NULL != pMeta) {
    return cloneTableMeta(*pMeta, &pCxt->pTableMeta);
  }
1085

X
Xiaoyu Wang 已提交
1086
  SToken sToken;
1087 1088
  // pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
  NEXT_TOKEN(pCxt->pSql, sToken);
D
dapan 已提交
1089 1090 1091 1092 1093

  SName sname;
  createSName(&sname, &sToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg);
  char stbFName[TSDB_TABLE_FNAME_LEN];
  tNameExtractFullName(&sname, stbFName);
X
Xiaoyu Wang 已提交
1094

D
dapan 已提交
1095
  CHECK_CODE(getSTableMeta(pCxt, &sname, stbFName));
1096 1097 1098
  if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) {
    return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed");
  }
D
dapan 已提交
1099
  CHECK_CODE(storeTableMeta(pCxt, pCxt->pSubTableHashObj, name, tbFName, len, pCxt->pTableMeta));
1100 1101

  SSchema* pTagsSchema = getTableTagSchema(pCxt->pTableMeta);
1102
  setBoundColumnInfo(&pCxt->tags, pTagsSchema, getNumOfTags(pCxt->pTableMeta));
1103 1104 1105

  // pSql -> [(tag1_name, ...)] TAGS (tag1_value, ...)
  NEXT_TOKEN(pCxt->pSql, sToken);
1106
  if (TK_NK_LP == sToken.type) {
1107
    CHECK_CODE(parseBoundColumns(pCxt, &pCxt->tags, pTagsSchema));
1108 1109 1110 1111 1112 1113 1114 1115
    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);
1116
  if (TK_NK_LP != sToken.type) {
1117 1118
    return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
  }
X
Xiaoyu Wang 已提交
1119
  CHECK_CODE(parseTagsClause(pCxt, pTagsSchema, getTableInfo(pCxt->pTableMeta).precision, name->tname));
1120 1121 1122 1123
  NEXT_VALID_TOKEN(pCxt->pSql, sToken);
  if (TK_NK_COMMA == sToken.type) {
    return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_TAGS_NOT_MATCHED);
  } else if (TK_NK_RP != sToken.type) {
X
Xiaoyu Wang 已提交
1124 1125
    return buildSyntaxErrMsg(&pCxt->msg, ") is expected", sToken.z);
  }
1126 1127 1128 1129

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1130 1131
static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks, int16_t timePrec, bool* gotRow,
                       char* tmpTokenBuf) {
1132
  SParsedDataColInfo* spd = &pDataBlocks->boundColumnInfo;
C
Cary Xu 已提交
1133 1134 1135 1136
  SRowBuilder*        pBuilder = &pDataBlocks->rowBuilder;
  STSRow*             row = (STSRow*)(pDataBlocks->pData + pDataBlocks->size);  // skip the SSubmitBlk header

  tdSRowResetBuf(pBuilder, row);
1137

H
refact  
Hongze Cheng 已提交
1138 1139
  bool      isParseBindParam = false;
  SSchema*  schema = getTableColumnSchema(pDataBlocks->pTableMeta);
C
Cary Xu 已提交
1140
  SMemParam param = {.rb = pBuilder};
H
refact  
Hongze Cheng 已提交
1141
  SToken    sToken = {0};
1142 1143
  // 1. set the parsed value from sql string
  for (int i = 0; i < spd->numOfBound; ++i) {
X
Xiaoyu Wang 已提交
1144
    NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken);
X
Xiaoyu Wang 已提交
1145
    SSchema* pSchema = &schema[spd->boundColumns[i]];
D
stmt  
dapan1121 已提交
1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158

    if (sToken.type == TK_NK_QUESTION) {
      isParseBindParam = true;
      if (NULL == pCxt->pStmtCb) {
        return buildSyntaxErrMsg(&pCxt->msg, "? only used in stmt", sToken.z);
      }

      continue;
    }

    if (isParseBindParam) {
      return buildInvalidOperationMsg(&pCxt->msg, "no mix usage for ? and values");
    }
X
Xiaoyu Wang 已提交
1159

1160
    param.schema = pSchema;
D
stmt  
dapan1121 已提交
1161
    getSTSRowAppendInfo(pBuilder->rowType, spd, i, &param.toffset, &param.colIdx);
1162
    CHECK_CODE(parseValueToken(&pCxt->pSql, &sToken, pSchema, timePrec, tmpTokenBuf, MemRowAppend, &param, &pCxt->msg));
1163 1164

    if (PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
C
Cary Xu 已提交
1165
      TSKEY tsKey = TD_ROW_KEY(row);
1166
      checkTimestamp(pDataBlocks, (const char*)&tsKey);
1167 1168 1169 1170
    }
  }

  if (!isParseBindParam) {
C
Cary Xu 已提交
1171
    // set the null value for the columns that do not assign values
C
Cary Xu 已提交
1172
    if ((spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
1173
      for (int32_t i = 0; i < spd->numOfCols; ++i) {
C
Cary Xu 已提交
1174 1175 1176
        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);
1177 1178 1179
        }
      }
    }
D
stmt  
dapan1121 已提交
1180 1181

    *gotRow = true;
C
Cary Xu 已提交
1182 1183
#ifdef TD_DEBUG_PRINT_ROW
    STSchema* pSTSchema = tdGetSTSChemaFromSSChema(&schema, spd->numOfCols);
C
Cary Xu 已提交
1184
    tdSRowPrint(row, pSTSchema, __func__);
C
Cary Xu 已提交
1185 1186
    taosMemoryFree(pSTSchema);
#endif
1187 1188
  }

C
Cary Xu 已提交
1189
  // *len = pBuilder->extendedRowSize;
1190 1191 1192 1193
  return TSDB_CODE_SUCCESS;
}

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

  (*numOfRows) = 0;
H
refact  
Hongze Cheng 已提交
1200
  char   tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0};  // used for deleting Escape character: \\, \', \"
1201 1202
  SToken sToken;
  while (1) {
1203 1204
    int32_t index = 0;
    NEXT_TOKEN_KEEP_SQL(pCxt->pSql, sToken, index);
1205
    if (TK_NK_LP != sToken.type) {
1206 1207
      break;
    }
1208
    pCxt->pSql += index;
1209 1210 1211 1212 1213 1214 1215 1216

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

D
stmt  
dapan1121 已提交
1217 1218 1219
    bool gotRow = false;
    CHECK_CODE(parseOneRow(pCxt, pDataBlock, tinfo.precision, &gotRow, tmpTokenBuf));
    if (gotRow) {
X
Xiaoyu Wang 已提交
1220
      pDataBlock->size += extendedRowSize;  // len;
D
stmt  
dapan1121 已提交
1221
    }
1222

1223 1224 1225 1226
    NEXT_VALID_TOKEN(pCxt->pSql, sToken);
    if (TK_NK_COMMA == sToken.type) {
      return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_INVALID_COLUMNS_NUM);
    } else if (TK_NK_RP != sToken.type) {
1227 1228 1229
      return buildSyntaxErrMsg(&pCxt->msg, ") expected", sToken.z);
    }

D
stmt  
dapan1121 已提交
1230 1231 1232
    if (gotRow) {
      (*numOfRows)++;
    }
1233 1234
  }

D
stmt  
dapan1121 已提交
1235
  if (0 == (*numOfRows) && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
X
Xiaoyu Wang 已提交
1236
    return buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL);
1237 1238 1239 1240
  }
  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
1241
static int32_t parseValuesClause(SInsertParseContext* pCxt, STableDataBlocks* dataBuf) {
1242 1243 1244 1245
  int32_t maxNumOfRows;
  CHECK_CODE(allocateMemIfNeed(dataBuf, getExtendedRowSize(dataBuf), &maxNumOfRows));

  int32_t numOfRows = 0;
1246
  CHECK_CODE(parseValues(pCxt, dataBuf, maxNumOfRows, &numOfRows));
1247

H
refact  
Hongze Cheng 已提交
1248
  SSubmitBlk* pBlocks = (SSubmitBlk*)(dataBuf->pData);
D
dapan1121 已提交
1249
  if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, dataBuf, numOfRows)) {
1250 1251 1252 1253 1254 1255 1256 1257
    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;
}

D
stmt  
dapan1121 已提交
1258
void destroyCreateSubTbReq(SVCreateTbReq* pReq) {
X
Xiaoyu Wang 已提交
1259
  taosMemoryFreeClear(pReq->name);
H
Hongze Cheng 已提交
1260
  taosMemoryFreeClear(pReq->ctb.pTag);
X
Xiaoyu Wang 已提交
1261 1262
}

X
Xiaoyu Wang 已提交
1263
static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) {
wafwerar's avatar
wafwerar 已提交
1264
  taosMemoryFreeClear(pCxt->pTableMeta);
X
Xiaoyu Wang 已提交
1265
  destroyBoundColumnInfo(&pCxt->tags);
X
Xiaoyu Wang 已提交
1266
  destroyCreateSubTbReq(&pCxt->createTblReq);
X
Xiaoyu Wang 已提交
1267 1268 1269 1270 1271
}

static void destroyInsertParseContext(SInsertParseContext* pCxt) {
  destroyInsertParseContextForTable(pCxt);
  taosHashCleanup(pCxt->pVgroupsHashObj);
1272
  taosHashCleanup(pCxt->pSubTableHashObj);
D
dapan1121 已提交
1273
  taosHashCleanup(pCxt->pTableNameHashObj);
1274 1275

  destroyBlockHashmap(pCxt->pTableBlockHashObj);
X
Xiaoyu Wang 已提交
1276 1277 1278
  destroyBlockArrayList(pCxt->pVgDataBlocks);
}

1279 1280 1281 1282 1283 1284
//   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) {
1285 1286 1287
  int32_t tbNum = 0;
  char    tbFName[TSDB_TABLE_FNAME_LEN];
  bool    autoCreateTbl = false;
X
Xiaoyu Wang 已提交
1288

X
Xiaoyu Wang 已提交
1289
  // for each table
1290 1291
  while (1) {
    SToken sToken;
X
Xiaoyu Wang 已提交
1292
    char*  tbName = NULL;
D
stmt  
dapan1121 已提交
1293

1294 1295 1296 1297 1298
    // pSql -> tb_name ...
    NEXT_TOKEN(pCxt->pSql, sToken);

    // no data in the sql string anymore.
    if (sToken.n == 0) {
D
dapan1121 已提交
1299 1300 1301
      if (sToken.type && pCxt->pSql[0]) {
        return buildSyntaxErrMsg(&pCxt->msg, "invalid charactor in SQL", sToken.z);
      }
X
Xiaoyu Wang 已提交
1302

D
stmt  
dapan1121 已提交
1303
      if (0 == pCxt->totalNum && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
H
refact  
Hongze Cheng 已提交
1304
        return buildInvalidOperationMsg(&pCxt->msg, "no data in sql");
1305 1306 1307 1308
      }
      break;
    }

D
stmt  
dapan1121 已提交
1309
    if (TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT) && tbNum > 0) {
X
Xiaoyu Wang 已提交
1310
      return buildInvalidOperationMsg(&pCxt->msg, "single table allowed in one stmt");
D
stmt  
dapan1121 已提交
1311 1312
    }

D
stmt  
dapan1121 已提交
1313 1314
    destroyInsertParseContextForTable(pCxt);

D
stmt  
dapan1121 已提交
1315 1316 1317
    if (TK_NK_QUESTION == sToken.type) {
      if (pCxt->pStmtCb) {
        CHECK_CODE((*pCxt->pStmtCb->getTbNameFn)(pCxt->pStmtCb->pStmt, &tbName));
X
Xiaoyu Wang 已提交
1318

D
stmt  
dapan1121 已提交
1319 1320 1321 1322 1323 1324
        sToken.z = tbName;
        sToken.n = strlen(tbName);
      } else {
        return buildSyntaxErrMsg(&pCxt->msg, "? only used in stmt", sToken.z);
      }
    }
X
Xiaoyu Wang 已提交
1325

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

D
dapan 已提交
1329
    SName name;
1330
    CHECK_CODE(createSName(&name, &tbnameToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg));
D
dapan 已提交
1331

1332
    tNameExtractFullName(&name, tbFName);
X
Xiaoyu Wang 已提交
1333 1334
    CHECK_CODE(taosHashPut(pCxt->pTableNameHashObj, tbFName, strlen(tbFName), &name, sizeof(SName)));

1335
    // USING clause
1336
    if (TK_USING == sToken.type) {
D
dapan 已提交
1337
      CHECK_CODE(parseUsingClause(pCxt, &name, tbFName));
1338
      NEXT_TOKEN(pCxt->pSql, sToken);
D
dapan 已提交
1339
      autoCreateTbl = true;
1340
    } else {
D
dapan1121 已提交
1341 1342 1343
      char dbFName[TSDB_DB_FNAME_LEN];
      tNameGetFullDbName(&name, dbFName);
      CHECK_CODE(getTableMeta(pCxt, &name, dbFName));
1344 1345
    }

H
refact  
Hongze Cheng 已提交
1346
    STableDataBlocks* dataBuf = NULL;
D
dapan 已提交
1347
    CHECK_CODE(getDataBlockFromList(pCxt->pTableBlockHashObj, tbFName, strlen(tbFName), TSDB_DEFAULT_PAYLOAD_SIZE,
H
refact  
Hongze Cheng 已提交
1348 1349
                                    sizeof(SSubmitBlk), getTableInfo(pCxt->pTableMeta).rowSize, pCxt->pTableMeta,
                                    &dataBuf, NULL, &pCxt->createTblReq));
1350

1351
    if (TK_NK_LP == sToken.type) {
1352
      // pSql -> field1_name, ...)
D
dapan1121 已提交
1353
      CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pCxt->pTableMeta)));
1354 1355 1356 1357 1358 1359
      NEXT_TOKEN(pCxt->pSql, sToken);
    }

    if (TK_VALUES == sToken.type) {
      // pSql -> (field1_value, ...) [(field1_value2, ...) ...]
      CHECK_CODE(parseValuesClause(pCxt, dataBuf));
D
stmt  
dapan1121 已提交
1360
      TSDB_QUERY_SET_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_INSERT);
D
stmt  
dapan1121 已提交
1361 1362

      tbNum++;
1363 1364 1365 1366
      continue;
    }

    // FILE csv_file_path
X
Xiaoyu Wang 已提交
1367
    if (TK_FILE == sToken.type) {
1368 1369
      // pSql -> csv_file_path
      NEXT_TOKEN(pCxt->pSql, sToken);
1370
      if (0 == sToken.n || (TK_NK_STRING != sToken.type && TK_NK_ID != sToken.type)) {
1371 1372 1373
        return buildSyntaxErrMsg(&pCxt->msg, "file path is required following keyword FILE", sToken.z);
      }
      // todo
1374
      pCxt->pOutput->insertType = TSDB_QUERY_TYPE_FILE_INSERT;
D
stmt  
dapan1121 已提交
1375 1376

      tbNum++;
1377 1378 1379 1380 1381
      continue;
    }

    return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is expected", sToken.z);
  }
X
Xiaoyu Wang 已提交
1382

D
stmt  
dapan1121 已提交
1383
  if (TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) {
X
Xiaoyu Wang 已提交
1384
    SParsedDataColInfo* tags = taosMemoryMalloc(sizeof(pCxt->tags));
D
stmt  
dapan1121 已提交
1385 1386 1387 1388
    if (NULL == tags) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    memcpy(tags, &pCxt->tags, sizeof(pCxt->tags));
1389 1390
    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl,
                                pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj);
D
stmt  
dapan1121 已提交
1391

D
dapan 已提交
1392
    memset(&pCxt->tags, 0, sizeof(pCxt->tags));
D
stmt  
dapan1121 已提交
1393 1394
    pCxt->pVgroupsHashObj = NULL;
    pCxt->pTableBlockHashObj = NULL;
D
dapan1121 已提交
1395
    pCxt->pTableMeta = NULL;
X
Xiaoyu Wang 已提交
1396

D
stmt  
dapan1121 已提交
1397 1398
    return TSDB_CODE_SUCCESS;
  }
X
Xiaoyu Wang 已提交
1399

1400
  // merge according to vgId
D
stmt  
dapan1121 已提交
1401
  if (taosHashGetSize(pCxt->pTableBlockHashObj) > 0) {
X
Xiaoyu Wang 已提交
1402
    CHECK_CODE(mergeTableDataBlocks(pCxt->pTableBlockHashObj, pCxt->pOutput->payloadType, &pCxt->pVgDataBlocks));
1403
  }
1404
  return buildOutput(pCxt);
1405 1406
}

1407
int32_t isNotSchemalessDb(SParseContext* pContext, char *dbName){
wmmhello's avatar
wmmhello 已提交
1408
  SName          name;
1409
  tNameSetDbName(&name, pContext->acctId, dbName, strlen(dbName));
wmmhello's avatar
wmmhello 已提交
1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423
  char dbFname[TSDB_DB_FNAME_LEN] = {0};
  tNameGetFullDbName(&name, dbFname);
  SDbCfgInfo pInfo = {0};
  int32_t code = catalogGetDBCfg(pContext->pCatalog, pContext->pTransporter, &pContext->mgmtEpSet, dbFname, &pInfo);
  if (code != TSDB_CODE_SUCCESS) {
    parserError("catalogGetDBCfg error, code:%s, dbFName:%s", tstrerror(code), dbFname);
    return code;
  }
  if (pInfo.schemaless){
    parserError("can not insert into schemaless db:%s", dbFname);
    return TSDB_CODE_SML_INVALID_DB_CONF;
  }
  return TSDB_CODE_SUCCESS;
}
1424 1425 1426 1427 1428 1429
// 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 已提交
1430
int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
1431
  SInsertParseContext context = {
X
Xiaoyu Wang 已提交
1432 1433 1434 1435
      .pComCxt = pContext,
      .pSql = (char*)pContext->pSql,
      .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
      .pTableMeta = NULL,
X
Xiaoyu Wang 已提交
1436 1437
      .pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
      .pTableNameHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
X
Xiaoyu Wang 已提交
1438 1439 1440
      .totalNum = 0,
      .pOutput = (SVnodeModifOpStmt*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT),
      .pStmtCb = pContext->pStmtCb};
1441

D
stmt  
dapan1121 已提交
1442
  if (pContext->pStmtCb && *pQuery) {
X
Xiaoyu Wang 已提交
1443 1444
    (*pContext->pStmtCb->getExecInfoFn)(pContext->pStmtCb->pStmt, &context.pVgroupsHashObj,
                                        &context.pTableBlockHashObj);
D
stmt  
dapan1121 已提交
1445
  } else {
X
Xiaoyu Wang 已提交
1446 1447 1448
    context.pVgroupsHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
    context.pTableBlockHashObj =
        taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
D
stmt  
dapan1121 已提交
1449
  }
X
Xiaoyu Wang 已提交
1450 1451

  if (NULL == context.pVgroupsHashObj || NULL == context.pTableBlockHashObj || NULL == context.pSubTableHashObj ||
X
Xiaoyu Wang 已提交
1452
      NULL == context.pTableNameHashObj || NULL == context.pOutput) {
X
Xiaoyu Wang 已提交
1453
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
1454 1455
  }

D
stmt  
dapan1121 已提交
1456 1457 1458 1459
  if (pContext->pStmtCb) {
    TSDB_QUERY_SET_TYPE(context.pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT);
  }

1460
  if (NULL == *pQuery) {
D
stmt  
dapan1121 已提交
1461 1462 1463 1464
    *pQuery = taosMemoryCalloc(1, sizeof(SQuery));
    if (NULL == *pQuery) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
1465
  }
1466 1467 1468 1469
  (*pQuery)->execMode = QUERY_EXEC_MODE_SCHEDULE;
  (*pQuery)->haveResultSet = false;
  (*pQuery)->msgType = TDMT_VND_SUBMIT;
  (*pQuery)->pRoot = (SNode*)context.pOutput;
X
Xiaoyu Wang 已提交
1470

wmmhello's avatar
wmmhello 已提交
1471 1472 1473 1474 1475
  int32_t code = isNotSchemalessDb(pContext);
  if(code != TSDB_CODE_SUCCESS){
    return code;
  }

D
dapan1121 已提交
1476 1477 1478 1479 1480 1481
  if (NULL == (*pQuery)->pTableList) {
    (*pQuery)->pTableList = taosArrayInit(taosHashGetSize(context.pTableNameHashObj), sizeof(SName));
    if (NULL == (*pQuery)->pTableList) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
  }
1482

1483
  context.pOutput->payloadType = PAYLOAD_TYPE_KV;
1484

wmmhello's avatar
wmmhello 已提交
1485
  code = skipInsertInto(&context.pSql, &context.msg);
1486 1487 1488
  if (TSDB_CODE_SUCCESS == code) {
    code = parseInsertBody(&context);
  }
1489
  if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) {
X
Xiaoyu Wang 已提交
1490 1491 1492 1493 1494 1495
    SName* pTable = taosHashIterate(context.pTableNameHashObj, NULL);
    while (NULL != pTable) {
      taosArrayPush((*pQuery)->pTableList, pTable);
      pTable = taosHashIterate(context.pTableNameHashObj, pTable);
    }
  }
1496
  destroyInsertParseContext(&context);
X
Xiaoyu Wang 已提交
1497
  return code;
1498
}
D
stmt  
dapan1121 已提交
1499

1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520 1521 1522 1523 1524 1525 1526 1527 1528 1529 1530 1531 1532 1533 1534 1535 1536 1537 1538 1539 1540 1541 1542 1543 1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555 1556 1557 1558 1559 1560 1561 1562 1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664
typedef struct SInsertParseSyntaxCxt {
  SParseContext*   pComCxt;
  char*            pSql;
  SMsgBuf          msg;
  SParseMetaCache* pMetaCache;
} SInsertParseSyntaxCxt;

static int32_t skipParentheses(SInsertParseSyntaxCxt* pCxt) {
  SToken sToken;
  while (1) {
    NEXT_TOKEN(pCxt->pSql, sToken);
    if (TK_NK_RP == sToken.type) {
      break;
    }
    if (0 == sToken.n) {
      return buildSyntaxErrMsg(&pCxt->msg, ") expected", NULL);
    }
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t skipBoundColumns(SInsertParseSyntaxCxt* pCxt) { return skipParentheses(pCxt); }

// pSql -> (field1_value, ...) [(field1_value2, ...) ...]
static int32_t skipValuesClause(SInsertParseSyntaxCxt* pCxt) {
  int32_t numOfRows = 0;
  SToken  sToken;
  while (1) {
    int32_t index = 0;
    NEXT_TOKEN_KEEP_SQL(pCxt->pSql, sToken, index);
    if (TK_NK_LP != sToken.type) {
      break;
    }
    pCxt->pSql += index;

    CHECK_CODE(skipParentheses(pCxt));
    ++numOfRows;
  }
  if (0 == numOfRows) {
    return buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL);
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t skipTagsClause(SInsertParseSyntaxCxt* pCxt) { return skipParentheses(pCxt); }

// pSql -> [(tag1_name, ...)] TAGS (tag1_value, ...)
static int32_t skipUsingClause(SInsertParseSyntaxCxt* pCxt) {
  SToken sToken;
  NEXT_TOKEN(pCxt->pSql, sToken);
  if (TK_NK_LP == sToken.type) {
    CHECK_CODE(skipBoundColumns(pCxt));
    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);
  if (TK_NK_LP != sToken.type) {
    return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
  }
  CHECK_CODE(skipTagsClause(pCxt));

  return TSDB_CODE_SUCCESS;
}

static int32_t collectTableMetaKey(SInsertParseSyntaxCxt* pCxt, SToken* pTbToken) {
  SName name;
  CHECK_CODE(createSName(&name, pTbToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg));
  CHECK_CODE(reserveUserAuthInCacheExt(pCxt->pComCxt->pUser, &name, AUTH_TYPE_WRITE, pCxt->pMetaCache));
  CHECK_CODE(reserveTableMetaInCacheExt(&name, pCxt->pMetaCache));
  CHECK_CODE(reserveTableVgroupInCacheExt(&name, pCxt->pMetaCache));
  return TSDB_CODE_SUCCESS;
}

static int32_t parseInsertBodySyntax(SInsertParseSyntaxCxt* pCxt) {
  bool hasData = false;
  // for each table
  while (1) {
    SToken sToken;

    // pSql -> tb_name ...
    NEXT_TOKEN(pCxt->pSql, sToken);

    // no data in the sql string anymore.
    if (sToken.n == 0) {
      if (sToken.type && pCxt->pSql[0]) {
        return buildSyntaxErrMsg(&pCxt->msg, "invalid charactor in SQL", sToken.z);
      }

      if (!hasData) {
        return buildInvalidOperationMsg(&pCxt->msg, "no data in sql");
      }
      break;
    }

    hasData = false;

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

    // USING clause
    if (TK_USING == sToken.type) {
      NEXT_TOKEN(pCxt->pSql, sToken);
      CHECK_CODE(collectTableMetaKey(pCxt, &sToken));
      CHECK_CODE(skipUsingClause(pCxt));
      NEXT_TOKEN(pCxt->pSql, sToken);
    } else {
      CHECK_CODE(collectTableMetaKey(pCxt, &tbnameToken));
    }

    if (TK_NK_LP == sToken.type) {
      // pSql -> field1_name, ...)
      CHECK_CODE(skipBoundColumns(pCxt));
      NEXT_TOKEN(pCxt->pSql, sToken);
    }

    if (TK_VALUES == sToken.type) {
      // pSql -> (field1_value, ...) [(field1_value2, ...) ...]
      CHECK_CODE(skipValuesClause(pCxt));
      hasData = true;
      continue;
    }

    // FILE csv_file_path
    if (TK_FILE == sToken.type) {
      // pSql -> csv_file_path
      NEXT_TOKEN(pCxt->pSql, sToken);
      if (0 == sToken.n || (TK_NK_STRING != sToken.type && TK_NK_ID != sToken.type)) {
        return buildSyntaxErrMsg(&pCxt->msg, "file path is required following keyword FILE", sToken.z);
      }
      hasData = true;
      continue;
    }

    return buildSyntaxErrMsg(&pCxt->msg, "keyword VALUES or FILE is expected", sToken.z);
  }

  return TSDB_CODE_SUCCESS;
}

int32_t parseInsertSyntax(SParseContext* pContext, SQuery** pQuery) {
  SInsertParseSyntaxCxt context = {.pComCxt = pContext,
                                   .pSql = (char*)pContext->pSql,
                                   .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
                                   .pMetaCache = taosMemoryCalloc(1, sizeof(SParseMetaCache))};
  if (NULL == context.pMetaCache) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  int32_t code = skipInsertInto(&context.pSql, &context.msg);
  if (TSDB_CODE_SUCCESS == code) {
    code = parseInsertBodySyntax(&context);
  }
  if (TSDB_CODE_SUCCESS == code) {
    *pQuery = taosMemoryCalloc(1, sizeof(SQuery));
    if (NULL == *pQuery) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
    TSWAP((*pQuery)->pMetaCache, context.pMetaCache);
  }
  return code;
}

X
Xiaoyu Wang 已提交
1665 1666 1667 1668
int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,
                     int32_t msgBufLen) {
  SMsgBuf msg = {.buf = msgBuf, .len = msgBufLen};
  SToken  sToken;
D
stmt  
dapan1121 已提交
1669
  int32_t code = 0;
X
Xiaoyu Wang 已提交
1670 1671
  char*   tbName = NULL;

D
stmt  
dapan1121 已提交
1672
  NEXT_TOKEN(pTableName, sToken);
X
Xiaoyu Wang 已提交
1673

D
stmt  
dapan1121 已提交
1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684
  if (sToken.n == 0) {
    return buildInvalidOperationMsg(&msg, "empty table name");
  }

  code = createSName(pName, &sToken, acctId, dbName, &msg);
  if (code) {
    return code;
  }

  NEXT_TOKEN(pTableName, sToken);

D
stmt  
dapan1121 已提交
1685
  if (sToken.n > 0) {
D
stmt  
dapan1121 已提交
1686 1687 1688 1689 1690 1691 1692
    return buildInvalidOperationMsg(&msg, "table name format is wrong");
  }

  return TSDB_CODE_SUCCESS;
}

int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash) {
X
Xiaoyu Wang 已提交
1693 1694
  SVnodeModifOpStmt*  modifyNode = (SVnodeModifOpStmt*)pQuery->pRoot;
  int32_t             code = 0;
D
stmt  
dapan1121 已提交
1695
  SInsertParseContext insertCtx = {
X
Xiaoyu Wang 已提交
1696 1697 1698
      .pVgroupsHashObj = pVgHash,
      .pTableBlockHashObj = pBlockHash,
      .pOutput = (SVnodeModifOpStmt*)pQuery->pRoot,
D
stmt  
dapan1121 已提交
1699
  };
X
Xiaoyu Wang 已提交
1700

D
stmt  
dapan1121 已提交
1701 1702
  // merge according to vgId
  if (taosHashGetSize(insertCtx.pTableBlockHashObj) > 0) {
D
stmt  
dapan1121 已提交
1703
    CHECK_CODE(mergeTableDataBlocks(insertCtx.pTableBlockHashObj, modifyNode->payloadType, &insertCtx.pVgDataBlocks));
D
stmt  
dapan1121 已提交
1704 1705 1706 1707
  }

  CHECK_CODE(buildOutput(&insertCtx));

wmmhello's avatar
wmmhello 已提交
1708
  destroyBlockArrayList(insertCtx.pVgDataBlocks);
D
stmt  
dapan1121 已提交
1709 1710 1711
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1712 1713 1714 1715
int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tName, TAOS_MULTI_BIND* bind,
                           char* msgBuf, int32_t msgBufLen) {
  STableDataBlocks*   pDataBlock = (STableDataBlocks*)pBlock;
  SMsgBuf             pBuf = {.buf = msgBuf, .len = msgBufLen};
D
stmt  
dapan1121 已提交
1716 1717 1718 1719 1720
  SParsedDataColInfo* tags = (SParsedDataColInfo*)boundTags;
  if (NULL == tags) {
    return TSDB_CODE_QRY_APP_ERROR;
  }

C
Cary Xu 已提交
1721 1722
  SArray* pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal));
  if (!pTagArray) {
C
Cary Xu 已提交
1723
    return buildInvalidOperationMsg(&pBuf, "out of memory");
D
stmt  
dapan1121 已提交
1724 1725
  }

1726
  int32_t  code = TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1727
  SSchema* pSchema = pDataBlock->pTableMeta->schema;
D
stmt  
dapan1121 已提交
1728

1729
  bool  isJson = false;
wmmhello's avatar
wmmhello 已提交
1730
  STag* pTag = NULL;
D
dapan1121 已提交
1731

D
stmt  
dapan1121 已提交
1732 1733 1734 1735
  for (int c = 0; c < tags->numOfBound; ++c) {
    if (bind[c].is_null && bind[c].is_null[0]) {
      continue;
    }
X
Xiaoyu Wang 已提交
1736

X
Xiaoyu Wang 已提交
1737
    SSchema* pTagSchema = &pSchema[tags->boundColumns[c]];
1738
    int32_t  colLen = pTagSchema->bytes;
D
stmt  
dapan1121 已提交
1739 1740 1741
    if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
      colLen = bind[c].length[0];
    }
wmmhello's avatar
wmmhello 已提交
1742 1743 1744 1745 1746
    if (pTagSchema->type == TSDB_DATA_TYPE_JSON) {
      if (colLen > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
        code = buildSyntaxErrMsg(&pBuf, "json string too long than 4095", bind[c].buffer);
        goto end;
      }
X
Xiaoyu Wang 已提交
1747

wmmhello's avatar
wmmhello 已提交
1748
      isJson = true;
1749
      char* tmp = taosMemoryCalloc(1, colLen + 1);
wmmhello's avatar
wmmhello 已提交
1750
      memcpy(tmp, bind[c].buffer, colLen);
wmmhello's avatar
wmmhello 已提交
1751
      code = parseJsontoTagData(tmp, pTagArray, &pTag, &pBuf);
wmmhello's avatar
wmmhello 已提交
1752
      taosMemoryFree(tmp);
1753
      if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1754 1755
        goto end;
      }
1756
    } else {
wmmhello's avatar
wmmhello 已提交
1757
      STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
1758
      if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
wmmhello's avatar
wmmhello 已提交
1759 1760
        val.pData = (uint8_t*)bind[c].buffer;
        val.nData = colLen;
1761
      } else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
1762
        int32_t output = 0;
1763 1764
        void*   p = taosMemoryCalloc(1, colLen * TSDB_NCHAR_SIZE);
        if (p == NULL) {
wmmhello's avatar
wmmhello 已提交
1765 1766 1767 1768 1769 1770 1771 1772 1773 1774 1775
          code = TSDB_CODE_OUT_OF_MEMORY;
          goto end;
        }
        if (!taosMbsToUcs4(bind[c].buffer, colLen, (TdUcs4*)(p), pSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
          if (errno == E2BIG) {
            taosMemoryFree(p);
            code = generateSyntaxErrMsg(&pBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pSchema->name);
            goto end;
          }
          char buf[512] = {0};
          snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
wmmhello's avatar
wmmhello 已提交
1776
          taosMemoryFree(p);
wmmhello's avatar
wmmhello 已提交
1777
          code = buildSyntaxErrMsg(&pBuf, buf, bind[c].buffer);
wmmhello's avatar
wmmhello 已提交
1778 1779
          goto end;
        }
wmmhello's avatar
wmmhello 已提交
1780 1781
        val.pData = p;
        val.nData = output;
1782
      } else {
wmmhello's avatar
wmmhello 已提交
1783
        memcpy(&val.i64, bind[c].buffer, colLen);
wmmhello's avatar
wmmhello 已提交
1784
      }
wmmhello's avatar
wmmhello 已提交
1785
      taosArrayPush(pTagArray, &val);
wmmhello's avatar
wmmhello 已提交
1786
    }
D
stmt  
dapan1121 已提交
1787 1788
  }

wmmhello's avatar
wmmhello 已提交
1789
  if (!isJson && (code = tTagNew(pTagArray, 1, false, &pTag)) != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1790
    goto end;
D
stmt  
dapan1121 已提交
1791 1792 1793
  }

  SVCreateTbReq tbReq = {0};
wmmhello's avatar
wmmhello 已提交
1794 1795
  buildCreateTbReq(&tbReq, tName, pTag, suid);
  code = buildCreateTbMsg(pDataBlock, &tbReq);
D
stmt  
dapan1121 已提交
1796 1797
  destroyCreateSubTbReq(&tbReq);

wmmhello's avatar
wmmhello 已提交
1798 1799
end:
  for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
1800 1801
    STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
    if (p->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
1802 1803 1804
      taosMemoryFree(p->pData);
    }
  }
C
Cary Xu 已提交
1805
  taosArrayDestroy(pTagArray);
D
stmt  
dapan1121 已提交
1806

wmmhello's avatar
wmmhello 已提交
1807
  return code;
D
stmt  
dapan1121 已提交
1808 1809
}

X
Xiaoyu Wang 已提交
1810 1811 1812 1813
int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen) {
  STableDataBlocks*   pDataBlock = (STableDataBlocks*)pBlock;
  SSchema*            pSchema = getTableColumnSchema(pDataBlock->pTableMeta);
  int32_t             extendedRowSize = getExtendedRowSize(pDataBlock);
D
stmt  
dapan1121 已提交
1814 1815
  SParsedDataColInfo* spd = &pDataBlock->boundColumnInfo;
  SRowBuilder*        pBuilder = &pDataBlock->rowBuilder;
X
Xiaoyu Wang 已提交
1816 1817 1818 1819
  SMemParam           param = {.rb = pBuilder};
  SMsgBuf             pBuf = {.buf = msgBuf, .len = msgBufLen};
  int32_t             rowNum = bind->num;

D
stmt  
dapan1121 已提交
1820 1821
  CHECK_CODE(initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo));

D
stmt  
dapan1121 已提交
1822
  CHECK_CODE(allocateMemForSize(pDataBlock, extendedRowSize * bind->num));
X
Xiaoyu Wang 已提交
1823

D
stmt  
dapan1121 已提交
1824 1825 1826
  for (int32_t r = 0; r < bind->num; ++r) {
    STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size);  // skip the SSubmitBlk header
    tdSRowResetBuf(pBuilder, row);
X
Xiaoyu Wang 已提交
1827

D
stmt  
dapan1121 已提交
1828
    for (int c = 0; c < spd->numOfBound; ++c) {
X
Xiaoyu Wang 已提交
1829
      SSchema* pColSchema = &pSchema[spd->boundColumns[c]];
D
stmt  
dapan1121 已提交
1830 1831 1832 1833

      if (bind[c].num != rowNum) {
        return buildInvalidOperationMsg(&pBuf, "row number in each bind param should be the same");
      }
X
Xiaoyu Wang 已提交
1834

D
stmt  
dapan1121 已提交
1835 1836 1837 1838
      param.schema = pColSchema;
      getSTSRowAppendInfo(pBuilder->rowType, spd, c, &param.toffset, &param.colIdx);

      if (bind[c].is_null && bind[c].is_null[r]) {
D
stmt  
dapan1121 已提交
1839 1840 1841
        if (pColSchema->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
          return buildInvalidOperationMsg(&pBuf, "primary timestamp should not be NULL");
        }
X
Xiaoyu Wang 已提交
1842

D
stmt  
dapan1121 已提交
1843 1844
        CHECK_CODE(MemRowAppend(&pBuf, NULL, 0, &param));
      } else {
D
dapan1121 已提交
1845 1846 1847 1848
        if (bind[c].buffer_type != pColSchema->type) {
          return buildInvalidOperationMsg(&pBuf, "column type mis-match with buffer type");
        }

D
stmt  
dapan1121 已提交
1849 1850 1851 1852
        int32_t colLen = pColSchema->bytes;
        if (IS_VAR_DATA_TYPE(pColSchema->type)) {
          colLen = bind[c].length[r];
        }
X
Xiaoyu Wang 已提交
1853 1854

        CHECK_CODE(MemRowAppend(&pBuf, (char*)bind[c].buffer + bind[c].buffer_length * r, colLen, &param));
D
stmt  
dapan1121 已提交
1855
      }
X
Xiaoyu Wang 已提交
1856

D
stmt  
dapan1121 已提交
1857 1858
      if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
        TSKEY tsKey = TD_ROW_KEY(row);
X
Xiaoyu Wang 已提交
1859
        checkTimestamp(pDataBlock, (const char*)&tsKey);
D
stmt  
dapan1121 已提交
1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870
      }
    }
    // set the null value for the columns that do not assign values
    if ((spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
      for (int32_t i = 0; i < spd->numOfCols; ++i) {
        if (spd->cols[i].valStat == VAL_STAT_NONE) {  // the primary TS key is not VAL_STAT_NONE
          tdAppendColValToTpRow(pBuilder, TD_VTYPE_NONE, getNullValue(pSchema[i].type), true, pSchema[i].type, i,
                                spd->cols[i].toffset);
        }
      }
    }
C
Cary Xu 已提交
1871 1872
#ifdef TD_DEBUG_PRINT_ROW
    STSchema* pSTSchema = tdGetSTSChemaFromSSChema(&pSchema, spd->numOfCols);
C
Cary Xu 已提交
1873
    tdSRowPrint(row, pSTSchema, __func__);
C
Cary Xu 已提交
1874 1875
    taosMemoryFree(pSTSchema);
#endif
D
stmt  
dapan1121 已提交
1876 1877
    pDataBlock->size += extendedRowSize;
  }
D
stmt  
dapan1121 已提交
1878

X
Xiaoyu Wang 已提交
1879
  SSubmitBlk* pBlocks = (SSubmitBlk*)(pDataBlock->pData);
D
stmt  
dapan1121 已提交
1880 1881 1882 1883 1884 1885 1886
  if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, pDataBlock, bind->num)) {
    return buildInvalidOperationMsg(&pBuf, "too many rows in sql, total number of rows should be less than 32767");
  }

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1887 1888 1889 1890 1891
int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,
                                int32_t rowNum) {
  STableDataBlocks*   pDataBlock = (STableDataBlocks*)pBlock;
  SSchema*            pSchema = getTableColumnSchema(pDataBlock->pTableMeta);
  int32_t             extendedRowSize = getExtendedRowSize(pDataBlock);
D
stmt  
dapan1121 已提交
1892 1893
  SParsedDataColInfo* spd = &pDataBlock->boundColumnInfo;
  SRowBuilder*        pBuilder = &pDataBlock->rowBuilder;
X
Xiaoyu Wang 已提交
1894 1895 1896 1897
  SMemParam           param = {.rb = pBuilder};
  SMsgBuf             pBuf = {.buf = msgBuf, .len = msgBufLen};
  bool                rowStart = (0 == colIdx);
  bool                rowEnd = ((colIdx + 1) == spd->numOfBound);
D
stmt  
dapan1121 已提交
1898 1899 1900 1901 1902

  if (rowStart) {
    CHECK_CODE(initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo));
    CHECK_CODE(allocateMemForSize(pDataBlock, extendedRowSize * bind->num));
  }
X
Xiaoyu Wang 已提交
1903

D
stmt  
dapan1121 已提交
1904 1905 1906 1907 1908 1909 1910
  for (int32_t r = 0; r < bind->num; ++r) {
    STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size + extendedRowSize * r);  // skip the SSubmitBlk header
    if (rowStart) {
      tdSRowResetBuf(pBuilder, row);
    } else {
      tdSRowGetBuf(pBuilder, row);
    }
D
stmt  
dapan1121 已提交
1911

X
Xiaoyu Wang 已提交
1912
    SSchema* pColSchema = &pSchema[spd->boundColumns[colIdx]];
D
stmt  
dapan1121 已提交
1913 1914 1915 1916

    if (bind->num != rowNum) {
      return buildInvalidOperationMsg(&pBuf, "row number in each bind param should be the same");
    }
X
Xiaoyu Wang 已提交
1917

D
stmt  
dapan1121 已提交
1918 1919 1920 1921 1922 1923 1924
    param.schema = pColSchema;
    getSTSRowAppendInfo(pBuilder->rowType, spd, colIdx, &param.toffset, &param.colIdx);

    if (bind->is_null && bind->is_null[r]) {
      if (pColSchema->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
        return buildInvalidOperationMsg(&pBuf, "primary timestamp should not be NULL");
      }
X
Xiaoyu Wang 已提交
1925

D
stmt  
dapan1121 已提交
1926 1927
      CHECK_CODE(MemRowAppend(&pBuf, NULL, 0, &param));
    } else {
D
dapan1121 已提交
1928 1929 1930 1931
      if (bind->buffer_type != pColSchema->type) {
        return buildInvalidOperationMsg(&pBuf, "column type mis-match with buffer type");
      }

D
stmt  
dapan1121 已提交
1932 1933 1934 1935
      int32_t colLen = pColSchema->bytes;
      if (IS_VAR_DATA_TYPE(pColSchema->type)) {
        colLen = bind->length[r];
      }
X
Xiaoyu Wang 已提交
1936 1937

      CHECK_CODE(MemRowAppend(&pBuf, (char*)bind->buffer + bind->buffer_length * r, colLen, &param));
D
stmt  
dapan1121 已提交
1938
    }
X
Xiaoyu Wang 已提交
1939

D
stmt  
dapan1121 已提交
1940 1941
    if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
      TSKEY tsKey = TD_ROW_KEY(row);
X
Xiaoyu Wang 已提交
1942
      checkTimestamp(pDataBlock, (const char*)&tsKey);
D
stmt  
dapan1121 已提交
1943
    }
X
Xiaoyu Wang 已提交
1944

D
stmt  
dapan1121 已提交
1945 1946 1947 1948 1949 1950 1951 1952
    // set the null value for the columns that do not assign values
    if (rowEnd && (spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
      for (int32_t i = 0; i < spd->numOfCols; ++i) {
        if (spd->cols[i].valStat == VAL_STAT_NONE) {  // the primary TS key is not VAL_STAT_NONE
          tdAppendColValToTpRow(pBuilder, TD_VTYPE_NONE, getNullValue(pSchema[i].type), true, pSchema[i].type, i,
                                spd->cols[i].toffset);
        }
      }
X
Xiaoyu Wang 已提交
1953
    }
C
Cary Xu 已提交
1954 1955

#ifdef TD_DEBUG_PRINT_ROW
X
Xiaoyu Wang 已提交
1956
    if (rowEnd) {
C
Cary Xu 已提交
1957
      STSchema* pSTSchema = tdGetSTSChemaFromSSChema(&pSchema, spd->numOfCols);
C
Cary Xu 已提交
1958
      tdSRowPrint(row, pSTSchema, __func__);
C
Cary Xu 已提交
1959 1960 1961
      taosMemoryFree(pSTSchema);
    }
#endif
D
stmt  
dapan1121 已提交
1962 1963
  }

D
stmt  
dapan1121 已提交
1964 1965 1966
  if (rowEnd) {
    pDataBlock->size += extendedRowSize * bind->num;

X
Xiaoyu Wang 已提交
1967
    SSubmitBlk* pBlocks = (SSubmitBlk*)(pDataBlock->pData);
D
stmt  
dapan1121 已提交
1968 1969 1970 1971 1972 1973 1974 1975
    if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, pDataBlock, bind->num)) {
      return buildInvalidOperationMsg(&pBuf, "too many rows in sql, total number of rows should be less than 32767");
    }
  }

  return TSDB_CODE_SUCCESS;
}

1976 1977
int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_t* fieldNum, TAOS_FIELD_E** fields,
                         uint8_t timePrec) {
D
stmt  
dapan1121 已提交
1978 1979 1980 1981 1982 1983
  if (fields) {
    *fields = taosMemoryCalloc(boundInfo->numOfBound, sizeof(TAOS_FIELD));
    if (NULL == *fields) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

D
dapan1121 已提交
1984 1985
    SSchema* schema = &pSchema[boundInfo->boundColumns[0]];
    if (TSDB_DATA_TYPE_TIMESTAMP == schema->type) {
D
dapan1121 已提交
1986 1987
      (*fields)[0].precision = timePrec;
    }
1988

D
stmt  
dapan1121 已提交
1989
    for (int32_t i = 0; i < boundInfo->numOfBound; ++i) {
D
dapan1121 已提交
1990 1991 1992 1993
      schema = &pSchema[boundInfo->boundColumns[i]];
      strcpy((*fields)[i].name, schema->name);
      (*fields)[i].type = schema->type;
      (*fields)[i].bytes = schema->bytes;
D
stmt  
dapan1121 已提交
1994
    }
D
stmt  
dapan1121 已提交
1995 1996 1997 1998 1999 2000 2001
  }

  *fieldNum = boundInfo->numOfBound;

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
2002
int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD_E** fields) {
X
Xiaoyu Wang 已提交
2003
  STableDataBlocks*   pDataBlock = (STableDataBlocks*)pBlock;
D
stmt  
dapan1121 已提交
2004 2005 2006 2007
  SParsedDataColInfo* tags = (SParsedDataColInfo*)boundTags;
  if (NULL == tags) {
    return TSDB_CODE_QRY_APP_ERROR;
  }
X
Xiaoyu Wang 已提交
2008 2009

  SSchema* pSchema = getTableTagSchema(pDataBlock->pTableMeta);
D
stmt  
dapan1121 已提交
2010 2011 2012 2013 2014 2015 2016
  if (tags->numOfBound <= 0) {
    *fieldNum = 0;
    *fields = NULL;

    return TSDB_CODE_SUCCESS;
  }

D
dapan1121 已提交
2017
  CHECK_CODE(buildBoundFields(tags, pSchema, fieldNum, fields, 0));
X
Xiaoyu Wang 已提交
2018

D
stmt  
dapan1121 已提交
2019 2020 2021
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
2022
int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_E** fields) {
X
Xiaoyu Wang 已提交
2023 2024
  STableDataBlocks* pDataBlock = (STableDataBlocks*)pBlock;
  SSchema*          pSchema = getTableColumnSchema(pDataBlock->pTableMeta);
D
stmt  
dapan1121 已提交
2025 2026
  if (pDataBlock->boundColumnInfo.numOfBound <= 0) {
    *fieldNum = 0;
D
stmt  
dapan1121 已提交
2027 2028 2029
    if (fields) {
      *fields = NULL;
    }
D
stmt  
dapan1121 已提交
2030 2031 2032 2033

    return TSDB_CODE_SUCCESS;
  }

2034 2035
  CHECK_CODE(buildBoundFields(&pDataBlock->boundColumnInfo, pSchema, fieldNum, fields,
                              pDataBlock->pTableMeta->tableInfo.precision));
X
Xiaoyu Wang 已提交
2036

D
stmt  
dapan1121 已提交
2037 2038 2039
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
2040
// schemaless logic start
D
stmt  
dapan1121 已提交
2041

wmmhello's avatar
wmmhello 已提交
2042
typedef struct SmlExecTableHandle {
X
Xiaoyu Wang 已提交
2043 2044
  SParsedDataColInfo tags;          // each table
  SVCreateTbReq      createTblReq;  // each table
wmmhello's avatar
wmmhello 已提交
2045
} SmlExecTableHandle;
wmmhello's avatar
wmmhello 已提交
2046

wmmhello's avatar
wmmhello 已提交
2047
typedef struct SmlExecHandle {
2048 2049 2050
  SHashObj*          pBlockHash;
  SmlExecTableHandle tableExecHandle;
  SQuery*            pQuery;
wmmhello's avatar
wmmhello 已提交
2051
} SSmlExecHandle;
wmmhello's avatar
wmmhello 已提交
2052

wmmhello's avatar
wmmhello 已提交
2053 2054 2055 2056 2057 2058
static void smlDestroyTableHandle(void* pHandle) {
  SmlExecTableHandle* handle = (SmlExecTableHandle*)pHandle;
  destroyBoundColumnInfo(&handle->tags);
  destroyCreateSubTbReq(&handle->createTblReq);
}

X
Xiaoyu Wang 已提交
2059
static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SSchema* pSchema) {
wmmhello's avatar
wmmhello 已提交
2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070 2071
  col_id_t nCols = pColList->numOfCols;

  pColList->numOfBound = 0;
  pColList->boundNullLen = 0;
  memset(pColList->boundColumns, 0, sizeof(col_id_t) * nCols);
  for (col_id_t i = 0; i < nCols; ++i) {
    pColList->cols[i].valStat = VAL_STAT_NONE;
  }

  bool     isOrdered = true;
  col_id_t lastColIdx = -1;  // last column found
  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
X
Xiaoyu Wang 已提交
2072 2073
    SSmlKv*  kv = taosArrayGetP(cols, i);
    SToken   sToken = {.n = kv->keyLen, .z = (char*)kv->key};
wmmhello's avatar
wmmhello 已提交
2074 2075 2076 2077 2078 2079 2080 2081 2082 2083 2084 2085 2086 2087
    col_id_t t = lastColIdx + 1;
    col_id_t index = findCol(&sToken, t, nCols, pSchema);
    if (index < 0 && t > 0) {
      index = findCol(&sToken, 0, t, pSchema);
      isOrdered = false;
    }
    if (index < 0) {
      return TSDB_CODE_SML_INVALID_DATA;
    }
    if (pColList->cols[index].valStat == VAL_STAT_HAS) {
      return TSDB_CODE_SML_INVALID_DATA;
    }
    lastColIdx = index;
    pColList->cols[index].valStat = VAL_STAT_HAS;
X
Xiaoyu Wang 已提交
2088
    pColList->boundColumns[pColList->numOfBound] = index;
wmmhello's avatar
wmmhello 已提交
2089 2090 2091 2092 2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105 2106 2107 2108 2109 2110 2111 2112 2113 2114 2115 2116 2117 2118 2119 2120 2121
    ++pColList->numOfBound;
    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;
    }
  }

  pColList->orderStatus = isOrdered ? ORDER_STATUS_ORDERED : ORDER_STATUS_DISORDERED;

  if (!isOrdered) {
    pColList->colIdxInfo = taosMemoryCalloc(pColList->numOfBound, sizeof(SBoundIdxInfo));
    if (NULL == pColList->colIdxInfo) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    SBoundIdxInfo* pColIdx = pColList->colIdxInfo;
    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
      pColIdx[i].schemaColIdx = pColList->boundColumns[i];
      pColIdx[i].boundIdx = i;
    }
    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar);
    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
      pColIdx[i].finalIdx = i;
    }
    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar);
  }

X
Xiaoyu Wang 已提交
2122
  if (pColList->numOfCols > pColList->numOfBound) {
wmmhello's avatar
wmmhello 已提交
2123 2124 2125 2126 2127 2128 2129
    memset(&pColList->boundColumns[pColList->numOfBound], 0,
           sizeof(col_id_t) * (pColList->numOfCols - pColList->numOfBound));
  }

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
2130 2131 2132 2133 2134 2135 2136 2137 2138 2139 2140
/**
 * @brief No json tag for schemaless
 *
 * @param cols
 * @param tags
 * @param pSchema
 * @param ppTag
 * @param msg
 * @return int32_t
 */
static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* pSchema, STag** ppTag, SMsgBuf* msg) {
C
Cary Xu 已提交
2141 2142
  SArray* pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal));
  if (!pTagArray) {
wmmhello's avatar
wmmhello 已提交
2143 2144 2145
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }

wmmhello's avatar
wmmhello 已提交
2146
  int32_t code = TSDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2147
  for (int i = 0; i < tags->numOfBound; ++i) {
X
Xiaoyu Wang 已提交
2148
    SSchema* pTagSchema = &pSchema[tags->boundColumns[i]];
2149
    SSmlKv*  kv = taosArrayGetP(cols, i);
wmmhello's avatar
wmmhello 已提交
2150

wmmhello's avatar
wmmhello 已提交
2151
    STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
2152 2153
    if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
      val.pData = (uint8_t*)kv->value;
wmmhello's avatar
wmmhello 已提交
2154
      val.nData = kv->length;
2155
    } else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
2156
      int32_t output = 0;
2157 2158
      void*   p = taosMemoryCalloc(1, pTagSchema->bytes - VARSTR_HEADER_SIZE);
      if (p == NULL) {
wmmhello's avatar
wmmhello 已提交
2159 2160 2161
        code = TSDB_CODE_OUT_OF_MEMORY;
        goto end;
      }
wmmhello's avatar
wmmhello 已提交
2162
      if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)(p), pTagSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
wmmhello's avatar
wmmhello 已提交
2163 2164
        if (errno == E2BIG) {
          taosMemoryFree(p);
wmmhello's avatar
wmmhello 已提交
2165
          code = generateSyntaxErrMsg(msg, TSDB_CODE_PAR_VALUE_TOO_LONG, pTagSchema->name);
wmmhello's avatar
wmmhello 已提交
2166 2167 2168 2169 2170 2171 2172 2173 2174 2175
          goto end;
        }
        char buf[512] = {0};
        snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
        taosMemoryFree(p);
        code = buildSyntaxErrMsg(msg, buf, kv->value);
        goto end;
      }
      val.pData = p;
      val.nData = output;
2176
    } else {
wmmhello's avatar
wmmhello 已提交
2177
      memcpy(&val.i64, &(kv->value), kv->length);
wmmhello's avatar
wmmhello 已提交
2178
    }
wmmhello's avatar
wmmhello 已提交
2179
    taosArrayPush(pTagArray, &val);
wmmhello's avatar
wmmhello 已提交
2180 2181
  }

wmmhello's avatar
wmmhello 已提交
2182 2183 2184
  code = tTagNew(pTagArray, 1, false, ppTag);
end:
  for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
2185 2186
    STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
    if (p->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
2187 2188
      taosMemoryFree(p->pData);
    }
wmmhello's avatar
wmmhello 已提交
2189
  }
C
Cary Xu 已提交
2190
  taosArrayDestroy(pTagArray);
wmmhello's avatar
wmmhello 已提交
2191
  return code;
wmmhello's avatar
wmmhello 已提交
2192 2193
}

2194 2195
int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
                    char* tableName, char* msgBuf, int16_t msgBufLen) {
wmmhello's avatar
wmmhello 已提交
2196 2197
  SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};

X
Xiaoyu Wang 已提交
2198
  SSmlExecHandle* smlHandle = (SSmlExecHandle*)handle;
2199 2200
  smlDestroyTableHandle(&smlHandle->tableExecHandle);  // free for each table
  SSchema* pTagsSchema = getTableTagSchema(pTableMeta);
wmmhello's avatar
wmmhello 已提交
2201 2202
  setBoundColumnInfo(&smlHandle->tableExecHandle.tags, pTagsSchema, getNumOfTags(pTableMeta));
  int ret = smlBoundColumnData(tags, &smlHandle->tableExecHandle.tags, pTagsSchema);
X
Xiaoyu Wang 已提交
2203
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2204 2205 2206
    buildInvalidOperationMsg(&pBuf, "bound tags error");
    return ret;
  }
C
Cary Xu 已提交
2207 2208
  STag* pTag = NULL;
  ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tags, pTagsSchema, &pTag, &pBuf);
X
Xiaoyu Wang 已提交
2209
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2210 2211
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2212

C
Cary Xu 已提交
2213
  buildCreateTbReq(&smlHandle->tableExecHandle.createTblReq, tableName, pTag, pTableMeta->suid);
wmmhello's avatar
wmmhello 已提交
2214

wmmhello's avatar
wmmhello 已提交
2215
  STableDataBlocks* pDataBlock = NULL;
X
Xiaoyu Wang 已提交
2216 2217
  ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid),
                             TSDB_DEFAULT_PAYLOAD_SIZE, sizeof(SSubmitBlk), getTableInfo(pTableMeta).rowSize,
wmmhello's avatar
wmmhello 已提交
2218
                             pTableMeta, &pDataBlock, NULL, &smlHandle->tableExecHandle.createTblReq);
X
Xiaoyu Wang 已提交
2219
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2220 2221 2222
    buildInvalidOperationMsg(&pBuf, "create data block error");
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2223 2224 2225

  SSchema* pSchema = getTableColumnSchema(pTableMeta);

2226
  ret = smlBoundColumnData(colsSchema, &pDataBlock->boundColumnInfo, pSchema);
X
Xiaoyu Wang 已提交
2227
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2228 2229 2230
    buildInvalidOperationMsg(&pBuf, "bound cols error");
    return ret;
  }
X
Xiaoyu Wang 已提交
2231
  int32_t             extendedRowSize = getExtendedRowSize(pDataBlock);
wmmhello's avatar
wmmhello 已提交
2232 2233
  SParsedDataColInfo* spd = &pDataBlock->boundColumnInfo;
  SRowBuilder*        pBuilder = &pDataBlock->rowBuilder;
X
Xiaoyu Wang 已提交
2234
  SMemParam           param = {.rb = pBuilder};
wmmhello's avatar
wmmhello 已提交
2235 2236 2237

  initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo);

2238
  int32_t rowNum = taosArrayGetSize(cols);
2239
  if (rowNum <= 0) {
wmmhello's avatar
wmmhello 已提交
2240 2241
    return buildInvalidOperationMsg(&pBuf, "cols size <= 0");
  }
wmmhello's avatar
wmmhello 已提交
2242
  ret = allocateMemForSize(pDataBlock, extendedRowSize * rowNum);
X
Xiaoyu Wang 已提交
2243
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2244 2245 2246
    buildInvalidOperationMsg(&pBuf, "allocate memory error");
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2247 2248 2249
  for (int32_t r = 0; r < rowNum; ++r) {
    STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size);  // skip the SSubmitBlk header
    tdSRowResetBuf(pBuilder, row);
2250
    void*  rowData = taosArrayGetP(cols, r);
2251
    size_t rowDataSize = 0;
2252
    if (format) {
2253
      rowDataSize = taosArrayGetSize(rowData);
wmmhello's avatar
wmmhello 已提交
2254
    }
wmmhello's avatar
wmmhello 已提交
2255 2256

    // 1. set the parsed value from sql string
2257
    for (int c = 0, j = 0; c < spd->numOfBound; ++c) {
2258
      SSchema* pColSchema = &pSchema[spd->boundColumns[c]];
wmmhello's avatar
wmmhello 已提交
2259 2260 2261 2262

      param.schema = pColSchema;
      getSTSRowAppendInfo(pBuilder->rowType, spd, c, &param.toffset, &param.colIdx);

X
Xiaoyu Wang 已提交
2263 2264 2265
      SSmlKv* kv = NULL;
      if (format) {
        if (j < rowDataSize) {
2266
          kv = taosArrayGetP(rowData, j);
X
Xiaoyu Wang 已提交
2267 2268
          if (rowDataSize != spd->numOfBound &&
              (kv->keyLen != strlen(pColSchema->name) || strncmp(kv->key, pColSchema->name, kv->keyLen) != 0)) {
2269
            kv = NULL;
X
Xiaoyu Wang 已提交
2270
          } else {
2271
            j++;
2272
          }
wmmhello's avatar
wmmhello 已提交
2273
        }
X
Xiaoyu Wang 已提交
2274 2275 2276
      } else {
        void** p = taosHashGet(rowData, pColSchema->name, strlen(pColSchema->name));
        if (p) kv = *p;
wmmhello's avatar
wmmhello 已提交
2277
      }
wmmhello's avatar
wmmhello 已提交
2278

2279
      if (!kv || kv->length == 0) {
wmmhello's avatar
wmmhello 已提交
2280 2281
        MemRowAppend(&pBuf, NULL, 0, &param);
      } else {
wmmhello's avatar
wmmhello 已提交
2282 2283
        int32_t colLen = kv->length;
        if (pColSchema->type == TSDB_DATA_TYPE_TIMESTAMP) {
2284
          kv->i = convertTimePrecision(kv->i, TSDB_TIME_PRECISION_NANO, pTableMeta->tableInfo.precision);
wmmhello's avatar
wmmhello 已提交
2285 2286
        }

2287
        if (IS_VAR_DATA_TYPE(kv->type)) {
wmmhello's avatar
wmmhello 已提交
2288
          MemRowAppend(&pBuf, kv->value, colLen, &param);
2289
        } else {
wmmhello's avatar
wmmhello 已提交
2290 2291
          MemRowAppend(&pBuf, &(kv->value), colLen, &param);
        }
wmmhello's avatar
wmmhello 已提交
2292 2293 2294 2295
      }

      if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
        TSKEY tsKey = TD_ROW_KEY(row);
X
Xiaoyu Wang 已提交
2296
        checkTimestamp(pDataBlock, (const char*)&tsKey);
wmmhello's avatar
wmmhello 已提交
2297 2298 2299 2300 2301 2302 2303 2304 2305 2306 2307 2308 2309 2310 2311 2312
      }
    }

    // set the null value for the columns that do not assign values
    if ((spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
      for (int32_t i = 0; i < spd->numOfCols; ++i) {
        if (spd->cols[i].valStat == VAL_STAT_NONE) {  // the primary TS key is not VAL_STAT_NONE
          tdAppendColValToTpRow(pBuilder, TD_VTYPE_NONE, getNullValue(pSchema[i].type), true, pSchema[i].type, i,
                                spd->cols[i].toffset);
        }
      }
    }

    pDataBlock->size += extendedRowSize;
  }

X
Xiaoyu Wang 已提交
2313
  SSubmitBlk* pBlocks = (SSubmitBlk*)(pDataBlock->pData);
wmmhello's avatar
wmmhello 已提交
2314 2315 2316 2317 2318 2319 2320
  if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, pDataBlock, rowNum)) {
    return buildInvalidOperationMsg(&pBuf, "too many rows in sql, total number of rows should be less than 32767");
  }

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
2321 2322 2323
void* smlInitHandle(SQuery* pQuery) {
  SSmlExecHandle* handle = taosMemoryCalloc(1, sizeof(SSmlExecHandle));
  if (!handle) return NULL;
wmmhello's avatar
wmmhello 已提交
2324 2325 2326 2327 2328 2329
  handle->pBlockHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
  handle->pQuery = pQuery;

  return handle;
}

X
Xiaoyu Wang 已提交
2330 2331 2332
void smlDestroyHandle(void* pHandle) {
  if (!pHandle) return;
  SSmlExecHandle* handle = (SSmlExecHandle*)pHandle;
wmmhello's avatar
wmmhello 已提交
2333
  destroyBlockHashmap(handle->pBlockHash);
wmmhello's avatar
wmmhello 已提交
2334
  smlDestroyTableHandle(&handle->tableExecHandle);
wmmhello's avatar
wmmhello 已提交
2335 2336 2337 2338
  taosMemoryFree(handle);
}

int32_t smlBuildOutput(void* handle, SHashObj* pVgHash) {
X
Xiaoyu Wang 已提交
2339
  SSmlExecHandle* smlHandle = (SSmlExecHandle*)handle;
wmmhello's avatar
wmmhello 已提交
2340 2341
  return qBuildStmtOutput(smlHandle->pQuery, pVgHash, smlHandle->pBlockHash);
}
wmmhello's avatar
wmmhello 已提交
2342
// schemaless logic end