parInsert.c 85.8 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 52 53 54 55 56
typedef struct SInsertParseBaseContext {
  SParseContext* pComCxt;
  char*          pSql;
  SMsgBuf        msg;
} SInsertParseBaseContext;

57
typedef struct SInsertParseContext {
X
Xiaoyu Wang 已提交
58 59 60 61 62 63 64 65 66 67
  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 已提交
68
  SHashObj*          pTableNameHashObj;   // global
D
dapan1121 已提交
69
  SHashObj*          pDbFNameHashObj;     // global
X
Xiaoyu Wang 已提交
70
  int32_t            totalNum;
X
Xiaoyu Wang 已提交
71
  SVnodeModifOpStmt* pOutput;
X
Xiaoyu Wang 已提交
72
  SStmtCallback*     pStmtCb;
73
  SParseMetaCache*   pMetaCache;
wmmhello's avatar
wmmhello 已提交
74
  char               sTableName[TSDB_TABLE_NAME_LEN];
75 76
} SInsertParseContext;

77 78 79 80 81 82 83
typedef struct SInsertParseSyntaxCxt {
  SParseContext*   pComCxt;
  char*            pSql;
  SMsgBuf          msg;
  SParseMetaCache* pMetaCache;
} SInsertParseSyntaxCxt;

H
refact  
Hongze Cheng 已提交
84
typedef int32_t (*_row_append_fn_t)(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param);
X
Xiaoyu Wang 已提交
85 86 87 88

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

D
stmt  
dapan1121 已提交
89
typedef struct SKvParam {
C
Cary Xu 已提交
90
  int16_t  pos;
C
Cary Xu 已提交
91
  SArray*  pTagVals;
C
Cary Xu 已提交
92 93
  SSchema* schema;
  char     buf[TSDB_MAX_TAGS_LEN];
D
stmt  
dapan1121 已提交
94 95 96 97 98 99 100 101 102
} SKvParam;

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

X
Xiaoyu Wang 已提交
103 104 105
#define CHECK_CODE(expr)             \
  do {                               \
    int32_t code = expr;             \
D
stmt  
dapan1121 已提交
106
    if (TSDB_CODE_SUCCESS != code) { \
X
Xiaoyu Wang 已提交
107 108
      return code;                   \
    }                                \
D
stmt  
dapan1121 已提交
109 110
  } while (0)

111
static int32_t skipInsertInto(char** pSql, SMsgBuf* pMsg) {
112
  SToken sToken;
113
  NEXT_TOKEN(*pSql, sToken);
114
  if (TK_INSERT != sToken.type && TK_IMPORT != sToken.type) {
115
    return buildSyntaxErrMsg(pMsg, "keyword INSERT is expected", sToken.z);
116
  }
117
  NEXT_TOKEN(*pSql, sToken);
118
  if (TK_INTO != sToken.type) {
119
    return buildSyntaxErrMsg(pMsg, "keyword INTO is expected", sToken.z);
120 121 122 123
  }
  return TSDB_CODE_SUCCESS;
}

D
stmt  
dapan1121 已提交
124
static int32_t createSName(SName* pName, SToken* pTableName, int32_t acctId, const char* dbName, SMsgBuf* pMsgBuf) {
X
Xiaoyu Wang 已提交
125 126 127
  const char* msg1 = "name too long";
  const char* msg2 = "invalid database name";
  const char* msg3 = "db is not specified";
128
  const char* msg4 = "invalid table name";
X
Xiaoyu Wang 已提交
129

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

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

    int32_t dbLen = p - pTableName->z;
X
Xiaoyu Wang 已提交
137 138 139 140
    if (dbLen <= 0) {
      return buildInvalidOperationMsg(pMsgBuf, msg2);
    }
    char name[TSDB_DB_FNAME_LEN] = {0};
X
Xiaoyu Wang 已提交
141 142 143
    strncpy(name, pTableName->z, dbLen);
    dbLen = strdequote(name);

D
stmt  
dapan1121 已提交
144
    code = tNameSetDbName(pName, acctId, name, dbLen);
X
Xiaoyu Wang 已提交
145 146 147 148 149
    if (code != TSDB_CODE_SUCCESS) {
      return buildInvalidOperationMsg(pMsgBuf, msg1);
    }

    int32_t tbLen = pTableName->n - dbLen - 1;
150 151 152 153
    if (tbLen <= 0) {
      return buildInvalidOperationMsg(pMsgBuf, msg4);
    }

154
    char tbname[TSDB_TABLE_FNAME_LEN] = {0};
X
Xiaoyu Wang 已提交
155
    strncpy(tbname, p + 1, tbLen);
H
refact  
Hongze Cheng 已提交
156
    /*tbLen = */ strdequote(tbname);
X
Xiaoyu Wang 已提交
157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172

    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 已提交
173
    if (dbName == NULL) {
X
Xiaoyu Wang 已提交
174 175 176
      return buildInvalidOperationMsg(pMsgBuf, msg3);
    }

D
stmt  
dapan1121 已提交
177
    code = tNameSetDbName(pName, acctId, dbName, strlen(dbName));
X
Xiaoyu Wang 已提交
178 179 180 181 182 183 184 185 186 187 188 189 190 191
    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;
}

192 193
static int32_t checkAuth(SInsertParseContext* pCxt, char* pDbFname, bool* pPass) {
  SParseContext* pBasicCtx = pCxt->pComCxt;
X
Xiaoyu Wang 已提交
194
  if (pBasicCtx->async) {
195 196
    return getUserAuthFromCache(pCxt->pMetaCache, pBasicCtx->pUser, pDbFname, AUTH_TYPE_WRITE, pPass);
  }
X
Xiaoyu Wang 已提交
197
  SRequestConnInfo conn = {.pTrans = pBasicCtx->pTransporter,
D
dapan1121 已提交
198 199 200 201 202
                           .requestId = pBasicCtx->requestId,
                           .requestObjRefId = pBasicCtx->requestRid,
                           .mgmtEps = pBasicCtx->mgmtEpSet};

  return catalogChkAuth(pBasicCtx->pCatalog, &conn, pBasicCtx->pUser, pDbFname, AUTH_TYPE_WRITE, pPass);
203 204 205 206
}

static int32_t getTableSchema(SInsertParseContext* pCxt, SName* pTbName, bool isStb, STableMeta** pTableMeta) {
  SParseContext* pBasicCtx = pCxt->pComCxt;
X
Xiaoyu Wang 已提交
207
  if (pBasicCtx->async) {
208 209
    return getTableMetaFromCache(pCxt->pMetaCache, pTbName, pTableMeta);
  }
X
Xiaoyu Wang 已提交
210
  SRequestConnInfo conn = {.pTrans = pBasicCtx->pTransporter,
D
dapan1121 已提交
211 212 213
                           .requestId = pBasicCtx->requestId,
                           .requestObjRefId = pBasicCtx->requestRid,
                           .mgmtEps = pBasicCtx->mgmtEpSet};
X
Xiaoyu Wang 已提交
214

215
  if (isStb) {
D
dapan1121 已提交
216
    return catalogGetSTableMeta(pBasicCtx->pCatalog, &conn, pTbName, pTableMeta);
217
  }
D
dapan1121 已提交
218
  return catalogGetTableMeta(pBasicCtx->pCatalog, &conn, pTbName, pTableMeta);
219 220 221
}

static int32_t getTableVgroup(SInsertParseContext* pCxt, SName* pTbName, SVgroupInfo* pVg) {
H
Haojun Liao 已提交
222
  SParseContext* pBasicCtx = pCxt->pComCxt;
X
Xiaoyu Wang 已提交
223
  if (pBasicCtx->async) {
224 225
    return getTableVgroupFromCache(pCxt->pMetaCache, pTbName, pVg);
  }
X
Xiaoyu Wang 已提交
226
  SRequestConnInfo conn = {.pTrans = pBasicCtx->pTransporter,
D
dapan1121 已提交
227 228 229 230
                           .requestId = pBasicCtx->requestId,
                           .requestObjRefId = pBasicCtx->requestRid,
                           .mgmtEps = pBasicCtx->mgmtEpSet};
  return catalogGetTableHashVgroup(pBasicCtx->pCatalog, &conn, pTbName, pVg);
231
}
D
dapan 已提交
232

233
static int32_t getTableMetaImpl(SInsertParseContext* pCxt, SName* name, char* dbFname, bool isStb) {
D
dapan 已提交
234
  bool pass = false;
235
  CHECK_CODE(checkAuth(pCxt, dbFname, &pass));
D
dapan 已提交
236 237 238
  if (!pass) {
    return TSDB_CODE_PAR_PERMISSION_DENIED;
  }
239 240 241

  CHECK_CODE(getTableSchema(pCxt, name, isStb, &pCxt->pTableMeta));
  if (!isStb) {
X
Xiaoyu Wang 已提交
242
    SVgroupInfo vg;
243
    CHECK_CODE(getTableVgroup(pCxt, name, &vg));
X
Xiaoyu Wang 已提交
244
    CHECK_CODE(taosHashPut(pCxt->pVgroupsHashObj, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg)));
D
stmt  
dapan1121 已提交
245
  }
H
refact  
Hongze Cheng 已提交
246
  return TSDB_CODE_SUCCESS;
D
stmt  
dapan1121 已提交
247 248
}

X
Xiaoyu Wang 已提交
249 250 251
static int32_t getTableMeta(SInsertParseContext* pCxt, SName* name, char* dbFname) {
  return getTableMetaImpl(pCxt, name, dbFname, false);
}
D
stmt  
dapan1121 已提交
252

X
Xiaoyu Wang 已提交
253 254 255
static int32_t getSTableMeta(SInsertParseContext* pCxt, SName* name, char* dbFname) {
  return getTableMetaImpl(pCxt, name, dbFname, true);
}
D
stmt  
dapan1121 已提交
256

X
Xiaoyu Wang 已提交
257 258 259 260 261
static int32_t getDBCfg(SInsertParseContext* pCxt, const char* pDbFName, SDbCfgInfo* pInfo) {
  SParseContext* pBasicCtx = pCxt->pComCxt;
  if (pBasicCtx->async) {
    CHECK_CODE(getDbCfgFromCache(pCxt->pMetaCache, pDbFName, pInfo));
  } else {
X
Xiaoyu Wang 已提交
262
    SRequestConnInfo conn = {.pTrans = pBasicCtx->pTransporter,
D
dapan1121 已提交
263 264 265 266
                             .requestId = pBasicCtx->requestId,
                             .requestObjRefId = pBasicCtx->requestRid,
                             .mgmtEps = pBasicCtx->mgmtEpSet};
    CHECK_CODE(catalogGetDBCfg(pBasicCtx->pCatalog, &conn, pDbFName, pInfo));
X
Xiaoyu Wang 已提交
267 268 269 270
  }
  return TSDB_CODE_SUCCESS;
}

271 272 273 274 275 276 277 278 279 280
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 已提交
281
static void buildMsgHeader(STableDataBlocks* src, SVgDataBlocks* blocks) {
H
refact  
Hongze Cheng 已提交
282 283 284 285 286 287 288 289 290
  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;
291
    int32_t schemaLen = blk->schemaLen;
H
refact  
Hongze Cheng 已提交
292 293 294 295 296 297 298
    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);
299
    blk = (SSubmitBlk*)(blk->data + schemaLen + dataLen);
H
refact  
Hongze Cheng 已提交
300
  }
301 302 303 304 305 306 307 308 309 310
}

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 已提交
311
    SVgDataBlocks*    dst = taosMemoryCalloc(1, sizeof(SVgDataBlocks));
312 313 314
    if (NULL == dst) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
H
Haojun Liao 已提交
315
    taosHashGetDup(pCxt->pVgroupsHashObj, (const char*)&src->vgId, sizeof(src->vgId), &dst->vg);
316 317
    dst->numOfTables = src->numOfTables;
    dst->size = src->size;
wafwerar's avatar
wafwerar 已提交
318
    TSWAP(dst->pData, src->pData);
D
dapan1121 已提交
319
    buildMsgHeader(src, dst);
320 321 322 323 324
    taosArrayPush(pCxt->pOutput->pDataBlocks, &dst);
  }
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
325
int32_t checkTimestamp(STableDataBlocks* pDataBlocks, const char* start) {
326 327 328 329 330
  // 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 已提交
331
  TSKEY k = *(TSKEY*)start;
332
  if (k <= pDataBlocks->prevTS) {
333 334 335 336 337 338 339
    pDataBlocks->ordered = false;
  }

  pDataBlocks->prevTS = k;
  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
340 341 342 343 344 345
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;
346 347

  if (pToken->type == TK_NOW) {
348
    ts = taosGetTimestamp(timePrec);
349 350
  } else if (pToken->type == TK_TODAY) {
    ts = taosGetTimestampToday(timePrec);
351
  } else if (pToken->type == TK_NK_INTEGER) {
X
Xiaoyu Wang 已提交
352
    toInteger(pToken->z, pToken->n, 10, &ts);
H
refact  
Hongze Cheng 已提交
353
  } else {  // parse the RFC-3339/ISO-8601 timestamp format string
S
os env  
Shengliang Guan 已提交
354
    if (taosParseTime(pToken->z, time, pToken->n, timePrec, tsDaylight) != TSDB_CODE_SUCCESS) {
355
      return buildSyntaxErrMsg(pMsgBuf, "invalid timestamp format", pToken->z);
356 357 358 359 360 361 362
    }

    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 已提交
363
    if (pToken->z[k] == '(' && pToken->z[k + 1] == ')') {  // for insert NOW()/TODAY()
364 365 366 367
      *end = pTokenEnd = &pToken->z[k + 2];
      k++;
      continue;
    }
368
    if (pToken->z[k] == ',') {
369 370
      *end = pTokenEnd;
      *time = ts;
371 372 373 374 375 376 377 378 379 380 381 382 383 384 385
      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 已提交
386
  if (sToken.type == TK_NK_MINUS || sToken.type == TK_NK_PLUS) {
387 388 389 390 391
    index = 0;
    valueToken = tStrGetToken(pTokenEnd, &index, false);
    pTokenEnd += index;

    if (valueToken.n < 2) {
392
      return buildSyntaxErrMsg(pMsgBuf, "value expected in timestamp", sToken.z);
393 394 395 396 397 398 399
    }

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

400
    if (sToken.type == TK_NK_PLUS) {
401
      ts += interval;
402
    } else {
403
      ts = ts - interval;
404 405
    }

406
    *end = pTokenEnd;
407 408
  }

409
  *time = ts;
410 411
  return TSDB_CODE_SUCCESS;
}
412

wmmhello's avatar
wmmhello 已提交
413
static FORCE_INLINE int32_t checkAndTrimValue(SToken* pToken, char* tmpTokenBuf, SMsgBuf* pMsgBuf) {
H
refact  
Hongze Cheng 已提交
414 415 416 417
  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) ||
418
      (pToken->n == 0) || (pToken->type == TK_NK_RP)) {
X
Xiaoyu Wang 已提交
419 420 421 422
    return buildSyntaxErrMsg(pMsgBuf, "invalid data or symbol", pToken->z);
  }

  // Remove quotation marks
X
Xiaoyu Wang 已提交
423
  if (TK_NK_STRING == pToken->type) {
X
Xiaoyu Wang 已提交
424 425 426 427
    if (pToken->n >= TSDB_MAX_BYTES_PER_ROW) {
      return buildSyntaxErrMsg(pMsgBuf, "too long string", pToken->z);
    }

428
    int32_t len = trimString(pToken->z, pToken->n, tmpTokenBuf, TSDB_MAX_BYTES_PER_ROW);
X
Xiaoyu Wang 已提交
429
    pToken->z = tmpTokenBuf;
430
    pToken->n = len;
X
Xiaoyu Wang 已提交
431 432 433 434 435
  }

  return TSDB_CODE_SUCCESS;
}

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

H
refact  
Hongze Cheng 已提交
441
static FORCE_INLINE int32_t toDouble(SToken* pToken, double* value, char** endPtr) {
X
Xiaoyu Wang 已提交
442
  errno = 0;
wafwerar's avatar
wafwerar 已提交
443
  *value = taosStr2Double(pToken->z, endPtr);
X
Xiaoyu Wang 已提交
444 445 446

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

  return pToken->type;
}

H
refact  
Hongze Cheng 已提交
453 454
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 已提交
455 456 457
  int64_t  iv;
  uint64_t uv;
  char*    endptr = NULL;
X
Xiaoyu Wang 已提交
458

wmmhello's avatar
wmmhello 已提交
459
  int32_t code = checkAndTrimValue(pToken, tmpTokenBuf, pMsgBuf);
X
Xiaoyu Wang 已提交
460 461 462 463 464 465
  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 已提交
466
      return buildSyntaxErrMsg(pMsgBuf, "primary timestamp should not be null", pToken->z);
X
Xiaoyu Wang 已提交
467 468
    }

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

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

    case TSDB_DATA_TYPE_TINYINT: {
X
Xiaoyu Wang 已提交
494
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
X
Xiaoyu Wang 已提交
495 496 497 498 499 500
        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 已提交
501
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
502 503
    }

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

    case TSDB_DATA_TYPE_SMALLINT: {
X
Xiaoyu Wang 已提交
515
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
X
Xiaoyu Wang 已提交
516 517 518 519 520
        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 已提交
521
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
522 523 524
    }

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

    case TSDB_DATA_TYPE_INT: {
X
Xiaoyu Wang 已提交
535
      if (TSDB_CODE_SUCCESS != toInteger(pToken->z, pToken->n, 10, &iv)) {
X
Xiaoyu Wang 已提交
536 537 538 539 540
        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 已提交
541
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
542 543 544
    }

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

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

    case TSDB_DATA_TYPE_UBIGINT: {
X
Xiaoyu Wang 已提交
564
      if (TSDB_CODE_SUCCESS != toUInteger(pToken->z, pToken->n, 10, &uv)) {
X
Xiaoyu Wang 已提交
565
        return buildSyntaxErrMsg(pMsgBuf, "invalid unsigned bigint data", pToken->z);
X
Xiaoyu Wang 已提交
566
      } else if (!IS_VALID_UBIGINT(uv)) {
X
Xiaoyu Wang 已提交
567 568
        return buildSyntaxErrMsg(pMsgBuf, "unsigned bigint data overflow", pToken->z);
      }
X
Xiaoyu Wang 已提交
569
      return func(pMsgBuf, &uv, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
570 571 572 573
    }

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

    case TSDB_DATA_TYPE_DOUBLE: {
      double dv;
587
      if (TK_NK_ILLEGAL == toDouble(pToken, &dv, &endptr)) {
X
Xiaoyu Wang 已提交
588 589 590 591 592
        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 已提交
593
      return func(pMsgBuf, &dv, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
594 595 596 597 598
    }

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

X
Xiaoyu Wang 已提交
602
      return func(pMsgBuf, pToken->z, pToken->n, param);
X
Xiaoyu Wang 已提交
603 604 605
    }

    case TSDB_DATA_TYPE_NCHAR: {
X
Xiaoyu Wang 已提交
606
      return func(pMsgBuf, pToken->z, pToken->n, param);
X
Xiaoyu Wang 已提交
607
    }
608
    case TSDB_DATA_TYPE_JSON: {
X
Xiaoyu Wang 已提交
609
      if (pToken->n > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE) {
610 611 612 613
        return buildSyntaxErrMsg(pMsgBuf, "json string too long than 4095", pToken->z);
      }
      return func(pMsgBuf, pToken->z, pToken->n, param);
    }
X
Xiaoyu Wang 已提交
614 615 616 617 618 619
    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 已提交
620
      return func(pMsgBuf, &tmpVal, pSchema->bytes, param);
X
Xiaoyu Wang 已提交
621 622 623 624 625 626
    }
  }

  return TSDB_CODE_FAILED;
}

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

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

636
  if (TSDB_DATA_TYPE_BINARY == pa->schema->type) {
C
Cary Xu 已提交
637
    const char* rowEnd = tdRowEnd(rb->pBuf);
638
    STR_WITH_SIZE_TO_VARSTR(rowEnd, value, len);
639
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, false, pa->toffset, pa->colIdx);
640 641
  } 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 已提交
642 643
    int32_t     output = 0;
    const char* rowEnd = tdRowEnd(rb->pBuf);
wafwerar's avatar
wafwerar 已提交
644
    if (!taosMbsToUcs4(value, len, (TdUcs4*)varDataVal(rowEnd), pa->schema->bytes - VARSTR_HEADER_SIZE, &output)) {
D
dapan1121 已提交
645 646 647
      if (errno == E2BIG) {
        return generateSyntaxErrMsg(pMsgBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pa->schema->name);
      }
X
Xiaoyu Wang 已提交
648 649 650
      char buf[512] = {0};
      snprintf(buf, tListLen(buf), "%s", strerror(errno));
      return buildSyntaxErrMsg(pMsgBuf, buf, value);
651
    }
652
    varDataSetLen(rowEnd, output);
C
Cary Xu 已提交
653
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, rowEnd, false, pa->toffset, pa->colIdx);
654
  } else {
655
    tdAppendColValToRow(rb, pa->schema->colId, pa->schema->type, TD_VTYPE_NORM, value, false, pa->toffset, pa->colIdx);
656
  }
657

658 659
  return TSDB_CODE_SUCCESS;
}
660 661 662

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

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

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

678
    if (TK_NK_RP == sToken.type) {
679 680 681
      break;
    }

C
Cary Xu 已提交
682 683
    col_id_t t = lastColIdx + 1;
    col_id_t index = findCol(&sToken, t, nCols, pSchema);
684 685 686 687 688
    if (index < 0 && t > 0) {
      index = findCol(&sToken, 0, t, pSchema);
      isOrdered = false;
    }
    if (index < 0) {
689
      return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_INVALID_COLUMN, sToken.z);
690 691 692 693 694 695
    }
    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 已提交
696
    pColList->boundColumns[pColList->numOfBound] = index;
697
    ++pColList->numOfBound;
C
Cary Xu 已提交
698 699 700 701 702 703 704 705 706 707 708
    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;
    }
709 710 711 712 713
  }

  pColList->orderStatus = isOrdered ? ORDER_STATUS_ORDERED : ORDER_STATUS_DISORDERED;

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

X
Xiaoyu Wang 已提交
730
  if (pColList->numOfCols > pColList->numOfBound) {
731 732 733
    memset(&pColList->boundColumns[pColList->numOfBound], 0,
           sizeof(col_id_t) * (pColList->numOfCols - pColList->numOfBound));
  }
734 735 736 737

  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
738
static void buildCreateTbReq(SVCreateTbReq* pTbReq, const char* tname, STag* pTag, int64_t suid, const char* sname, SArray* tagName) {
wmmhello's avatar
wmmhello 已提交
739 740 741
  pTbReq->type = TD_CHILD_TABLE;
  pTbReq->name = strdup(tname);
  pTbReq->ctb.suid = suid;
wmmhello's avatar
wmmhello 已提交
742
  if(sname) pTbReq->ctb.name = strdup(sname);
wmmhello's avatar
wmmhello 已提交
743
  pTbReq->ctb.pTag = (uint8_t*)pTag;
wmmhello's avatar
wmmhello 已提交
744
  pTbReq->ctb.tagName = taosArrayDup(tagName);
wmmhello's avatar
wmmhello 已提交
745
  pTbReq->commentLen = -1;
746

wmmhello's avatar
wmmhello 已提交
747 748
  return;
}
749

750 751
static int32_t parseTagToken(char** end, SToken* pToken, SSchema* pSchema, int16_t timePrec, STagVal* val,
                             SMsgBuf* pMsgBuf) {
wmmhello's avatar
wmmhello 已提交
752 753 754 755 756 757 758 759
  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);
    }
760 761 762 763

    return TSDB_CODE_SUCCESS;
  }

wmmhello's avatar
wmmhello 已提交
764
//  strcpy(val->colName, pSchema->name);
wmmhello's avatar
wmmhello 已提交
765
  val->cid = pSchema->colId;
wmmhello's avatar
wmmhello 已提交
766
  val->type = pSchema->type;
C
Cary Xu 已提交
767

wmmhello's avatar
wmmhello 已提交
768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783
  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 已提交
784
      }
wmmhello's avatar
wmmhello 已提交
785 786
      break;
    }
787

wmmhello's avatar
wmmhello 已提交
788 789 790 791 792
    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 已提交
793
      }
wmmhello's avatar
wmmhello 已提交
794 795 796

      *(int8_t*)(&val->i64) = iv;
      break;
797 798
    }

wmmhello's avatar
wmmhello 已提交
799 800 801 802 803 804 805 806 807
    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;
    }
808

wmmhello's avatar
wmmhello 已提交
809 810 811 812 813 814 815 816 817
    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;
    }
818

wmmhello's avatar
wmmhello 已提交
819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 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
    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;
908
      void*   p = taosMemoryCalloc(1, pSchema->bytes - VARSTR_HEADER_SIZE);
909
      if (p == NULL) {
wmmhello's avatar
wmmhello 已提交
910 911
        return TSDB_CODE_OUT_OF_MEMORY;
      }
912
      if (!taosMbsToUcs4(pToken->z, pToken->n, (TdUcs4*)(p), pSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
wmmhello's avatar
wmmhello 已提交
913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934
        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 已提交
935 936 937 938

  return TSDB_CODE_SUCCESS;
}

939
// pSql -> tag1_value, ...)
wmmhello's avatar
wmmhello 已提交
940
static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const char* tName) {
wmmhello's avatar
wmmhello 已提交
941
  int32_t code = TSDB_CODE_SUCCESS;
942
  SArray* pTagVals = taosArrayInit(pCxt->tags.numOfBound, sizeof(STagVal));
wmmhello's avatar
wmmhello 已提交
943
  SArray* tagName = taosArrayInit(8, TSDB_COL_NAME_LEN);
944 945 946 947
  SToken  sToken;
  bool    isParseBindParam = false;
  bool    isJson = false;
  STag*   pTag = NULL;
948
  for (int i = 0; i < pCxt->tags.numOfBound; ++i) {
X
Xiaoyu Wang 已提交
949
    NEXT_TOKEN_WITH_PREV(pCxt->pSql, sToken);
D
stmt  
dapan1121 已提交
950 951 952 953

    if (sToken.type == TK_NK_QUESTION) {
      isParseBindParam = true;
      if (NULL == pCxt->pStmtCb) {
wmmhello's avatar
wmmhello 已提交
954 955
        code = buildSyntaxErrMsg(&pCxt->msg, "? only used in stmt", sToken.z);
        goto end;
D
stmt  
dapan1121 已提交
956 957 958 959 960 961
      }

      continue;
    }

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

X
Xiaoyu Wang 已提交
966
    SSchema* pTagSchema = &pSchema[pCxt->tags.boundColumns[i]];
X
Xiaoyu Wang 已提交
967
    char*    tmpTokenBuf = taosMemoryCalloc(1, sToken.n);  // todo this can be optimize with parse column
wmmhello's avatar
wmmhello 已提交
968 969 970 971 972
    code = checkAndTrimValue(&sToken, tmpTokenBuf, &pCxt->msg);
    if (code != TSDB_CODE_SUCCESS) {
      taosMemoryFree(tmpTokenBuf);
      goto end;
    }
wmmhello's avatar
wmmhello 已提交
973 974 975 976

    if (!isNullStr(&sToken)) {
      taosArrayPush(tagName, pTagSchema->name);
    }
977
    if (pTagSchema->type == TSDB_DATA_TYPE_JSON) {
wmmhello's avatar
wmmhello 已提交
978 979 980 981 982
      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;
      }
X
Xiaoyu Wang 已提交
983
      if (isNullStr(&sToken)) {
984 985
        code = tTagNew(pTagVals, 1, true, &pTag);
      } else {
wmmhello's avatar
wmmhello 已提交
986
        code = parseJsontoTagData(sToken.z, pTagVals, &pTag, &pCxt->msg);
987
      }
wmmhello's avatar
wmmhello 已提交
988
      taosMemoryFree(tmpTokenBuf);
989
      if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
990 991
        goto end;
      }
wmmhello's avatar
wmmhello 已提交
992
      isJson = true;
993
    } else {
wmmhello's avatar
wmmhello 已提交
994
      STagVal val = {0};
wmmhello's avatar
wmmhello 已提交
995
      code = parseTagToken(&pCxt->pSql, &sToken, pTagSchema, precision, &val, &pCxt->msg);
wmmhello's avatar
wmmhello 已提交
996 997 998 999
      if (TSDB_CODE_SUCCESS != code) {
        taosMemoryFree(tmpTokenBuf);
        goto end;
      }
1000
      if (pTagSchema->type != TSDB_DATA_TYPE_BINARY) {
wmmhello's avatar
wmmhello 已提交
1001 1002 1003 1004
        taosMemoryFree(tmpTokenBuf);
      }
      taosArrayPush(pTagVals, &val);
    }
1005 1006
  }

D
stmt  
dapan1121 已提交
1007
  if (isParseBindParam) {
wmmhello's avatar
wmmhello 已提交
1008 1009
    code = TSDB_CODE_SUCCESS;
    goto end;
D
stmt  
dapan1121 已提交
1010 1011
  }

1012
  if (!isJson && (code = tTagNew(pTagVals, 1, false, &pTag)) != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1013
    goto end;
1014 1015
  }

wmmhello's avatar
wmmhello 已提交
1016
  buildCreateTbReq(&pCxt->createTblReq, tName, pTag, pCxt->pTableMeta->suid, pCxt->sTableName, tagName);
wmmhello's avatar
wmmhello 已提交
1017 1018 1019

end:
  for (int i = 0; i < taosArrayGetSize(pTagVals); ++i) {
1020
    STagVal* p = (STagVal*)taosArrayGet(pTagVals, i);
wmmhello's avatar
wmmhello 已提交
1021
    if (p->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
1022 1023 1024 1025
      taosMemoryFree(p->pData);
    }
  }
  taosArrayDestroy(pTagVals);
wmmhello's avatar
wmmhello 已提交
1026
  taosArrayDestroy(tagName);
wmmhello's avatar
wmmhello 已提交
1027
  return code;
X
Xiaoyu Wang 已提交
1028
}
1029

X
Xiaoyu Wang 已提交
1030 1031
static int32_t storeTableMeta(SInsertParseContext* pCxt, SHashObj* pHash, SName* pTableName, const char* pName,
                              int32_t len, STableMeta* pMeta) {
1032 1033
  SVgroupInfo vg;
  CHECK_CODE(getTableVgroup(pCxt, pTableName, &vg));
X
Xiaoyu Wang 已提交
1034 1035
  CHECK_CODE(taosHashPut(pCxt->pVgroupsHashObj, (const char*)&vg.vgId, sizeof(vg.vgId), (char*)&vg, sizeof(vg)));

D
dapan 已提交
1036
  pMeta->uid = 0;
X
Xiaoyu Wang 已提交
1037
  pMeta->vgId = vg.vgId;
D
dapan 已提交
1038
  pMeta->tableType = TSDB_CHILD_TABLE;
X
Xiaoyu Wang 已提交
1039

X
Xiaoyu Wang 已提交
1040 1041 1042 1043 1044
  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);
1045 1046
}

1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072
static int32_t skipParentheses(SInsertParseSyntaxCxt* pCxt) {
  SToken  sToken;
  int32_t expectRightParenthesis = 1;
  while (1) {
    NEXT_TOKEN(pCxt->pSql, sToken);
    if (TK_NK_LP == sToken.type) {
      ++expectRightParenthesis;
    } else if (TK_NK_RP == sToken.type && 0 == --expectRightParenthesis) {
      break;
    }
    if (0 == sToken.n) {
      return buildSyntaxErrMsg(&pCxt->msg, ") expected", NULL);
    }
  }
  return TSDB_CODE_SUCCESS;
}

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

static int32_t ignoreBoundColumns(SInsertParseContext* pCxt) {
  SInsertParseSyntaxCxt cxt = {.pComCxt = pCxt->pComCxt, .pSql = pCxt->pSql, .msg = pCxt->msg, .pMetaCache = NULL};
  int32_t               code = skipBoundColumns(&cxt);
  pCxt->pSql = cxt.pSql;
  return code;
}

1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084
static int32_t skipUsingClause(SInsertParseSyntaxCxt* pCxt);

// pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
static int32_t ignoreAutoCreateTableClause(SInsertParseContext* pCxt) {
  SToken sToken;
  NEXT_TOKEN(pCxt->pSql, sToken);
  SInsertParseSyntaxCxt cxt = {.pComCxt = pCxt->pComCxt, .pSql = pCxt->pSql, .msg = pCxt->msg, .pMetaCache = NULL};
  int32_t               code = skipUsingClause(&cxt);
  pCxt->pSql = cxt.pSql;
  return code;
}

1085
// pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
D
dapan 已提交
1086
static int32_t parseUsingClause(SInsertParseContext* pCxt, SName* name, char* tbFName) {
H
refact  
Hongze Cheng 已提交
1087
  int32_t      len = strlen(tbFName);
X
Xiaoyu Wang 已提交
1088 1089
  STableMeta** pMeta = taosHashGet(pCxt->pSubTableHashObj, tbFName, len);
  if (NULL != pMeta) {
1090
    CHECK_CODE(ignoreAutoCreateTableClause(pCxt));
X
Xiaoyu Wang 已提交
1091 1092
    return cloneTableMeta(*pMeta, &pCxt->pTableMeta);
  }
1093

X
Xiaoyu Wang 已提交
1094
  SToken sToken;
1095 1096
  // pSql -> stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)
  NEXT_TOKEN(pCxt->pSql, sToken);
D
dapan 已提交
1097 1098 1099

  SName sname;
  createSName(&sname, &sToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg);
X
Xiaoyu Wang 已提交
1100 1101
  char dbFName[TSDB_DB_FNAME_LEN];
  tNameGetFullDbName(&sname, dbFName);
wmmhello's avatar
wmmhello 已提交
1102
  strcpy(pCxt->sTableName, sname.tname);
X
Xiaoyu Wang 已提交
1103

X
Xiaoyu Wang 已提交
1104
  CHECK_CODE(getSTableMeta(pCxt, &sname, dbFName));
1105 1106 1107
  if (TSDB_SUPER_TABLE != pCxt->pTableMeta->tableType) {
    return buildInvalidOperationMsg(&pCxt->msg, "create table only from super table is allowed");
  }
D
dapan 已提交
1108
  CHECK_CODE(storeTableMeta(pCxt, pCxt->pSubTableHashObj, name, tbFName, len, pCxt->pTableMeta));
1109 1110

  SSchema* pTagsSchema = getTableTagSchema(pCxt->pTableMeta);
1111
  setBoundColumnInfo(&pCxt->tags, pTagsSchema, getNumOfTags(pCxt->pTableMeta));
1112 1113 1114

  // pSql -> [(tag1_name, ...)] TAGS (tag1_value, ...)
  NEXT_TOKEN(pCxt->pSql, sToken);
1115
  if (TK_NK_LP == sToken.type) {
1116
    CHECK_CODE(parseBoundColumns(pCxt, &pCxt->tags, pTagsSchema));
1117 1118 1119 1120 1121 1122 1123 1124
    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);
1125
  if (TK_NK_LP != sToken.type) {
1126 1127
    return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
  }
X
Xiaoyu Wang 已提交
1128
  CHECK_CODE(parseTagsClause(pCxt, pTagsSchema, getTableInfo(pCxt->pTableMeta).precision, name->tname));
1129 1130 1131 1132
  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 已提交
1133 1134
    return buildSyntaxErrMsg(&pCxt->msg, ") is expected", sToken.z);
  }
1135 1136 1137 1138

  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
1139 1140
static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks, int16_t timePrec, bool* gotRow,
                       char* tmpTokenBuf) {
1141
  SParsedDataColInfo* spd = &pDataBlocks->boundColumnInfo;
C
Cary Xu 已提交
1142 1143 1144 1145
  SRowBuilder*        pBuilder = &pDataBlocks->rowBuilder;
  STSRow*             row = (STSRow*)(pDataBlocks->pData + pDataBlocks->size);  // skip the SSubmitBlk header

  tdSRowResetBuf(pBuilder, row);
1146

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

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

      continue;
    }

D
dapan1121 已提交
1165 1166 1167 1168
    if (TK_NK_RP == sToken.type) {
      return generateSyntaxErrMsg(&pCxt->msg, TSDB_CODE_PAR_INVALID_COLUMNS_NUM);
    }

D
stmt  
dapan1121 已提交
1169 1170 1171
    if (isParseBindParam) {
      return buildInvalidOperationMsg(&pCxt->msg, "no mix usage for ? and values");
    }
X
Xiaoyu Wang 已提交
1172

1173
    param.schema = pSchema;
D
stmt  
dapan1121 已提交
1174
    getSTSRowAppendInfo(pBuilder->rowType, spd, i, &param.toffset, &param.colIdx);
1175
    CHECK_CODE(parseValueToken(&pCxt->pSql, &sToken, pSchema, timePrec, tmpTokenBuf, MemRowAppend, &param, &pCxt->msg));
1176 1177

    if (PRIMARYKEY_TIMESTAMP_COL_ID == pSchema->colId) {
C
Cary Xu 已提交
1178
      TSKEY tsKey = TD_ROW_KEY(row);
1179
      checkTimestamp(pDataBlocks, (const char*)&tsKey);
1180
    }
X
Xiaoyu Wang 已提交
1181 1182 1183 1184 1185 1186 1187

    if (i < spd->numOfBound - 1) {
      NEXT_VALID_TOKEN(pCxt->pSql, sToken);
      if (TK_NK_COMMA != sToken.type) {
        return buildSyntaxErrMsg(&pCxt->msg, ", expected", sToken.z);
      }
    }
1188 1189 1190
  }

  if (!isParseBindParam) {
C
Cary Xu 已提交
1191
    // set the null value for the columns that do not assign values
C
Cary Xu 已提交
1192
    if ((spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
1193
      for (int32_t i = 0; i < spd->numOfCols; ++i) {
C
Cary Xu 已提交
1194 1195 1196
        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);
1197 1198 1199
        }
      }
    }
D
stmt  
dapan1121 已提交
1200 1201

    *gotRow = true;
C
Cary Xu 已提交
1202 1203
#ifdef TD_DEBUG_PRINT_ROW
    STSchema* pSTSchema = tdGetSTSChemaFromSSChema(&schema, spd->numOfCols);
C
Cary Xu 已提交
1204
    tdSRowPrint(row, pSTSchema, __func__);
C
Cary Xu 已提交
1205 1206
    taosMemoryFree(pSTSchema);
#endif
1207 1208
  }

C
Cary Xu 已提交
1209
  // *len = pBuilder->extendedRowSize;
1210 1211 1212 1213
  return TSDB_CODE_SUCCESS;
}

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

  (*numOfRows) = 0;
H
refact  
Hongze Cheng 已提交
1220
  char   tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0};  // used for deleting Escape character: \\, \', \"
1221 1222
  SToken sToken;
  while (1) {
1223 1224
    int32_t index = 0;
    NEXT_TOKEN_KEEP_SQL(pCxt->pSql, sToken, index);
1225
    if (TK_NK_LP != sToken.type) {
1226 1227
      break;
    }
1228
    pCxt->pSql += index;
1229 1230 1231 1232 1233 1234 1235 1236

    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 已提交
1237 1238 1239
    bool gotRow = false;
    CHECK_CODE(parseOneRow(pCxt, pDataBlock, tinfo.precision, &gotRow, tmpTokenBuf));
    if (gotRow) {
X
Xiaoyu Wang 已提交
1240
      pDataBlock->size += extendedRowSize;  // len;
D
stmt  
dapan1121 已提交
1241
    }
1242

1243 1244 1245 1246
    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) {
1247 1248 1249
      return buildSyntaxErrMsg(&pCxt->msg, ") expected", sToken.z);
    }

D
stmt  
dapan1121 已提交
1250 1251 1252
    if (gotRow) {
      (*numOfRows)++;
    }
1253 1254
  }

D
stmt  
dapan1121 已提交
1255
  if (0 == (*numOfRows) && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
X
Xiaoyu Wang 已提交
1256
    return buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL);
1257 1258 1259 1260
  }
  return TSDB_CODE_SUCCESS;
}

H
refact  
Hongze Cheng 已提交
1261
static int32_t parseValuesClause(SInsertParseContext* pCxt, STableDataBlocks* dataBuf) {
1262 1263 1264 1265
  int32_t maxNumOfRows;
  CHECK_CODE(allocateMemIfNeed(dataBuf, getExtendedRowSize(dataBuf), &maxNumOfRows));

  int32_t numOfRows = 0;
1266
  CHECK_CODE(parseValues(pCxt, dataBuf, maxNumOfRows, &numOfRows));
1267

H
refact  
Hongze Cheng 已提交
1268
  SSubmitBlk* pBlocks = (SSubmitBlk*)(dataBuf->pData);
D
dapan1121 已提交
1269
  if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, dataBuf, numOfRows)) {
1270 1271 1272 1273 1274 1275 1276 1277
    return buildInvalidOperationMsg(&pCxt->msg, "too many rows in sql, total number of rows should be less than 32767");
  }

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

X
Xiaoyu Wang 已提交
1278 1279 1280 1281 1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325 1326 1327 1328 1329 1330 1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341 1342 1343 1344 1345
static int32_t parseCsvFile(SInsertParseContext* pCxt, TdFilePtr fp, STableDataBlocks* pDataBlock, int maxRows,
                            int32_t* numOfRows) {
  STableComInfo tinfo = getTableInfo(pDataBlock->pTableMeta);
  int32_t       extendedRowSize = getExtendedRowSize(pDataBlock);
  CHECK_CODE(initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo));

  (*numOfRows) = 0;
  char    tmpTokenBuf[TSDB_MAX_BYTES_PER_ROW] = {0};  // used for deleting Escape character: \\, \', \"
  char*   pLine = NULL;
  int64_t readLen = 0;
  while ((readLen = taosGetLineFile(fp, &pLine)) != -1) {
    if (('\r' == pLine[readLen - 1]) || ('\n' == pLine[readLen - 1])) {
      pLine[--readLen] = '\0';
    }

    if (readLen == 0) {
      continue;
    }

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

    strtolower(pLine, pLine);
    char* pRawSql = pCxt->pSql;
    pCxt->pSql = pLine;
    bool gotRow = false;
    CHECK_CODE(parseOneRow(pCxt, pDataBlock, tinfo.precision, &gotRow, tmpTokenBuf));
    if (gotRow) {
      pDataBlock->size += extendedRowSize;  // len;
      (*numOfRows)++;
    }
    pCxt->pSql = pRawSql;
  }

  if (0 == (*numOfRows) && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
    return buildSyntaxErrMsg(&pCxt->msg, "no any data points", NULL);
  }
  return TSDB_CODE_SUCCESS;
}

static int32_t parseDataFromFile(SInsertParseContext* pCxt, SToken filePath, STableDataBlocks* dataBuf) {
  char filePathStr[TSDB_FILENAME_LEN] = {0};
  strncpy(filePathStr, filePath.z, filePath.n);
  TdFilePtr fp = taosOpenFile(filePathStr, TD_FILE_READ | TD_FILE_STREAM);
  if (NULL == fp) {
    return TAOS_SYSTEM_ERROR(errno);
  }

  int32_t maxNumOfRows;
  CHECK_CODE(allocateMemIfNeed(dataBuf, getExtendedRowSize(dataBuf), &maxNumOfRows));

  int32_t numOfRows = 0;
  CHECK_CODE(parseCsvFile(pCxt, fp, dataBuf, maxNumOfRows, &numOfRows));

  SSubmitBlk* pBlocks = (SSubmitBlk*)(dataBuf->pData);
  if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, dataBuf, numOfRows)) {
    return buildInvalidOperationMsg(&pCxt->msg, "too many rows in sql, total number of rows should be less than 32767");
  }

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

X
Xiaoyu Wang 已提交
1346
static void destroyInsertParseContextForTable(SInsertParseContext* pCxt) {
wafwerar's avatar
wafwerar 已提交
1347
  taosMemoryFreeClear(pCxt->pTableMeta);
X
Xiaoyu Wang 已提交
1348
  destroyBoundColumnInfo(&pCxt->tags);
wmmhello's avatar
wmmhello 已提交
1349
  tdDestroySVCreateTbReq(&pCxt->createTblReq);
X
Xiaoyu Wang 已提交
1350 1351
}

1352 1353
static void destroySubTableHashElem(void* p) { taosMemoryFree(*(STableMeta**)p); }

X
Xiaoyu Wang 已提交
1354 1355 1356
static void destroyInsertParseContext(SInsertParseContext* pCxt) {
  destroyInsertParseContextForTable(pCxt);
  taosHashCleanup(pCxt->pVgroupsHashObj);
1357
  taosHashCleanup(pCxt->pSubTableHashObj);
D
dapan1121 已提交
1358
  taosHashCleanup(pCxt->pTableNameHashObj);
D
dapan1121 已提交
1359
  taosHashCleanup(pCxt->pDbFNameHashObj);
1360 1361

  destroyBlockHashmap(pCxt->pTableBlockHashObj);
X
Xiaoyu Wang 已提交
1362 1363 1364
  destroyBlockArrayList(pCxt->pVgDataBlocks);
}

1365 1366 1367 1368 1369 1370
//   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) {
1371 1372 1373
  int32_t tbNum = 0;
  char    tbFName[TSDB_TABLE_FNAME_LEN];
  bool    autoCreateTbl = false;
X
Xiaoyu Wang 已提交
1374

X
Xiaoyu Wang 已提交
1375
  // for each table
1376 1377
  while (1) {
    SToken sToken;
X
Xiaoyu Wang 已提交
1378
    char*  tbName = NULL;
D
stmt  
dapan1121 已提交
1379

1380 1381 1382 1383 1384
    // pSql -> tb_name ...
    NEXT_TOKEN(pCxt->pSql, sToken);

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

D
stmt  
dapan1121 已提交
1389
      if (0 == pCxt->totalNum && (!TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT))) {
H
refact  
Hongze Cheng 已提交
1390
        return buildInvalidOperationMsg(&pCxt->msg, "no data in sql");
1391 1392 1393 1394
      }
      break;
    }

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

D
stmt  
dapan1121 已提交
1399 1400
    destroyInsertParseContextForTable(pCxt);

D
stmt  
dapan1121 已提交
1401 1402 1403
    if (TK_NK_QUESTION == sToken.type) {
      if (pCxt->pStmtCb) {
        CHECK_CODE((*pCxt->pStmtCb->getTbNameFn)(pCxt->pStmtCb->pStmt, &tbName));
X
Xiaoyu Wang 已提交
1404

D
stmt  
dapan1121 已提交
1405 1406 1407 1408 1409 1410
        sToken.z = tbName;
        sToken.n = strlen(tbName);
      } else {
        return buildSyntaxErrMsg(&pCxt->msg, "? only used in stmt", sToken.z);
      }
    }
X
Xiaoyu Wang 已提交
1411

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

D
dapan 已提交
1415
    SName name;
1416
    CHECK_CODE(createSName(&name, &tbnameToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg));
D
dapan 已提交
1417

1418
    tNameExtractFullName(&name, tbFName);
X
Xiaoyu Wang 已提交
1419
    CHECK_CODE(taosHashPut(pCxt->pTableNameHashObj, tbFName, strlen(tbFName), &name, sizeof(SName)));
D
dapan1121 已提交
1420 1421 1422
    char dbFName[TSDB_DB_FNAME_LEN];
    tNameGetFullDbName(&name, dbFName);
    CHECK_CODE(taosHashPut(pCxt->pDbFNameHashObj, dbFName, strlen(dbFName), dbFName, sizeof(dbFName)));
X
Xiaoyu Wang 已提交
1423

1424
    bool existedUsing = false;
1425
    // USING clause
1426
    if (TK_USING == sToken.type) {
1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441
      existedUsing = true;
      CHECK_CODE(parseUsingClause(pCxt, &name, tbFName));
      NEXT_TOKEN(pCxt->pSql, sToken);
      autoCreateTbl = true;
    }

    char* pBoundColsStart = NULL;
    if (TK_NK_LP == sToken.type) {
      // pSql -> field1_name, ...)
      pBoundColsStart = pCxt->pSql;
      CHECK_CODE(ignoreBoundColumns(pCxt));
      // CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pCxt->pTableMeta)));
      NEXT_TOKEN(pCxt->pSql, sToken);
    }

1442
    if (TK_USING == sToken.type) {
D
dapan 已提交
1443
      CHECK_CODE(parseUsingClause(pCxt, &name, tbFName));
1444
      NEXT_TOKEN(pCxt->pSql, sToken);
D
dapan 已提交
1445
      autoCreateTbl = true;
1446
    } else if (!existedUsing) {
D
dapan1121 已提交
1447
      CHECK_CODE(getTableMeta(pCxt, &name, dbFName));
1448 1449
    }

H
refact  
Hongze Cheng 已提交
1450
    STableDataBlocks* dataBuf = NULL;
D
dapan 已提交
1451
    CHECK_CODE(getDataBlockFromList(pCxt->pTableBlockHashObj, tbFName, strlen(tbFName), TSDB_DEFAULT_PAYLOAD_SIZE,
H
refact  
Hongze Cheng 已提交
1452 1453
                                    sizeof(SSubmitBlk), getTableInfo(pCxt->pTableMeta).rowSize, pCxt->pTableMeta,
                                    &dataBuf, NULL, &pCxt->createTblReq));
1454

1455 1456 1457
    if (NULL != pBoundColsStart) {
      char* pCurrPos = pCxt->pSql;
      pCxt->pSql = pBoundColsStart;
D
dapan1121 已提交
1458
      CHECK_CODE(parseBoundColumns(pCxt, &dataBuf->boundColumnInfo, getTableColumnSchema(pCxt->pTableMeta)));
1459
      pCxt->pSql = pCurrPos;
1460 1461 1462 1463 1464
    }

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

      tbNum++;
1468 1469 1470 1471
      continue;
    }

    // FILE csv_file_path
X
Xiaoyu Wang 已提交
1472
    if (TK_FILE == sToken.type) {
1473 1474
      // pSql -> csv_file_path
      NEXT_TOKEN(pCxt->pSql, sToken);
1475
      if (0 == sToken.n || (TK_NK_STRING != sToken.type && TK_NK_ID != sToken.type)) {
1476 1477
        return buildSyntaxErrMsg(&pCxt->msg, "file path is required following keyword FILE", sToken.z);
      }
X
Xiaoyu Wang 已提交
1478
      CHECK_CODE(parseDataFromFile(pCxt, sToken, dataBuf));
1479
      pCxt->pOutput->insertType = TSDB_QUERY_TYPE_FILE_INSERT;
D
stmt  
dapan1121 已提交
1480 1481

      tbNum++;
1482 1483 1484 1485 1486
      continue;
    }

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

D
stmt  
dapan1121 已提交
1488
  if (TSDB_QUERY_HAS_TYPE(pCxt->pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT)) {
X
Xiaoyu Wang 已提交
1489
    SParsedDataColInfo* tags = taosMemoryMalloc(sizeof(pCxt->tags));
D
stmt  
dapan1121 已提交
1490 1491 1492 1493
    if (NULL == tags) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    memcpy(tags, &pCxt->tags, sizeof(pCxt->tags));
1494
    (*pCxt->pStmtCb->setInfoFn)(pCxt->pStmtCb->pStmt, pCxt->pTableMeta, tags, tbFName, autoCreateTbl,
wmmhello's avatar
wmmhello 已提交
1495
                                pCxt->pVgroupsHashObj, pCxt->pTableBlockHashObj, pCxt->sTableName);
D
stmt  
dapan1121 已提交
1496

D
dapan 已提交
1497
    memset(&pCxt->tags, 0, sizeof(pCxt->tags));
D
stmt  
dapan1121 已提交
1498 1499
    pCxt->pVgroupsHashObj = NULL;
    pCxt->pTableBlockHashObj = NULL;
X
Xiaoyu Wang 已提交
1500

D
stmt  
dapan1121 已提交
1501 1502
    return TSDB_CODE_SUCCESS;
  }
X
Xiaoyu Wang 已提交
1503

1504
  // merge according to vgId
D
stmt  
dapan1121 已提交
1505
  if (taosHashGetSize(pCxt->pTableBlockHashObj) > 0) {
X
Xiaoyu Wang 已提交
1506
    CHECK_CODE(mergeTableDataBlocks(pCxt->pTableBlockHashObj, pCxt->pOutput->payloadType, &pCxt->pVgDataBlocks));
1507
  }
1508
  return buildOutput(pCxt);
1509 1510 1511 1512 1513 1514 1515 1516
}

// INSERT INTO
//   tb_name
//       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
//       [(field1_name, ...)]
//       VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path
//   [...];
1517
int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache) {
1518
  SInsertParseContext context = {
X
Xiaoyu Wang 已提交
1519 1520 1521 1522
      .pComCxt = pContext,
      .pSql = (char*)pContext->pSql,
      .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
      .pTableMeta = NULL,
wmmhello's avatar
wmmhello 已提交
1523
      .createTblReq = {0},
X
Xiaoyu Wang 已提交
1524 1525
      .pSubTableHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
      .pTableNameHashObj = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
D
dapan1121 已提交
1526
      .pDbFNameHashObj = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_VARCHAR), true, HASH_NO_LOCK),
X
Xiaoyu Wang 已提交
1527 1528
      .totalNum = 0,
      .pOutput = (SVnodeModifOpStmt*)nodesMakeNode(QUERY_NODE_VNODE_MODIF_STMT),
1529 1530
      .pStmtCb = pContext->pStmtCb,
      .pMetaCache = pMetaCache};
1531

D
stmt  
dapan1121 已提交
1532
  if (pContext->pStmtCb && *pQuery) {
X
Xiaoyu Wang 已提交
1533 1534
    (*pContext->pStmtCb->getExecInfoFn)(pContext->pStmtCb->pStmt, &context.pVgroupsHashObj,
                                        &context.pTableBlockHashObj);
D
stmt  
dapan1121 已提交
1535
  } else {
X
Xiaoyu Wang 已提交
1536 1537 1538
    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 已提交
1539
  }
X
Xiaoyu Wang 已提交
1540 1541

  if (NULL == context.pVgroupsHashObj || NULL == context.pTableBlockHashObj || NULL == context.pSubTableHashObj ||
D
dapan1121 已提交
1542
      NULL == context.pTableNameHashObj || NULL == context.pDbFNameHashObj || NULL == context.pOutput) {
X
Xiaoyu Wang 已提交
1543
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
1544
  }
1545
  taosHashSetFreeFp(context.pSubTableHashObj, destroySubTableHashElem);
1546

D
stmt  
dapan1121 已提交
1547 1548 1549 1550
  if (pContext->pStmtCb) {
    TSDB_QUERY_SET_TYPE(context.pOutput->insertType, TSDB_QUERY_TYPE_STMT_INSERT);
  }

1551
  if (NULL == *pQuery) {
1552
    *pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY);
D
stmt  
dapan1121 已提交
1553 1554 1555
    if (NULL == *pQuery) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
D
dapan1121 已提交
1556 1557
  } else {
    nodesDestroyNode((*pQuery)->pRoot);
1558
  }
D
dapan1121 已提交
1559
  
1560 1561 1562 1563
  (*pQuery)->execMode = QUERY_EXEC_MODE_SCHEDULE;
  (*pQuery)->haveResultSet = false;
  (*pQuery)->msgType = TDMT_VND_SUBMIT;
  (*pQuery)->pRoot = (SNode*)context.pOutput;
X
Xiaoyu Wang 已提交
1564

D
dapan1121 已提交
1565 1566 1567 1568 1569 1570
  if (NULL == (*pQuery)->pTableList) {
    (*pQuery)->pTableList = taosArrayInit(taosHashGetSize(context.pTableNameHashObj), sizeof(SName));
    if (NULL == (*pQuery)->pTableList) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
  }
1571

D
dapan1121 已提交
1572 1573 1574 1575 1576 1577 1578
  if (NULL == (*pQuery)->pDbList) {
    (*pQuery)->pDbList = taosArrayInit(taosHashGetSize(context.pDbFNameHashObj), TSDB_DB_FNAME_LEN);
    if (NULL == (*pQuery)->pDbList) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
  }

1579
  context.pOutput->payloadType = PAYLOAD_TYPE_KV;
1580

1581
  int32_t code = skipInsertInto(&context.pSql, &context.msg);
1582 1583 1584
  if (TSDB_CODE_SUCCESS == code) {
    code = parseInsertBody(&context);
  }
1585
  if (TSDB_CODE_SUCCESS == code || NEED_CLIENT_HANDLE_ERROR(code)) {
X
Xiaoyu Wang 已提交
1586 1587 1588 1589 1590
    SName* pTable = taosHashIterate(context.pTableNameHashObj, NULL);
    while (NULL != pTable) {
      taosArrayPush((*pQuery)->pTableList, pTable);
      pTable = taosHashIterate(context.pTableNameHashObj, pTable);
    }
D
dapan1121 已提交
1591 1592 1593 1594 1595 1596

    char* pDb = taosHashIterate(context.pDbFNameHashObj, NULL);
    while (NULL != pDb) {
      taosArrayPush((*pQuery)->pDbList, pDb);
      pDb = taosHashIterate(context.pDbFNameHashObj, pDb);
    }
X
Xiaoyu Wang 已提交
1597
  }
1598
  destroyInsertParseContext(&context);
X
Xiaoyu Wang 已提交
1599
  return code;
1600
}
D
stmt  
dapan1121 已提交
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
// 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));
X
Xiaoyu Wang 已提交
1650
  CHECK_CODE(reserveDbCfgInCache(pCxt->pComCxt->acctId, name.dbname, pCxt->pMetaCache));
1651 1652 1653 1654 1655 1656
  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;
}

X
Xiaoyu Wang 已提交
1657 1658 1659 1660 1661 1662 1663
static int32_t collectAutoCreateTableMetaKey(SInsertParseSyntaxCxt* pCxt, SToken* pTbToken) {
  SName name;
  CHECK_CODE(createSName(&name, pTbToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg));
  CHECK_CODE(reserveTableVgroupInCacheExt(&name, pCxt->pMetaCache));
  return TSDB_CODE_SUCCESS;
}

1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676 1677 1678 1679 1680 1681 1682 1683 1684 1685 1686 1687 1688 1689
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);

1690
    bool existedUsing = false;
1691 1692
    // USING clause
    if (TK_USING == sToken.type) {
1693
      existedUsing = true;
X
Xiaoyu Wang 已提交
1694
      CHECK_CODE(collectAutoCreateTableMetaKey(pCxt, &tbnameToken));
1695 1696 1697 1698 1699 1700 1701 1702 1703 1704 1705 1706
      NEXT_TOKEN(pCxt->pSql, sToken);
      CHECK_CODE(collectTableMetaKey(pCxt, &sToken));
      CHECK_CODE(skipUsingClause(pCxt));
      NEXT_TOKEN(pCxt->pSql, sToken);
    }

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

1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717
    if (TK_USING == sToken.type && !existedUsing) {
      existedUsing = true;
      CHECK_CODE(collectAutoCreateTableMetaKey(pCxt, &tbnameToken));
      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));
    }

1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739 1740 1741
    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;
}

1742
int32_t parseInsertSyntax(SParseContext* pContext, SQuery** pQuery, SParseMetaCache* pMetaCache) {
1743 1744 1745
  SInsertParseSyntaxCxt context = {.pComCxt = pContext,
                                   .pSql = (char*)pContext->pSql,
                                   .msg = {.buf = pContext->pMsg, .len = pContext->msgLen},
1746 1747
                                   .pMetaCache = pMetaCache};
  int32_t               code = skipInsertInto(&context.pSql, &context.msg);
1748 1749 1750 1751
  if (TSDB_CODE_SUCCESS == code) {
    code = parseInsertBodySyntax(&context);
  }
  if (TSDB_CODE_SUCCESS == code) {
1752
    *pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY);
1753 1754 1755 1756 1757 1758 1759
    if (NULL == *pQuery) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
  }
  return code;
}

X
Xiaoyu Wang 已提交
1760 1761 1762 1763
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 已提交
1764
  int32_t code = 0;
X
Xiaoyu Wang 已提交
1765 1766
  char*   tbName = NULL;

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

D
stmt  
dapan1121 已提交
1769 1770 1771 1772 1773 1774 1775 1776 1777 1778 1779
  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 已提交
1780
  if (sToken.n > 0) {
D
stmt  
dapan1121 已提交
1781 1782 1783 1784 1785 1786 1787
    return buildInvalidOperationMsg(&msg, "table name format is wrong");
  }

  return TSDB_CODE_SUCCESS;
}

int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash) {
X
Xiaoyu Wang 已提交
1788 1789
  SVnodeModifOpStmt*  modifyNode = (SVnodeModifOpStmt*)pQuery->pRoot;
  int32_t             code = 0;
D
stmt  
dapan1121 已提交
1790
  SInsertParseContext insertCtx = {
X
Xiaoyu Wang 已提交
1791 1792 1793
      .pVgroupsHashObj = pVgHash,
      .pTableBlockHashObj = pBlockHash,
      .pOutput = (SVnodeModifOpStmt*)pQuery->pRoot,
D
stmt  
dapan1121 已提交
1794
  };
X
Xiaoyu Wang 已提交
1795

D
stmt  
dapan1121 已提交
1796 1797
  // merge according to vgId
  if (taosHashGetSize(insertCtx.pTableBlockHashObj) > 0) {
D
stmt  
dapan1121 已提交
1798
    CHECK_CODE(mergeTableDataBlocks(insertCtx.pTableBlockHashObj, modifyNode->payloadType, &insertCtx.pVgDataBlocks));
D
stmt  
dapan1121 已提交
1799 1800 1801 1802
  }

  CHECK_CODE(buildOutput(&insertCtx));

wmmhello's avatar
wmmhello 已提交
1803
  destroyBlockArrayList(insertCtx.pVgDataBlocks);
D
stmt  
dapan1121 已提交
1804 1805 1806
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1807
int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const char* sTableName, char* tName, TAOS_MULTI_BIND* bind,
X
Xiaoyu Wang 已提交
1808 1809 1810
                           char* msgBuf, int32_t msgBufLen) {
  STableDataBlocks*   pDataBlock = (STableDataBlocks*)pBlock;
  SMsgBuf             pBuf = {.buf = msgBuf, .len = msgBufLen};
D
stmt  
dapan1121 已提交
1811 1812 1813 1814 1815
  SParsedDataColInfo* tags = (SParsedDataColInfo*)boundTags;
  if (NULL == tags) {
    return TSDB_CODE_QRY_APP_ERROR;
  }

C
Cary Xu 已提交
1816 1817
  SArray* pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal));
  if (!pTagArray) {
C
Cary Xu 已提交
1818
    return buildInvalidOperationMsg(&pBuf, "out of memory");
D
stmt  
dapan1121 已提交
1819 1820
  }

wmmhello's avatar
wmmhello 已提交
1821 1822 1823 1824 1825
  SArray* tagName = taosArrayInit(8, TSDB_COL_NAME_LEN);
  if (!tagName) {
    return buildInvalidOperationMsg(&pBuf, "out of memory");
  }

1826
  int32_t  code = TSDB_CODE_SUCCESS;
D
dapan1121 已提交
1827
  SSchema* pSchema = getTableTagSchema(pDataBlock->pTableMeta);
D
stmt  
dapan1121 已提交
1828

1829
  bool  isJson = false;
wmmhello's avatar
wmmhello 已提交
1830
  STag* pTag = NULL;
D
dapan1121 已提交
1831

D
stmt  
dapan1121 已提交
1832 1833 1834 1835
  for (int c = 0; c < tags->numOfBound; ++c) {
    if (bind[c].is_null && bind[c].is_null[0]) {
      continue;
    }
X
Xiaoyu Wang 已提交
1836

X
Xiaoyu Wang 已提交
1837
    SSchema* pTagSchema = &pSchema[tags->boundColumns[c]];
1838
    int32_t  colLen = pTagSchema->bytes;
D
stmt  
dapan1121 已提交
1839 1840 1841
    if (IS_VAR_DATA_TYPE(pTagSchema->type)) {
      colLen = bind[c].length[0];
    }
wmmhello's avatar
wmmhello 已提交
1842
    taosArrayPush(tagName, pTagSchema->name);
wmmhello's avatar
wmmhello 已提交
1843 1844 1845 1846 1847
    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 已提交
1848

wmmhello's avatar
wmmhello 已提交
1849
      isJson = true;
1850
      char* tmp = taosMemoryCalloc(1, colLen + 1);
wmmhello's avatar
wmmhello 已提交
1851
      memcpy(tmp, bind[c].buffer, colLen);
wmmhello's avatar
wmmhello 已提交
1852
      code = parseJsontoTagData(tmp, pTagArray, &pTag, &pBuf);
wmmhello's avatar
wmmhello 已提交
1853
      taosMemoryFree(tmp);
1854
      if (code != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1855 1856
        goto end;
      }
1857
    } else {
wmmhello's avatar
wmmhello 已提交
1858
      STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
wmmhello's avatar
wmmhello 已提交
1859
//      strcpy(val.colName, pTagSchema->name);
1860
      if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
wmmhello's avatar
wmmhello 已提交
1861 1862
        val.pData = (uint8_t*)bind[c].buffer;
        val.nData = colLen;
1863
      } else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
1864
        int32_t output = 0;
1865 1866
        void*   p = taosMemoryCalloc(1, colLen * TSDB_NCHAR_SIZE);
        if (p == NULL) {
wmmhello's avatar
wmmhello 已提交
1867 1868 1869
          code = TSDB_CODE_OUT_OF_MEMORY;
          goto end;
        }
wmmhello's avatar
wmmhello 已提交
1870
        if (!taosMbsToUcs4(bind[c].buffer, colLen, (TdUcs4*)(p), colLen * TSDB_NCHAR_SIZE, &output)) {
wmmhello's avatar
wmmhello 已提交
1871 1872
          if (errno == E2BIG) {
            taosMemoryFree(p);
wmmhello's avatar
wmmhello 已提交
1873
            code = generateSyntaxErrMsg(&pBuf, TSDB_CODE_PAR_VALUE_TOO_LONG, pTagSchema->name);
wmmhello's avatar
wmmhello 已提交
1874 1875 1876 1877
            goto end;
          }
          char buf[512] = {0};
          snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
wmmhello's avatar
wmmhello 已提交
1878
          taosMemoryFree(p);
wmmhello's avatar
wmmhello 已提交
1879
          code = buildSyntaxErrMsg(&pBuf, buf, bind[c].buffer);
wmmhello's avatar
wmmhello 已提交
1880 1881
          goto end;
        }
wmmhello's avatar
wmmhello 已提交
1882 1883
        val.pData = p;
        val.nData = output;
1884
      } else {
wmmhello's avatar
wmmhello 已提交
1885
        memcpy(&val.i64, bind[c].buffer, colLen);
wmmhello's avatar
wmmhello 已提交
1886
      }
wmmhello's avatar
wmmhello 已提交
1887
      taosArrayPush(pTagArray, &val);
wmmhello's avatar
wmmhello 已提交
1888
    }
D
stmt  
dapan1121 已提交
1889 1890
  }

wmmhello's avatar
wmmhello 已提交
1891
  if (!isJson && (code = tTagNew(pTagArray, 1, false, &pTag)) != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
1892
    goto end;
D
stmt  
dapan1121 已提交
1893 1894 1895
  }

  SVCreateTbReq tbReq = {0};
wmmhello's avatar
wmmhello 已提交
1896
  buildCreateTbReq(&tbReq, tName, pTag, suid, sTableName, tagName);
wmmhello's avatar
wmmhello 已提交
1897
  code = buildCreateTbMsg(pDataBlock, &tbReq);
wmmhello's avatar
wmmhello 已提交
1898
  tdDestroySVCreateTbReq(&tbReq);
D
stmt  
dapan1121 已提交
1899

wmmhello's avatar
wmmhello 已提交
1900 1901
end:
  for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
1902 1903
    STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
    if (p->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
1904 1905 1906
      taosMemoryFree(p->pData);
    }
  }
C
Cary Xu 已提交
1907
  taosArrayDestroy(pTagArray);
wmmhello's avatar
wmmhello 已提交
1908
  taosArrayDestroy(tagName);
D
stmt  
dapan1121 已提交
1909

wmmhello's avatar
wmmhello 已提交
1910
  return code;
D
stmt  
dapan1121 已提交
1911 1912
}

X
Xiaoyu Wang 已提交
1913 1914 1915 1916
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 已提交
1917 1918
  SParsedDataColInfo* spd = &pDataBlock->boundColumnInfo;
  SRowBuilder*        pBuilder = &pDataBlock->rowBuilder;
X
Xiaoyu Wang 已提交
1919 1920 1921 1922
  SMemParam           param = {.rb = pBuilder};
  SMsgBuf             pBuf = {.buf = msgBuf, .len = msgBufLen};
  int32_t             rowNum = bind->num;

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

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

D
stmt  
dapan1121 已提交
1927 1928 1929
  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 已提交
1930

D
stmt  
dapan1121 已提交
1931
    for (int c = 0; c < spd->numOfBound; ++c) {
X
Xiaoyu Wang 已提交
1932
      SSchema* pColSchema = &pSchema[spd->boundColumns[c]];
D
stmt  
dapan1121 已提交
1933 1934 1935 1936

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

D
stmt  
dapan1121 已提交
1938 1939 1940 1941
      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 已提交
1942 1943 1944
        if (pColSchema->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
          return buildInvalidOperationMsg(&pBuf, "primary timestamp should not be NULL");
        }
X
Xiaoyu Wang 已提交
1945

D
stmt  
dapan1121 已提交
1946 1947
        CHECK_CODE(MemRowAppend(&pBuf, NULL, 0, &param));
      } else {
D
dapan1121 已提交
1948 1949 1950 1951
        if (bind[c].buffer_type != pColSchema->type) {
          return buildInvalidOperationMsg(&pBuf, "column type mis-match with buffer type");
        }

D
stmt  
dapan1121 已提交
1952 1953 1954 1955
        int32_t colLen = pColSchema->bytes;
        if (IS_VAR_DATA_TYPE(pColSchema->type)) {
          colLen = bind[c].length[r];
        }
X
Xiaoyu Wang 已提交
1956 1957

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

D
stmt  
dapan1121 已提交
1960 1961
      if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
        TSKEY tsKey = TD_ROW_KEY(row);
X
Xiaoyu Wang 已提交
1962
        checkTimestamp(pDataBlock, (const char*)&tsKey);
D
stmt  
dapan1121 已提交
1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973
      }
    }
    // 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 已提交
1974 1975
#ifdef TD_DEBUG_PRINT_ROW
    STSchema* pSTSchema = tdGetSTSChemaFromSSChema(&pSchema, spd->numOfCols);
C
Cary Xu 已提交
1976
    tdSRowPrint(row, pSTSchema, __func__);
C
Cary Xu 已提交
1977 1978
    taosMemoryFree(pSTSchema);
#endif
D
stmt  
dapan1121 已提交
1979 1980
    pDataBlock->size += extendedRowSize;
  }
D
stmt  
dapan1121 已提交
1981

X
Xiaoyu Wang 已提交
1982
  SSubmitBlk* pBlocks = (SSubmitBlk*)(pDataBlock->pData);
D
stmt  
dapan1121 已提交
1983 1984 1985 1986 1987 1988 1989
  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 已提交
1990 1991 1992 1993 1994
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 已提交
1995 1996
  SParsedDataColInfo* spd = &pDataBlock->boundColumnInfo;
  SRowBuilder*        pBuilder = &pDataBlock->rowBuilder;
X
Xiaoyu Wang 已提交
1997 1998 1999 2000
  SMemParam           param = {.rb = pBuilder};
  SMsgBuf             pBuf = {.buf = msgBuf, .len = msgBufLen};
  bool                rowStart = (0 == colIdx);
  bool                rowEnd = ((colIdx + 1) == spd->numOfBound);
D
stmt  
dapan1121 已提交
2001 2002 2003 2004 2005

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

D
stmt  
dapan1121 已提交
2007 2008 2009 2010 2011 2012 2013
  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 已提交
2014

X
Xiaoyu Wang 已提交
2015
    SSchema* pColSchema = &pSchema[spd->boundColumns[colIdx]];
D
stmt  
dapan1121 已提交
2016 2017 2018 2019

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

D
stmt  
dapan1121 已提交
2021 2022 2023 2024 2025 2026 2027
    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 已提交
2028

D
stmt  
dapan1121 已提交
2029 2030
      CHECK_CODE(MemRowAppend(&pBuf, NULL, 0, &param));
    } else {
D
dapan1121 已提交
2031 2032 2033 2034
      if (bind->buffer_type != pColSchema->type) {
        return buildInvalidOperationMsg(&pBuf, "column type mis-match with buffer type");
      }

D
stmt  
dapan1121 已提交
2035 2036 2037 2038
      int32_t colLen = pColSchema->bytes;
      if (IS_VAR_DATA_TYPE(pColSchema->type)) {
        colLen = bind->length[r];
      }
X
Xiaoyu Wang 已提交
2039 2040

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

D
stmt  
dapan1121 已提交
2043 2044
    if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
      TSKEY tsKey = TD_ROW_KEY(row);
X
Xiaoyu Wang 已提交
2045
      checkTimestamp(pDataBlock, (const char*)&tsKey);
D
stmt  
dapan1121 已提交
2046
    }
X
Xiaoyu Wang 已提交
2047

D
stmt  
dapan1121 已提交
2048 2049 2050 2051 2052 2053 2054 2055
    // 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 已提交
2056
    }
C
Cary Xu 已提交
2057 2058

#ifdef TD_DEBUG_PRINT_ROW
X
Xiaoyu Wang 已提交
2059
    if (rowEnd) {
C
Cary Xu 已提交
2060
      STSchema* pSTSchema = tdGetSTSChemaFromSSChema(&pSchema, spd->numOfCols);
C
Cary Xu 已提交
2061
      tdSRowPrint(row, pSTSchema, __func__);
C
Cary Xu 已提交
2062 2063 2064
      taosMemoryFree(pSTSchema);
    }
#endif
D
stmt  
dapan1121 已提交
2065 2066
  }

D
stmt  
dapan1121 已提交
2067 2068 2069
  if (rowEnd) {
    pDataBlock->size += extendedRowSize * bind->num;

X
Xiaoyu Wang 已提交
2070
    SSubmitBlk* pBlocks = (SSubmitBlk*)(pDataBlock->pData);
D
stmt  
dapan1121 已提交
2071 2072 2073 2074 2075 2076 2077 2078
    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;
}

2079 2080
int32_t buildBoundFields(SParsedDataColInfo* boundInfo, SSchema* pSchema, int32_t* fieldNum, TAOS_FIELD_E** fields,
                         uint8_t timePrec) {
D
stmt  
dapan1121 已提交
2081 2082 2083 2084 2085 2086
  if (fields) {
    *fields = taosMemoryCalloc(boundInfo->numOfBound, sizeof(TAOS_FIELD));
    if (NULL == *fields) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

D
dapan1121 已提交
2087 2088
    SSchema* schema = &pSchema[boundInfo->boundColumns[0]];
    if (TSDB_DATA_TYPE_TIMESTAMP == schema->type) {
D
dapan1121 已提交
2089 2090
      (*fields)[0].precision = timePrec;
    }
2091

D
stmt  
dapan1121 已提交
2092
    for (int32_t i = 0; i < boundInfo->numOfBound; ++i) {
D
dapan1121 已提交
2093 2094 2095 2096
      schema = &pSchema[boundInfo->boundColumns[i]];
      strcpy((*fields)[i].name, schema->name);
      (*fields)[i].type = schema->type;
      (*fields)[i].bytes = schema->bytes;
D
stmt  
dapan1121 已提交
2097
    }
D
stmt  
dapan1121 已提交
2098 2099 2100 2101 2102 2103 2104
  }

  *fieldNum = boundInfo->numOfBound;

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
2105
int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD_E** fields) {
X
Xiaoyu Wang 已提交
2106
  STableDataBlocks*   pDataBlock = (STableDataBlocks*)pBlock;
D
stmt  
dapan1121 已提交
2107 2108 2109 2110
  SParsedDataColInfo* tags = (SParsedDataColInfo*)boundTags;
  if (NULL == tags) {
    return TSDB_CODE_QRY_APP_ERROR;
  }
X
Xiaoyu Wang 已提交
2111

D
dapan1121 已提交
2112 2113 2114 2115
  if (pDataBlock->pTableMeta->tableType != TSDB_SUPER_TABLE && pDataBlock->pTableMeta->tableType != TSDB_CHILD_TABLE) {
    return TSDB_CODE_TSC_STMT_API_ERROR;
  }

X
Xiaoyu Wang 已提交
2116
  SSchema* pSchema = getTableTagSchema(pDataBlock->pTableMeta);
D
stmt  
dapan1121 已提交
2117 2118 2119 2120 2121 2122 2123
  if (tags->numOfBound <= 0) {
    *fieldNum = 0;
    *fields = NULL;

    return TSDB_CODE_SUCCESS;
  }

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

D
stmt  
dapan1121 已提交
2126 2127 2128
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
2129
int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_E** fields) {
X
Xiaoyu Wang 已提交
2130 2131
  STableDataBlocks* pDataBlock = (STableDataBlocks*)pBlock;
  SSchema*          pSchema = getTableColumnSchema(pDataBlock->pTableMeta);
D
stmt  
dapan1121 已提交
2132 2133
  if (pDataBlock->boundColumnInfo.numOfBound <= 0) {
    *fieldNum = 0;
D
stmt  
dapan1121 已提交
2134 2135 2136
    if (fields) {
      *fields = NULL;
    }
D
stmt  
dapan1121 已提交
2137 2138 2139 2140

    return TSDB_CODE_SUCCESS;
  }

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

D
stmt  
dapan1121 已提交
2144 2145 2146
  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
2147
// schemaless logic start
D
stmt  
dapan1121 已提交
2148

wmmhello's avatar
wmmhello 已提交
2149
typedef struct SmlExecTableHandle {
X
Xiaoyu Wang 已提交
2150 2151
  SParsedDataColInfo tags;          // each table
  SVCreateTbReq      createTblReq;  // each table
wmmhello's avatar
wmmhello 已提交
2152
} SmlExecTableHandle;
wmmhello's avatar
wmmhello 已提交
2153

wmmhello's avatar
wmmhello 已提交
2154
typedef struct SmlExecHandle {
2155 2156 2157
  SHashObj*          pBlockHash;
  SmlExecTableHandle tableExecHandle;
  SQuery*            pQuery;
wmmhello's avatar
wmmhello 已提交
2158
} SSmlExecHandle;
wmmhello's avatar
wmmhello 已提交
2159

wmmhello's avatar
wmmhello 已提交
2160 2161 2162
static void smlDestroyTableHandle(void* pHandle) {
  SmlExecTableHandle* handle = (SmlExecTableHandle*)pHandle;
  destroyBoundColumnInfo(&handle->tags);
wmmhello's avatar
wmmhello 已提交
2163
  tdDestroySVCreateTbReq(&handle->createTblReq);
wmmhello's avatar
wmmhello 已提交
2164 2165
}

2166
static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SSchema* pSchema, bool isTag) {
wmmhello's avatar
wmmhello 已提交
2167 2168 2169 2170 2171 2172 2173 2174 2175 2176 2177 2178
  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 已提交
2179 2180
    SSmlKv*  kv = taosArrayGetP(cols, i);
    SToken   sToken = {.n = kv->keyLen, .z = (char*)kv->key};
wmmhello's avatar
wmmhello 已提交
2181
    col_id_t t = lastColIdx + 1;
2182 2183
    col_id_t index = ((t == 0 && !isTag) ? 0 : findCol(&sToken, t, nCols, pSchema));
    uDebug("SML, index:%d, t:%d, ncols:%d, kv->name:%s", index, t, nCols, kv->key);
wmmhello's avatar
wmmhello 已提交
2184 2185 2186 2187 2188
    if (index < 0 && t > 0) {
      index = findCol(&sToken, 0, t, pSchema);
      isOrdered = false;
    }
    if (index < 0) {
2189
      uError("smlBoundColumnData. index:%d", index);
wmmhello's avatar
wmmhello 已提交
2190 2191 2192
      return TSDB_CODE_SML_INVALID_DATA;
    }
    if (pColList->cols[index].valStat == VAL_STAT_HAS) {
2193
      uError("smlBoundColumnData. already set. index:%d", index);
wmmhello's avatar
wmmhello 已提交
2194 2195 2196 2197
      return TSDB_CODE_SML_INVALID_DATA;
    }
    lastColIdx = index;
    pColList->cols[index].valStat = VAL_STAT_HAS;
X
Xiaoyu Wang 已提交
2198
    pColList->boundColumns[pColList->numOfBound] = index;
wmmhello's avatar
wmmhello 已提交
2199 2200 2201 2202 2203 2204 2205 2206 2207 2208 2209 2210 2211 2212 2213 2214 2215 2216 2217 2218 2219 2220 2221 2222 2223 2224
    ++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;
    }
wafwerar's avatar
wafwerar 已提交
2225
    taosSort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar);
wmmhello's avatar
wmmhello 已提交
2226 2227 2228
    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
      pColIdx[i].finalIdx = i;
    }
wafwerar's avatar
wafwerar 已提交
2229
    taosSort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar);
wmmhello's avatar
wmmhello 已提交
2230 2231
  }

X
Xiaoyu Wang 已提交
2232
  if (pColList->numOfCols > pColList->numOfBound) {
wmmhello's avatar
wmmhello 已提交
2233 2234 2235 2236 2237 2238 2239
    memset(&pColList->boundColumns[pColList->numOfBound], 0,
           sizeof(col_id_t) * (pColList->numOfCols - pColList->numOfBound));
  }

  return TSDB_CODE_SUCCESS;
}

C
Cary Xu 已提交
2240 2241 2242 2243 2244 2245 2246 2247 2248 2249
/**
 * @brief No json tag for schemaless
 *
 * @param cols
 * @param tags
 * @param pSchema
 * @param ppTag
 * @param msg
 * @return int32_t
 */
wmmhello's avatar
wmmhello 已提交
2250
static int32_t smlBuildTagRow(SArray* cols, SParsedDataColInfo* tags, SSchema* pSchema, STag** ppTag, SArray** tagName, SMsgBuf* msg) {
C
Cary Xu 已提交
2251 2252
  SArray* pTagArray = taosArrayInit(tags->numOfBound, sizeof(STagVal));
  if (!pTagArray) {
wmmhello's avatar
wmmhello 已提交
2253 2254
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
wmmhello's avatar
wmmhello 已提交
2255 2256 2257 2258
  *tagName = taosArrayInit(8, TSDB_COL_NAME_LEN);
  if (!*tagName) {
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
  }
wmmhello's avatar
wmmhello 已提交
2259

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

wmmhello's avatar
wmmhello 已提交
2265
    taosArrayPush(*tagName, pTagSchema->name);
wmmhello's avatar
wmmhello 已提交
2266
    STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
wmmhello's avatar
wmmhello 已提交
2267
//    strcpy(val.colName, pTagSchema->name);
2268 2269
    if (pTagSchema->type == TSDB_DATA_TYPE_BINARY) {
      val.pData = (uint8_t*)kv->value;
wmmhello's avatar
wmmhello 已提交
2270
      val.nData = kv->length;
2271
    } else if (pTagSchema->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
2272
      int32_t output = 0;
X
Xiaoyu Wang 已提交
2273 2274
      void*   p = taosMemoryCalloc(1, kv->length * TSDB_NCHAR_SIZE);
      if (p == NULL) {
wmmhello's avatar
wmmhello 已提交
2275 2276 2277
        code = TSDB_CODE_OUT_OF_MEMORY;
        goto end;
      }
wmmhello's avatar
wmmhello 已提交
2278
      if (!taosMbsToUcs4(kv->value, kv->length, (TdUcs4*)(p), kv->length * TSDB_NCHAR_SIZE, &output)) {
wmmhello's avatar
wmmhello 已提交
2279 2280
        if (errno == E2BIG) {
          taosMemoryFree(p);
wmmhello's avatar
wmmhello 已提交
2281
          code = generateSyntaxErrMsg(msg, TSDB_CODE_PAR_VALUE_TOO_LONG, pTagSchema->name);
wmmhello's avatar
wmmhello 已提交
2282 2283 2284 2285 2286 2287 2288 2289 2290 2291
          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;
2292
    } else {
wmmhello's avatar
wmmhello 已提交
2293
      memcpy(&val.i64, &(kv->value), kv->length);
wmmhello's avatar
wmmhello 已提交
2294
    }
wmmhello's avatar
wmmhello 已提交
2295
    taosArrayPush(pTagArray, &val);
wmmhello's avatar
wmmhello 已提交
2296 2297
  }

wmmhello's avatar
wmmhello 已提交
2298 2299 2300
  code = tTagNew(pTagArray, 1, false, ppTag);
end:
  for (int i = 0; i < taosArrayGetSize(pTagArray); ++i) {
2301 2302
    STagVal* p = (STagVal*)taosArrayGet(pTagArray, i);
    if (p->type == TSDB_DATA_TYPE_NCHAR) {
wmmhello's avatar
wmmhello 已提交
2303 2304
      taosMemoryFree(p->pData);
    }
wmmhello's avatar
wmmhello 已提交
2305
  }
C
Cary Xu 已提交
2306
  taosArrayDestroy(pTagArray);
wmmhello's avatar
wmmhello 已提交
2307
  return code;
wmmhello's avatar
wmmhello 已提交
2308 2309
}

2310
int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols, bool format, STableMeta* pTableMeta,
wmmhello's avatar
wmmhello 已提交
2311
                    char* tableName, const char* sTableName, int32_t sTableNameLen, char* msgBuf, int16_t msgBufLen) {
wmmhello's avatar
wmmhello 已提交
2312 2313
  SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};

X
Xiaoyu Wang 已提交
2314
  SSmlExecHandle* smlHandle = (SSmlExecHandle*)handle;
2315 2316
  smlDestroyTableHandle(&smlHandle->tableExecHandle);  // free for each table
  SSchema* pTagsSchema = getTableTagSchema(pTableMeta);
wmmhello's avatar
wmmhello 已提交
2317
  setBoundColumnInfo(&smlHandle->tableExecHandle.tags, pTagsSchema, getNumOfTags(pTableMeta));
2318
  int ret = smlBoundColumnData(tags, &smlHandle->tableExecHandle.tags, pTagsSchema, true);
X
Xiaoyu Wang 已提交
2319
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2320 2321 2322
    buildInvalidOperationMsg(&pBuf, "bound tags error");
    return ret;
  }
C
Cary Xu 已提交
2323
  STag* pTag = NULL;
wmmhello's avatar
wmmhello 已提交
2324 2325
  SArray* tagName = NULL;
  ret = smlBuildTagRow(tags, &smlHandle->tableExecHandle.tags, pTagsSchema, &pTag, &tagName, &pBuf);
X
Xiaoyu Wang 已提交
2326
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2327
    taosArrayDestroy(tagName);
wmmhello's avatar
wmmhello 已提交
2328 2329
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2330

wmmhello's avatar
wmmhello 已提交
2331 2332 2333
  buildCreateTbReq(&smlHandle->tableExecHandle.createTblReq, tableName, pTag, pTableMeta->suid, NULL, tagName);
  taosArrayDestroy(tagName);

wmmhello's avatar
wmmhello 已提交
2334 2335 2336
  smlHandle->tableExecHandle.createTblReq.ctb.name = taosMemoryMalloc(sTableNameLen + 1);
  memcpy(smlHandle->tableExecHandle.createTblReq.ctb.name, sTableName, sTableNameLen);
  smlHandle->tableExecHandle.createTblReq.ctb.name[sTableNameLen] = 0;
wmmhello's avatar
wmmhello 已提交
2337

wmmhello's avatar
wmmhello 已提交
2338
  STableDataBlocks* pDataBlock = NULL;
X
Xiaoyu Wang 已提交
2339 2340
  ret = getDataBlockFromList(smlHandle->pBlockHash, &pTableMeta->uid, sizeof(pTableMeta->uid),
                             TSDB_DEFAULT_PAYLOAD_SIZE, sizeof(SSubmitBlk), getTableInfo(pTableMeta).rowSize,
wmmhello's avatar
wmmhello 已提交
2341
                             pTableMeta, &pDataBlock, NULL, &smlHandle->tableExecHandle.createTblReq);
X
Xiaoyu Wang 已提交
2342
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2343 2344 2345
    buildInvalidOperationMsg(&pBuf, "create data block error");
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2346 2347 2348

  SSchema* pSchema = getTableColumnSchema(pTableMeta);

2349
  ret = smlBoundColumnData(colsSchema, &pDataBlock->boundColumnInfo, pSchema, false);
X
Xiaoyu Wang 已提交
2350
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2351 2352 2353
    buildInvalidOperationMsg(&pBuf, "bound cols error");
    return ret;
  }
X
Xiaoyu Wang 已提交
2354
  int32_t             extendedRowSize = getExtendedRowSize(pDataBlock);
wmmhello's avatar
wmmhello 已提交
2355 2356
  SParsedDataColInfo* spd = &pDataBlock->boundColumnInfo;
  SRowBuilder*        pBuilder = &pDataBlock->rowBuilder;
X
Xiaoyu Wang 已提交
2357
  SMemParam           param = {.rb = pBuilder};
wmmhello's avatar
wmmhello 已提交
2358 2359 2360

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

2361
  int32_t rowNum = taosArrayGetSize(cols);
2362
  if (rowNum <= 0) {
wmmhello's avatar
wmmhello 已提交
2363 2364
    return buildInvalidOperationMsg(&pBuf, "cols size <= 0");
  }
wmmhello's avatar
wmmhello 已提交
2365
  ret = allocateMemForSize(pDataBlock, extendedRowSize * rowNum);
X
Xiaoyu Wang 已提交
2366
  if (ret != TSDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
2367 2368 2369
    buildInvalidOperationMsg(&pBuf, "allocate memory error");
    return ret;
  }
wmmhello's avatar
wmmhello 已提交
2370 2371 2372
  for (int32_t r = 0; r < rowNum; ++r) {
    STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size);  // skip the SSubmitBlk header
    tdSRowResetBuf(pBuilder, row);
2373
    void*  rowData = taosArrayGetP(cols, r);
2374
    size_t rowDataSize = 0;
2375
    if (format) {
2376
      rowDataSize = taosArrayGetSize(rowData);
wmmhello's avatar
wmmhello 已提交
2377
    }
wmmhello's avatar
wmmhello 已提交
2378 2379

    // 1. set the parsed value from sql string
2380
    for (int c = 0, j = 0; c < spd->numOfBound; ++c) {
2381
      SSchema* pColSchema = &pSchema[spd->boundColumns[c]];
wmmhello's avatar
wmmhello 已提交
2382 2383 2384 2385

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

X
Xiaoyu Wang 已提交
2386 2387 2388
      SSmlKv* kv = NULL;
      if (format) {
        if (j < rowDataSize) {
2389
          kv = taosArrayGetP(rowData, j);
2390
          if (rowDataSize != spd->numOfBound && j != 0 &&
X
Xiaoyu Wang 已提交
2391
              (kv->keyLen != strlen(pColSchema->name) || strncmp(kv->key, pColSchema->name, kv->keyLen) != 0)) {
2392
            kv = NULL;
X
Xiaoyu Wang 已提交
2393
          } else {
2394
            j++;
2395
          }
wmmhello's avatar
wmmhello 已提交
2396
        }
X
Xiaoyu Wang 已提交
2397 2398 2399
      } else {
        void** p = taosHashGet(rowData, pColSchema->name, strlen(pColSchema->name));
        if (p) kv = *p;
wmmhello's avatar
wmmhello 已提交
2400
      }
wmmhello's avatar
wmmhello 已提交
2401

2402
      if (!kv || kv->length == 0) {
wmmhello's avatar
wmmhello 已提交
2403 2404
        MemRowAppend(&pBuf, NULL, 0, &param);
      } else {
wmmhello's avatar
wmmhello 已提交
2405 2406
        int32_t colLen = kv->length;
        if (pColSchema->type == TSDB_DATA_TYPE_TIMESTAMP) {
2407
//          uError("SML:data before:%ld, precision:%d", kv->i, pTableMeta->tableInfo.precision);
2408
          kv->i = convertTimePrecision(kv->i, TSDB_TIME_PRECISION_NANO, pTableMeta->tableInfo.precision);
2409
//          uError("SML:data after:%ld, precision:%d", kv->i, pTableMeta->tableInfo.precision);
wmmhello's avatar
wmmhello 已提交
2410 2411
        }

2412
        if (IS_VAR_DATA_TYPE(kv->type)) {
wmmhello's avatar
wmmhello 已提交
2413
          MemRowAppend(&pBuf, kv->value, colLen, &param);
2414
        } else {
wmmhello's avatar
wmmhello 已提交
2415 2416
          MemRowAppend(&pBuf, &(kv->value), colLen, &param);
        }
wmmhello's avatar
wmmhello 已提交
2417 2418 2419 2420
      }

      if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
        TSKEY tsKey = TD_ROW_KEY(row);
X
Xiaoyu Wang 已提交
2421
        checkTimestamp(pDataBlock, (const char*)&tsKey);
wmmhello's avatar
wmmhello 已提交
2422 2423 2424 2425 2426 2427 2428 2429 2430 2431 2432 2433 2434 2435 2436 2437
      }
    }

    // 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 已提交
2438
  SSubmitBlk* pBlocks = (SSubmitBlk*)(pDataBlock->pData);
wmmhello's avatar
wmmhello 已提交
2439 2440 2441 2442 2443 2444 2445
  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 已提交
2446 2447 2448
void* smlInitHandle(SQuery* pQuery) {
  SSmlExecHandle* handle = taosMemoryCalloc(1, sizeof(SSmlExecHandle));
  if (!handle) return NULL;
wmmhello's avatar
wmmhello 已提交
2449 2450 2451 2452 2453 2454
  handle->pBlockHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
  handle->pQuery = pQuery;

  return handle;
}

X
Xiaoyu Wang 已提交
2455 2456 2457
void smlDestroyHandle(void* pHandle) {
  if (!pHandle) return;
  SSmlExecHandle* handle = (SSmlExecHandle*)pHandle;
wmmhello's avatar
wmmhello 已提交
2458
  destroyBlockHashmap(handle->pBlockHash);
wmmhello's avatar
wmmhello 已提交
2459
  smlDestroyTableHandle(&handle->tableExecHandle);
wmmhello's avatar
wmmhello 已提交
2460 2461 2462 2463
  taosMemoryFree(handle);
}

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