tscParseInsert.c 46.7 KB
Newer Older
H
hzcheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
/*
 * 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/>.
 */

#define _DEFAULT_SOURCE /* See feature_test_macros(7) */
#define _GNU_SOURCE

#define _XOPEN_SOURCE

21
#include "os.h"
22 23

#include "hash.h"
H
hzcheng 已提交
24 25 26
#include "tscUtil.h"
#include "tschemautil.h"
#include "tsclient.h"
27
#include "ttokendef.h"
H
hzcheng 已提交
28
#include "taosdef.h"
H
hzcheng 已提交
29

S
slguan 已提交
30
#include "tscLog.h"
H
hjxilinx 已提交
31
#include "tscSubquery.h"
H
hzcheng 已提交
32 33
#include "tstoken.h"

S
slguan 已提交
34
#include "tdataformat.h"
35

S
slguan 已提交
36
enum {
S
slguan 已提交
37 38 39 40
  TSDB_USE_SERVER_TS = 0,
  TSDB_USE_CLI_TS = 1,
};

L
lihui 已提交
41
static int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t * numOfRows);
H
hzcheng 已提交
42

H
Haojun Liao 已提交
43
static int32_t tscToInteger(SStrToken *pToken, int64_t *value, char **endPtr) {
44 45 46 47
  if (pToken->n == 0) {
    return TK_ILLEGAL;
  }
  
B
Bomin Zhang 已提交
48 49 50 51 52 53 54 55

  int32_t radix = 10;
  if (pToken->type == TK_HEX) {
    radix = 16;
  } else if (pToken->type == TK_BIN) {
    radix = 2;
  }
  
L
lihui 已提交
56
  errno = 0;
B
Bomin Zhang 已提交
57
  *value = strtoll(pToken->z, endPtr, radix);
B
Bomin Zhang 已提交
58 59 60 61 62 63
  if (**endPtr == 'e' || **endPtr == 'E' || **endPtr == '.') {
    errno = 0;
    double v = round(strtod(pToken->z, endPtr));
    if (v > INT64_MAX || v <= INT64_MIN) {
      errno = ERANGE;
    } else {
S
TD-1057  
Shengliang Guan 已提交
64
      *value = (int64_t)v;
B
Bomin Zhang 已提交
65 66
    }
  }
67 68
  
  // not a valid integer number, return error
B
Bomin Zhang 已提交
69
  if (*endPtr - pToken->z != pToken->n) {
70 71
    return TK_ILLEGAL;
  }
S
slguan 已提交
72

H
Haojun Liao 已提交
73
  return pToken->type;
H
hzcheng 已提交
74 75
}

H
Haojun Liao 已提交
76
static int32_t tscToDouble(SStrToken *pToken, double *value, char **endPtr) {
77 78 79 80
  if (pToken->n == 0) {
    return TK_ILLEGAL;
  }
  
L
lihui 已提交
81
  errno = 0;
S
slguan 已提交
82
  *value = strtod(pToken->z, endPtr);
83 84
  
  // not a valid integer number, return error
B
Bomin Zhang 已提交
85
  if ((*endPtr - pToken->z) != pToken->n) {
86 87
    return TK_ILLEGAL;
  }
B
Bomin Zhang 已提交
88 89

  return pToken->type;
S
slguan 已提交
90
}
H
hzcheng 已提交
91

H
Haojun Liao 已提交
92
int tsParseTime(SStrToken *pToken, int64_t *time, char **next, char *error, int16_t timePrec) {
S
slguan 已提交
93
  int32_t   index = 0;
H
Haojun Liao 已提交
94
  SStrToken sToken;
S
slguan 已提交
95 96 97
  int64_t   interval;
  int64_t   useconds = 0;
  char *    pTokenEnd = *next;
H
hzcheng 已提交
98

S
slguan 已提交
99
  index = 0;
H
hzcheng 已提交
100

S
slguan 已提交
101
  if (pToken->type == TK_NOW) {
H
hzcheng 已提交
102
    useconds = taosGetTimestamp(timePrec);
S
slguan 已提交
103
  } else if (strncmp(pToken->z, "0", 1) == 0 && pToken->n == 1) {
H
hzcheng 已提交
104
    // do nothing
S
slguan 已提交
105
  } else if (pToken->type == TK_INTEGER) {
S
Shengliang Guan 已提交
106
    useconds = tsosStr2int64(pToken->z);
H
hzcheng 已提交
107 108
  } else {
    // strptime("2001-11-12 18:31:01", "%Y-%m-%d %H:%M:%S", &tm);
dengyihao's avatar
dengyihao 已提交
109
    if (taosParseTime(pToken->z, time, pToken->n, timePrec, tsDaylight) != TSDB_CODE_SUCCESS) {
H
hjxilinx 已提交
110
      return tscInvalidSQLErrMsg(error, "invalid timestamp format", pToken->z);
H
hzcheng 已提交
111 112 113 114 115
    }

    return TSDB_CODE_SUCCESS;
  }

S
slguan 已提交
116 117 118 119
  for (int k = pToken->n; pToken->z[k] != '\0'; k++) {
    if (pToken->z[k] == ' ' || pToken->z[k] == '\t') continue;
    if (pToken->z[k] == ',') {
      *next = pTokenEnd;
H
hzcheng 已提交
120 121 122 123 124 125 126 127
      *time = useconds;
      return 0;
    }

    break;
  }

  /*
S
slguan 已提交
128 129 130
   * time expression:
   * e.g., now+12a, now-5h
   */
H
Haojun Liao 已提交
131
  SStrToken valueToken;
S
slguan 已提交
132 133 134
  index = 0;
  sToken = tStrGetToken(pTokenEnd, &index, false, 0, NULL);
  pTokenEnd += index;
135

S
slguan 已提交
136 137 138 139
  if (sToken.type == TK_MINUS || sToken.type == TK_PLUS) {
    index = 0;
    valueToken = tStrGetToken(pTokenEnd, &index, false, 0, NULL);
    pTokenEnd += index;
140

S
slguan 已提交
141
    if (valueToken.n < 2) {
H
hjxilinx 已提交
142
      return tscInvalidSQLErrMsg(error, "value expected in timestamp", sToken.z);
H
hzcheng 已提交
143 144
    }

145
    if (parseAbsoluteDuration(valueToken.z, valueToken.n, &interval) != TSDB_CODE_SUCCESS) {
146
      return TSDB_CODE_TSC_INVALID_SQL;
H
hzcheng 已提交
147
    }
148

H
hzcheng 已提交
149 150 151 152
    if (timePrec == TSDB_TIME_PRECISION_MILLI) {
      interval /= 1000;
    }

S
slguan 已提交
153
    if (sToken.type == TK_PLUS) {
H
hzcheng 已提交
154 155 156 157 158 159 160 161 162 163 164 165
      useconds += interval;
    } else {
      useconds = (useconds >= interval) ? useconds - interval : 0;
    }

    *next = pTokenEnd;
  }

  *time = useconds;
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
166
int32_t tsParseOneColumnData(SSchema *pSchema, SStrToken *pToken, char *payload, char *msg, char **str, bool primaryKey,
S
slguan 已提交
167 168 169
                             int16_t timePrec) {
  int64_t iv;
  int32_t numType;
S
slguan 已提交
170
  char *  endptr = NULL;
171 172
  errno = 0;  // clear the previous existed error information

H
hzcheng 已提交
173 174
  switch (pSchema->type) {
    case TSDB_DATA_TYPE_BOOL: {  // bool
S
slguan 已提交
175 176
      if ((pToken->type == TK_BOOL || pToken->type == TK_STRING) && (pToken->n != 0)) {
        if (strncmp(pToken->z, "true", pToken->n) == 0) {
S
slguan 已提交
177
          *(uint8_t *)payload = TSDB_TRUE;
S
slguan 已提交
178
        } else if (strncmp(pToken->z, "false", pToken->n) == 0) {
S
slguan 已提交
179
          *(uint8_t *)payload = TSDB_FALSE;
S
slguan 已提交
180 181
        } else if (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0) {
          *(uint8_t *)payload = TSDB_DATA_BOOL_NULL;
H
hzcheng 已提交
182
        } else {
Y
TD-934  
yihaoDeng 已提交
183
          return tscSQLSyntaxErrMsg(msg, "invalid bool data", pToken->z);
H
hzcheng 已提交
184
        }
S
slguan 已提交
185 186 187 188 189 190 191 192 193
      } else if (pToken->type == TK_INTEGER) {
        iv = strtoll(pToken->z, NULL, 10);
        *(uint8_t *)payload = (int8_t)((iv == 0) ? TSDB_FALSE : TSDB_TRUE);
      } else if (pToken->type == TK_FLOAT) {
        double dv = strtod(pToken->z, NULL);
        *(uint8_t *)payload = (int8_t)((dv == 0) ? TSDB_FALSE : TSDB_TRUE);
      } else if (pToken->type == TK_NULL) {
        *(uint8_t *)payload = TSDB_DATA_BOOL_NULL;
      } else {
H
hjxilinx 已提交
194
        return tscInvalidSQLErrMsg(msg, "invalid bool data", pToken->z);
H
hzcheng 已提交
195 196 197 198
      }
      break;
    }
    case TSDB_DATA_TYPE_TINYINT:
S
slguan 已提交
199 200 201 202 203
      if (pToken->type == TK_NULL) {
        *((int8_t *)payload) = TSDB_DATA_TINYINT_NULL;
      } else if ((pToken->type == TK_STRING) && (pToken->n != 0) &&
                 (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0)) {
        *((int8_t *)payload) = TSDB_DATA_TINYINT_NULL;
H
hzcheng 已提交
204
      } else {
S
slguan 已提交
205
        numType = tscToInteger(pToken, &iv, &endptr);
L
lihui 已提交
206
        if (TK_ILLEGAL == numType) {
H
hjxilinx 已提交
207
          return tscInvalidSQLErrMsg(msg, "invalid tinyint data", pToken->z);
L
lihui 已提交
208
        } else if (errno == ERANGE || iv > INT8_MAX || iv <= INT8_MIN) {
H
hjxilinx 已提交
209
          return tscInvalidSQLErrMsg(msg, "tinyint data overflow", pToken->z);
H
hzcheng 已提交
210 211
        }

212
        *((int8_t *)payload) = (int8_t)iv;
H
hzcheng 已提交
213 214 215 216 217
      }

      break;

    case TSDB_DATA_TYPE_SMALLINT:
S
slguan 已提交
218 219 220 221 222
      if (pToken->type == TK_NULL) {
        *((int16_t *)payload) = TSDB_DATA_SMALLINT_NULL;
      } else if ((pToken->type == TK_STRING) && (pToken->n != 0) &&
                 (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0)) {
        *((int16_t *)payload) = TSDB_DATA_SMALLINT_NULL;
H
hzcheng 已提交
223
      } else {
S
slguan 已提交
224
        numType = tscToInteger(pToken, &iv, &endptr);
L
lihui 已提交
225
        if (TK_ILLEGAL == numType) {
H
hjxilinx 已提交
226
          return tscInvalidSQLErrMsg(msg, "invalid smallint data", pToken->z);
L
lihui 已提交
227
        } else if (errno == ERANGE || iv > INT16_MAX || iv <= INT16_MIN) {
H
hjxilinx 已提交
228
          return tscInvalidSQLErrMsg(msg, "smallint data overflow", pToken->z);
H
hzcheng 已提交
229 230
        }

S
slguan 已提交
231
        *((int16_t *)payload) = (int16_t)iv;
H
hzcheng 已提交
232 233 234 235
      }
      break;

    case TSDB_DATA_TYPE_INT:
S
slguan 已提交
236 237 238 239
      if (pToken->type == TK_NULL) {
        *((int32_t *)payload) = TSDB_DATA_INT_NULL;
      } else if ((pToken->type == TK_STRING) && (pToken->n != 0) &&
                 (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0)) {
S
slguan 已提交
240
        *((int32_t *)payload) = TSDB_DATA_INT_NULL;
H
hzcheng 已提交
241
      } else {
S
slguan 已提交
242
        numType = tscToInteger(pToken, &iv, &endptr);
L
lihui 已提交
243
        if (TK_ILLEGAL == numType) {
H
hjxilinx 已提交
244
          return tscInvalidSQLErrMsg(msg, "invalid int data", pToken->z);
L
lihui 已提交
245
        } else if (errno == ERANGE || iv > INT32_MAX || iv <= INT32_MIN) {
H
hjxilinx 已提交
246
          return tscInvalidSQLErrMsg(msg, "int data overflow", pToken->z);
H
hzcheng 已提交
247 248
        }

S
slguan 已提交
249
        *((int32_t *)payload) = (int32_t)iv;
H
hzcheng 已提交
250 251 252 253 254
      }

      break;

    case TSDB_DATA_TYPE_BIGINT:
S
slguan 已提交
255 256 257 258
      if (pToken->type == TK_NULL) {
        *((int64_t *)payload) = TSDB_DATA_BIGINT_NULL;
      } else if ((pToken->type == TK_STRING) && (pToken->n != 0) &&
                 (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0)) {
S
slguan 已提交
259
        *((int64_t *)payload) = TSDB_DATA_BIGINT_NULL;
H
hzcheng 已提交
260
      } else {
S
slguan 已提交
261
        numType = tscToInteger(pToken, &iv, &endptr);
L
lihui 已提交
262
        if (TK_ILLEGAL == numType) {
H
hjxilinx 已提交
263
          return tscInvalidSQLErrMsg(msg, "invalid bigint data", pToken->z);
B
Bomin Zhang 已提交
264
        } else if (errno == ERANGE || iv == INT64_MIN) {
H
hjxilinx 已提交
265
          return tscInvalidSQLErrMsg(msg, "bigint data overflow", pToken->z);
H
hzcheng 已提交
266
        }
S
slguan 已提交
267 268

        *((int64_t *)payload) = iv;
H
hzcheng 已提交
269 270 271 272
      }
      break;

    case TSDB_DATA_TYPE_FLOAT:
S
slguan 已提交
273 274 275 276
      if (pToken->type == TK_NULL) {
        *((int32_t *)payload) = TSDB_DATA_FLOAT_NULL;
      } else if ((pToken->type == TK_STRING) && (pToken->n != 0) &&
                 (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0)) {
S
slguan 已提交
277
        *((int32_t *)payload) = TSDB_DATA_FLOAT_NULL;
H
hzcheng 已提交
278
      } else {
S
slguan 已提交
279 280
        double dv;
        if (TK_ILLEGAL == tscToDouble(pToken, &dv, &endptr)) {
H
hjxilinx 已提交
281
          return tscInvalidSQLErrMsg(msg, "illegal float data", pToken->z);
S
slguan 已提交
282 283 284 285
        }

        float fv = (float)dv;
        if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || (fv > FLT_MAX || fv < -FLT_MAX)) {
H
hjxilinx 已提交
286
          return tscInvalidSQLErrMsg(msg, "illegal float data", pToken->z);
H
hzcheng 已提交
287 288
        }

S
slguan 已提交
289 290
        if (isinf(fv) || isnan(fv)) {
          *((int32_t *)payload) = TSDB_DATA_FLOAT_NULL;
H
hzcheng 已提交
291
        }
S
slguan 已提交
292 293

        *((float *)payload) = fv;
H
hzcheng 已提交
294 295 296 297
      }
      break;

    case TSDB_DATA_TYPE_DOUBLE:
S
slguan 已提交
298 299 300 301
      if (pToken->type == TK_NULL) {
        *((int64_t *)payload) = TSDB_DATA_DOUBLE_NULL;
      } else if ((pToken->type == TK_STRING) && (pToken->n != 0) &&
                 (strncasecmp(TSDB_DATA_NULL_STR_L, pToken->z, pToken->n) == 0)) {
S
slguan 已提交
302
        *((int64_t *)payload) = TSDB_DATA_DOUBLE_NULL;
H
hzcheng 已提交
303
      } else {
S
slguan 已提交
304 305
        double dv;
        if (TK_ILLEGAL == tscToDouble(pToken, &dv, &endptr)) {
H
hjxilinx 已提交
306
          return tscInvalidSQLErrMsg(msg, "illegal double data", pToken->z);
H
hzcheng 已提交
307 308
        }

S
slguan 已提交
309
        if (((dv == HUGE_VAL || dv == -HUGE_VAL) && errno == ERANGE) || (dv > DBL_MAX || dv < -DBL_MAX)) {
H
hjxilinx 已提交
310
          return tscInvalidSQLErrMsg(msg, "illegal double data", pToken->z);
S
slguan 已提交
311 312 313 314 315 316
        }

        if (isinf(dv) || isnan(dv)) {
          *((int64_t *)payload) = TSDB_DATA_DOUBLE_NULL;
        } else {
          *((double *)payload) = dv;
H
hzcheng 已提交
317 318 319 320 321
        }
      }
      break;

    case TSDB_DATA_TYPE_BINARY:
S
slguan 已提交
322 323
      // binary data cannot be null-terminated char string, otherwise the last char of the string is lost
      if (pToken->type == TK_NULL) {
324
        setVardataNull(payload, TSDB_DATA_TYPE_BINARY);
H
hjxilinx 已提交
325
      } else { // too long values will return invalid sql, not be truncated automatically
H
hjxilinx 已提交
326
        if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) { //todo refactor
H
hjxilinx 已提交
327
          return tscInvalidSQLErrMsg(msg, "string data overflow", pToken->z);
S
slguan 已提交
328
        }
H
hjxilinx 已提交
329
        
330
        STR_WITH_SIZE_TO_VARSTR(payload, pToken->z, pToken->n);
H
hzcheng 已提交
331 332 333 334 335
      }

      break;

    case TSDB_DATA_TYPE_NCHAR:
S
slguan 已提交
336
      if (pToken->type == TK_NULL) {
337
        setVardataNull(payload, TSDB_DATA_TYPE_NCHAR);
H
hzcheng 已提交
338
      } else {
H
hjxilinx 已提交
339
        // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
340 341
        size_t output = 0;
        if (!taosMbsToUcs4(pToken->z, pToken->n, varDataVal(payload), pSchema->bytes - VARSTR_HEADER_SIZE, &output)) {
H
hjxilinx 已提交
342 343
          char buf[512] = {0};
          snprintf(buf, tListLen(buf), "%s", strerror(errno));
H
hjxilinx 已提交
344
          return tscInvalidSQLErrMsg(msg, buf, pToken->z);
H
hzcheng 已提交
345
        }
346
        
347
        varDataSetLen(payload, output);
H
hzcheng 已提交
348 349 350 351
      }
      break;

    case TSDB_DATA_TYPE_TIMESTAMP: {
S
slguan 已提交
352
      if (pToken->type == TK_NULL) {
H
hzcheng 已提交
353
        if (primaryKey) {
S
slguan 已提交
354
          *((int64_t *)payload) = 0;
H
hzcheng 已提交
355
        } else {
S
slguan 已提交
356
          *((int64_t *)payload) = TSDB_DATA_BIGINT_NULL;
H
hzcheng 已提交
357 358
        }
      } else {
S
slguan 已提交
359 360
        int64_t temp;
        if (tsParseTime(pToken, &temp, str, msg, timePrec) != TSDB_CODE_SUCCESS) {
H
hjxilinx 已提交
361
          return tscInvalidSQLErrMsg(msg, "invalid timestamp", pToken->z);
H
hzcheng 已提交
362
        }
H
hjxilinx 已提交
363
        
S
slguan 已提交
364
        *((int64_t *)payload) = temp;
H
hzcheng 已提交
365 366 367 368 369 370
      }

      break;
    }
  }

H
hjxilinx 已提交
371
  return TSDB_CODE_SUCCESS;
H
hzcheng 已提交
372 373
}

S
slguan 已提交
374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393
/*
 * The server time/client time should not be mixed up in one sql string
 * Do not employ sort operation is not involved if server time is used.
 */
static int32_t tsCheckTimestamp(STableDataBlocks *pDataBlocks, const char *start) {
  // once the data block is disordered, we do NOT keep previous timestamp any more
  if (!pDataBlocks->ordered) {
    return TSDB_CODE_SUCCESS;
  }

  TSKEY k = *(TSKEY *)start;

  if (k == 0) {
    if (pDataBlocks->tsSource == TSDB_USE_CLI_TS) {
      return -1;
    } else if (pDataBlocks->tsSource == -1) {
      pDataBlocks->tsSource = TSDB_USE_SERVER_TS;
    }
  } else {
    if (pDataBlocks->tsSource == TSDB_USE_SERVER_TS) {
H
hjxilinx 已提交
394
      return -1;  // client time/server time can not be mixed
395

S
slguan 已提交
396 397 398 399 400 401 402 403 404 405 406 407 408
    } else if (pDataBlocks->tsSource == -1) {
      pDataBlocks->tsSource = TSDB_USE_CLI_TS;
    }
  }

  if (k <= pDataBlocks->prevTS && (pDataBlocks->tsSource == TSDB_USE_CLI_TS)) {
    pDataBlocks->ordered = false;
  }

  pDataBlocks->prevTS = k;
  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
409
int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[], SParsedDataColInfo *spd, SSqlCmd* pCmd,
410 411
                      int16_t timePrec, int32_t *code, char *tmpTokenBuf) {
  int32_t index = 0;
H
Haojun Liao 已提交
412
  SStrToken sToken = {0};
S
slguan 已提交
413
  char *    payload = pDataBlocks->pData + pDataBlocks->size;
S
slguan 已提交
414

S
slguan 已提交
415
  // 1. set the parsed value from sql string
H
hzcheng 已提交
416
  int32_t rowSize = 0;
417
  for (int i = 0; i < spd->numOfAssignedCols; ++i) {
S
slguan 已提交
418
    // the start position in data block buffer of current value in sql
419 420
    char *   start = payload + spd->elems[i].offset;
    int16_t  colIndex = spd->elems[i].colIndex;
S
slguan 已提交
421
    SSchema *pSchema = schema + colIndex;
S
slguan 已提交
422
    rowSize += pSchema->bytes;
H
hzcheng 已提交
423

S
slguan 已提交
424 425 426 427 428
    index = 0;
    sToken = tStrGetToken(*str, &index, true, 0, NULL);
    *str += index;

    if (sToken.type == TK_QUESTION) {
H
Haojun Liao 已提交
429 430 431 432 433
      if (pCmd->insertType != TSDB_QUERY_TYPE_STMT_INSERT) {
        *code = tscSQLSyntaxErrMsg(pCmd->payload, "? only allowed in binding insertion", *str);
        return -1;
      }

S
TD-1057  
Shengliang Guan 已提交
434
      uint32_t offset = (uint32_t)(start - pDataBlocks->pData);
S
slguan 已提交
435 436 437
      if (tscAddParamToDataBlock(pDataBlocks, pSchema->type, (uint8_t)timePrec, pSchema->bytes, offset) != NULL) {
        continue;
      }
438

H
Haojun Liao 已提交
439
      strcpy(pCmd->payload, "client out of memory");
440
      *code = TSDB_CODE_TSC_OUT_OF_MEMORY;
H
hzcheng 已提交
441 442 443
      return -1;
    }

444 445 446
    int16_t type = sToken.type;
    if ((type != TK_NOW && type != TK_INTEGER && type != TK_STRING && type != TK_FLOAT && type != TK_BOOL &&
         type != TK_NULL && type != TK_HEX && type != TK_OCT && type != TK_BIN) || (sToken.n == 0) || (type == TK_RP)) {
H
Haojun Liao 已提交
447
      *code = tscSQLSyntaxErrMsg(pCmd->payload, "invalid data or symbol", sToken.z);
S
slguan 已提交
448
      return -1;
H
hzcheng 已提交
449 450
    }

S
slguan 已提交
451 452
    // Remove quotation marks
    if (TK_STRING == sToken.type) {
L
[1292]  
lihui 已提交
453
      // delete escape character: \\, \', \"
454
      char    delim = sToken.z[0];
L
[1292]  
lihui 已提交
455 456
      int32_t cnt = 0;
      int32_t j = 0;
S
TD-1057  
Shengliang Guan 已提交
457
      for (uint32_t k = 1; k < sToken.n - 1; ++k) {
F
fang 已提交
458 459
        if (sToken.z[k] == delim || sToken.z[k] == '\\') {
          if (sToken.z[k + 1] == delim) {
L
[1292]  
lihui 已提交
460
            cnt++;
L
lihui 已提交
461 462 463
            tmpTokenBuf[j] = sToken.z[k + 1];
            j++;
            k++;
L
[1292]  
lihui 已提交
464 465 466
            continue;
          }
        }
467

L
[NONE]  
lihui 已提交
468
        tmpTokenBuf[j] = sToken.z[k];
L
[1292]  
lihui 已提交
469 470
        j++;
      }
471
      tmpTokenBuf[j] = 0;
L
[1292]  
lihui 已提交
472
      sToken.z = tmpTokenBuf;
473
      sToken.n -= 2 + cnt;
H
hzcheng 已提交
474 475
    }

S
slguan 已提交
476
    bool    isPrimaryKey = (colIndex == PRIMARYKEY_TIMESTAMP_COL_INDEX);
H
Haojun Liao 已提交
477
    int32_t ret = tsParseOneColumnData(pSchema, &sToken, start, pCmd->payload, str, isPrimaryKey, timePrec);
S
slguan 已提交
478
    if (ret != TSDB_CODE_SUCCESS) {
Y
TD-934  
yihaoDeng 已提交
479
      *code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
H
hzcheng 已提交
480 481
      return -1;  // NOTE: here 0 mean error!
    }
482

S
slguan 已提交
483
    if (isPrimaryKey && tsCheckTimestamp(pDataBlocks, start) != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
484
      tscInvalidSQLErrMsg(pCmd->payload, "client time/server time can not be mixed up", sToken.z);
485
      *code = TSDB_CODE_TSC_INVALID_TIME_STAMP;
S
slguan 已提交
486
      return -1;
487
    }
H
hzcheng 已提交
488 489
  }

S
slguan 已提交
490
  // 2. set the null value for the columns that do not assign values
491
  if (spd->numOfAssignedCols < spd->numOfCols) {
S
slguan 已提交
492
    char *ptr = payload;
H
hzcheng 已提交
493 494

    for (int32_t i = 0; i < spd->numOfCols; ++i) {
495
      
496
      if (!spd->hasVal[i]) {  // current column do not have any value to insert, set it to null
497 498 499 500 501 502 503 504 505
        if (schema[i].type == TSDB_DATA_TYPE_BINARY) {
          varDataSetLen(ptr, sizeof(int8_t));
          *(uint8_t*) varDataVal(ptr) = TSDB_DATA_BINARY_NULL;
        } else if (schema[i].type == TSDB_DATA_TYPE_NCHAR) {
          varDataSetLen(ptr, sizeof(int32_t));
          *(uint32_t*) varDataVal(ptr) = TSDB_DATA_NCHAR_NULL;
        } else {
          setNull(ptr, schema[i].type, schema[i].bytes);
        }
H
hzcheng 已提交
506
      }
507
      
H
hzcheng 已提交
508 509 510
      ptr += schema[i].bytes;
    }

S
TD-1057  
Shengliang Guan 已提交
511
    rowSize = (int32_t)(ptr - payload);
H
hzcheng 已提交
512 513 514 515 516
  }

  return rowSize;
}

S
slguan 已提交
517 518 519 520 521 522 523 524 525
static int32_t rowDataCompar(const void *lhs, const void *rhs) {
  TSKEY left = *(TSKEY *)lhs;
  TSKEY right = *(TSKEY *)rhs;

  if (left == right) {
    return 0;
  } else {
    return left > right ? 1 : -1;
  }
526 527
}

H
hjxilinx 已提交
528
int tsParseValues(char **str, STableDataBlocks *pDataBlock, STableMeta *pTableMeta, int maxRows,
H
Haojun Liao 已提交
529
                  SParsedDataColInfo *spd, SSqlCmd* pCmd, int32_t *code, char *tmpTokenBuf) {
S
slguan 已提交
530
  int32_t   index = 0;
H
Haojun Liao 已提交
531
  SStrToken sToken;
H
hzcheng 已提交
532

S
TD-1225  
Shengliang Guan 已提交
533
  int32_t numOfRows = 0;
H
hzcheng 已提交
534

H
hjxilinx 已提交
535
  SSchema *pSchema = tscGetTableSchema(pTableMeta);
H
hjxilinx 已提交
536
  STableComInfo tinfo = tscGetTableInfo(pTableMeta);
H
hjxilinx 已提交
537 538
  
  int32_t  precision = tinfo.precision;
S
slguan 已提交
539 540

  if (spd->hasVal[0] == false) {
H
Haojun Liao 已提交
541
    *code = tscInvalidSQLErrMsg(pCmd->payload, "primary timestamp column can not be null", *str);
H
hzcheng 已提交
542 543 544 545
    return -1;
  }

  while (1) {
S
slguan 已提交
546 547 548
    index = 0;
    sToken = tStrGetToken(*str, &index, false, 0, NULL);
    if (sToken.n == 0 || sToken.type != TK_LP) break;
H
hzcheng 已提交
549

S
slguan 已提交
550
    *str += index;
H
hjxilinx 已提交
551
    if (numOfRows >= maxRows || pDataBlock->size + tinfo.rowSize >= pDataBlock->nAllocSize) {
L
lihui 已提交
552
      int32_t tSize;
H
Haojun Liao 已提交
553 554 555
      *code = tscAllocateMemIfNeed(pDataBlock, tinfo.rowSize, &tSize);
      if (*code != TSDB_CODE_SUCCESS) {  //TODO pass the correct error code to client
        strcpy(pCmd->payload, "client out of memory");
S
slguan 已提交
556 557
        return -1;
      }
H
Haojun Liao 已提交
558

L
lihui 已提交
559 560
      ASSERT(tSize > maxRows);
      maxRows = tSize;
H
hzcheng 已提交
561 562
    }

H
Haojun Liao 已提交
563
    int32_t len = tsParseOneRowData(str, pDataBlock, pSchema, spd, pCmd, precision, code, tmpTokenBuf);
564
    if (len <= 0) {  // error message has been set in tsParseOneRowData
H
hzcheng 已提交
565 566 567 568 569
      return -1;
    }

    pDataBlock->size += len;

S
slguan 已提交
570 571 572 573
    index = 0;
    sToken = tStrGetToken(*str, &index, false, 0, NULL);
    *str += index;
    if (sToken.n == 0 || sToken.type != TK_RP) {
H
Haojun Liao 已提交
574
      tscSQLSyntaxErrMsg(pCmd->payload, ") expected", *str);
Y
TD-934  
yihaoDeng 已提交
575
      *code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
H
hzcheng 已提交
576 577 578 579 580 581 582
      return -1;
    }

    numOfRows++;
  }

  if (numOfRows <= 0) {
H
Haojun Liao 已提交
583
    strcpy(pCmd->payload, "no any data points");
Y
TD-934  
yihaoDeng 已提交
584
    *code = TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
S
slguan 已提交
585 586 587
    return -1;
  } else {
    return numOfRows;
H
hzcheng 已提交
588 589 590
  }
}

S
slguan 已提交
591
static void tscSetAssignedColumnInfo(SParsedDataColInfo *spd, SSchema *pSchema, int32_t numOfCols) {
H
hzcheng 已提交
592
  spd->numOfCols = numOfCols;
593
  spd->numOfAssignedCols = numOfCols;
H
hzcheng 已提交
594 595 596 597 598 599 600 601 602 603 604

  for (int32_t i = 0; i < numOfCols; ++i) {
    spd->hasVal[i] = true;
    spd->elems[i].colIndex = i;

    if (i > 0) {
      spd->elems[i].offset = spd->elems[i - 1].offset + pSchema[i - 1].bytes;
    }
  }
}

L
lihui 已提交
605
int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t * numOfRows) {
S
slguan 已提交
606
  size_t    remain = pDataBlock->nAllocSize - pDataBlock->size;
S
slguan 已提交
607
  const int factor = 5;
S
slguan 已提交
608
  uint32_t nAllocSizeOld = pDataBlock->nAllocSize;
L
[#1102]  
lihui 已提交
609
  
H
hzcheng 已提交
610
  // expand the allocated size
S
slguan 已提交
611 612
  if (remain < rowSize * factor) {
    while (remain < rowSize * factor) {
S
slguan 已提交
613
      pDataBlock->nAllocSize = (uint32_t)(pDataBlock->nAllocSize * 1.5);
S
slguan 已提交
614 615
      remain = pDataBlock->nAllocSize - pDataBlock->size;
    }
H
hzcheng 已提交
616

S
slguan 已提交
617 618 619 620 621
    char *tmp = realloc(pDataBlock->pData, (size_t)pDataBlock->nAllocSize);
    if (tmp != NULL) {
      pDataBlock->pData = tmp;
      memset(pDataBlock->pData + pDataBlock->size, 0, pDataBlock->nAllocSize - pDataBlock->size);
    } else {
H
hjxilinx 已提交
622
      // do nothing, if allocate more memory failed
S
slguan 已提交
623
      pDataBlock->nAllocSize = nAllocSizeOld;
L
[#1102]  
lihui 已提交
624
      *numOfRows = (int32_t)(pDataBlock->nAllocSize - pDataBlock->headerSize) / rowSize;
625
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
S
slguan 已提交
626
    }
H
hzcheng 已提交
627 628
  }

L
[#1102]  
lihui 已提交
629
  *numOfRows = (int32_t)(pDataBlock->nAllocSize - pDataBlock->headerSize) / rowSize;
L
lihui 已提交
630
  return TSDB_CODE_SUCCESS;
H
hzcheng 已提交
631 632
}

633
static int32_t tsSetBlockInfo(SSubmitBlk *pBlocks, const STableMeta *pTableMeta, int32_t numOfRows) {
634 635
  pBlocks->tid = pTableMeta->id.tid;
  pBlocks->uid = pTableMeta->id.uid;
H
hjxilinx 已提交
636
  pBlocks->sversion = pTableMeta->sversion;
637 638 639 640 641 642 643

  if (pBlocks->numOfRows + numOfRows >= INT16_MAX) {
    return TSDB_CODE_TSC_INVALID_SQL;
  } else {
    pBlocks->numOfRows += numOfRows;
    return TSDB_CODE_SUCCESS;
  }
H
hzcheng 已提交
644 645
}

S
slguan 已提交
646
// data block is disordered, sort it in ascending order
H
hjxilinx 已提交
647
void tscSortRemoveDataBlockDupRows(STableDataBlocks *dataBuf) {
648
  SSubmitBlk *pBlocks = (SSubmitBlk *)dataBuf->pData;
S
slguan 已提交
649 650

  // size is less than the total size, since duplicated rows may be removed yet.
651
  assert(pBlocks->numOfRows * dataBuf->rowSize + sizeof(SSubmitBlk) == dataBuf->size);
S
slguan 已提交
652

S
slguan 已提交
653 654 655 656 657
  // if use server time, this block must be ordered
  if (dataBuf->tsSource == TSDB_USE_SERVER_TS) {
    assert(dataBuf->ordered);
  }

S
slguan 已提交
658
  if (!dataBuf->ordered) {
659
    char *pBlockData = pBlocks->data;
S
slguan 已提交
660
    qsort(pBlockData, pBlocks->numOfRows, dataBuf->rowSize, rowDataCompar);
H
hzcheng 已提交
661

S
slguan 已提交
662 663
    int32_t i = 0;
    int32_t j = 1;
H
hzcheng 已提交
664

S
slguan 已提交
665
    while (j < pBlocks->numOfRows) {
S
slguan 已提交
666 667
      TSKEY ti = *(TSKEY *)(pBlockData + dataBuf->rowSize * i);
      TSKEY tj = *(TSKEY *)(pBlockData + dataBuf->rowSize * j);
H
hzcheng 已提交
668

S
slguan 已提交
669 670 671 672
      if (ti == tj) {
        ++j;
        continue;
      }
H
hzcheng 已提交
673

S
slguan 已提交
674 675 676 677 678 679 680 681 682
      int32_t nextPos = (++i);
      if (nextPos != j) {
        memmove(pBlockData + dataBuf->rowSize * nextPos, pBlockData + dataBuf->rowSize * j, dataBuf->rowSize);
      }

      ++j;
    }

    dataBuf->ordered = true;
H
hzcheng 已提交
683

S
slguan 已提交
684
    pBlocks->numOfRows = i + 1;
685
    dataBuf->size = sizeof(SSubmitBlk) + dataBuf->rowSize * pBlocks->numOfRows;
S
slguan 已提交
686
  }
S
slguan 已提交
687 688
}

689
static int32_t doParseInsertStatement(SSqlCmd* pCmd, char **str, SParsedDataColInfo *spd, int32_t *totalNum) {
690
  STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
691
  STableMeta     *pTableMeta = pTableMetaInfo->pTableMeta;
H
hjxilinx 已提交
692
  STableComInfo tinfo = tscGetTableInfo(pTableMeta);
H
hjxilinx 已提交
693
  
H
hjxilinx 已提交
694
  STableDataBlocks *dataBuf = NULL;
695 696
  int32_t ret = tscGetDataBlockFromList(pCmd->pTableList, pTableMeta->id.uid, TSDB_DEFAULT_PAYLOAD_SIZE,
                                        sizeof(SSubmitBlk), tinfo.rowSize, pTableMetaInfo->name, pTableMeta, &dataBuf, NULL);
H
hjxilinx 已提交
697 698 699 700
  if (ret != TSDB_CODE_SUCCESS) {
    return ret;
  }
  
L
lihui 已提交
701
  int32_t maxNumOfRows;
H
hjxilinx 已提交
702
  ret = tscAllocateMemIfNeed(dataBuf, tinfo.rowSize, &maxNumOfRows);
L
lihui 已提交
703
  if (TSDB_CODE_SUCCESS != ret) {
704
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
S
slguan 已提交
705
  }
706

707
  int32_t code = TSDB_CODE_TSC_INVALID_SQL;
Y
TD-1857  
yihaoDeng 已提交
708
  char *  tmpTokenBuf = calloc(1, 16*1024);  // used for deleting Escape character: \\, \', \"
L
[1292]  
lihui 已提交
709
  if (NULL == tmpTokenBuf) {
710
    return TSDB_CODE_TSC_OUT_OF_MEMORY;
L
[1292]  
lihui 已提交
711
  }
L
lihui 已提交
712

H
Haojun Liao 已提交
713
  int32_t numOfRows = tsParseValues(str, dataBuf, pTableMeta, maxNumOfRows, spd, pCmd, &code, tmpTokenBuf);
L
[1292]  
lihui 已提交
714
  free(tmpTokenBuf);
H
hzcheng 已提交
715
  if (numOfRows <= 0) {
L
[1292]  
lihui 已提交
716
    return code;
H
hzcheng 已提交
717 718
  }

S
slguan 已提交
719
  for (uint32_t i = 0; i < dataBuf->numOfParams; ++i) {
720
    SParamInfo *param = dataBuf->params + i;
S
slguan 已提交
721 722
    if (param->idx == -1) {
      param->idx = pCmd->numOfParams++;
723
      param->offset -= sizeof(SSubmitBlk);
S
slguan 已提交
724 725 726
    }
  }

727
  SSubmitBlk *pBlocks = (SSubmitBlk *)(dataBuf->pData);
728 729 730 731 732
  code = tsSetBlockInfo(pBlocks, pTableMeta, numOfRows);
  if (code != TSDB_CODE_SUCCESS) {
    tscInvalidSQLErrMsg(pCmd->payload, "too many rows in sql, total number of rows should be less than 32767", *str);
    return code;
  }
S
slguan 已提交
733

H
hjxilinx 已提交
734
  dataBuf->vgId = pTableMeta->vgroupInfo.vgId;
S
slguan 已提交
735
  dataBuf->numOfTables = 1;
H
hzcheng 已提交
736

S
slguan 已提交
737
  *totalNum += numOfRows;
H
hzcheng 已提交
738 739 740
  return TSDB_CODE_SUCCESS;
}

741
static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
S
slguan 已提交
742
  int32_t   index = 0;
H
Haojun Liao 已提交
743 744
  SStrToken sToken = {0};
  SStrToken tableToken = {0};
S
slguan 已提交
745
  int32_t   code = TSDB_CODE_SUCCESS;
746 747 748 749 750 751
  
  const int32_t TABLE_INDEX = 0;
  const int32_t STABLE_INDEX = 1;
  
  SSqlCmd *   pCmd = &pSql->cmd;
  SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
H
hzcheng 已提交
752

S
slguan 已提交
753
  char *sql = *sqlstr;
754

S
slguan 已提交
755 756 757 758
  // get the token of specified table
  index = 0;
  tableToken = tStrGetToken(sql, &index, false, 0, NULL);
  sql += index;
H
hzcheng 已提交
759

S
slguan 已提交
760 761
  char *cstart = NULL;
  char *cend = NULL;
H
hzcheng 已提交
762

S
slguan 已提交
763 764 765 766 767
  // skip possibly exists column list
  index = 0;
  sToken = tStrGetToken(sql, &index, false, 0, NULL);
  sql += index;

H
hzcheng 已提交
768
  int32_t numOfColList = 0;
S
slguan 已提交
769
  bool    createTable = false;
H
hzcheng 已提交
770

S
slguan 已提交
771 772 773
  if (sToken.type == TK_LP) {
    cstart = &sToken.z[0];
    index = 0;
H
hzcheng 已提交
774
    while (1) {
S
slguan 已提交
775 776 777
      sToken = tStrGetToken(sql, &index, false, 0, NULL);
      if (sToken.type == TK_RP) {
        cend = &sToken.z[0];
H
hzcheng 已提交
778 779 780 781 782 783
        break;
      }

      ++numOfColList;
    }

S
slguan 已提交
784 785
    sToken = tStrGetToken(sql, &index, false, 0, NULL);
    sql += index;
H
hzcheng 已提交
786 787 788
  }

  if (numOfColList == 0 && cstart != NULL) {
789
    return TSDB_CODE_TSC_INVALID_SQL;
H
hzcheng 已提交
790
  }
791
  
H
hjxilinx 已提交
792
  STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, TABLE_INDEX);
793 794
  
  if (sToken.type == TK_USING) {  // create table if not exists according to the super table
S
slguan 已提交
795 796 797 798
    index = 0;
    sToken = tStrGetToken(sql, &index, false, 0, NULL);
    sql += index;

H
Haojun Liao 已提交
799
    tscAllocPayload(pCmd, sizeof(STagData));
800
    
H
Haojun Liao 已提交
801
    //the source super table is moved to the secondary position of the pTableMetaInfo list
802
    if (pQueryInfo->numOfTables < 2) {
H
hjxilinx 已提交
803
      tscAddEmptyMetaInfo(pQueryInfo);
804
    }
H
hzcheng 已提交
805

H
hjxilinx 已提交
806
    STableMetaInfo *pSTableMeterMetaInfo = tscGetMetaInfo(pQueryInfo, STABLE_INDEX);
B
Bomin Zhang 已提交
807 808 809 810
    code = tscSetTableFullName(pSTableMeterMetaInfo, &sToken, pSql);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
811

812 813 814 815 816
    STagData *pTag = realloc(pCmd->pTagData, offsetof(STagData, data));
    if (pTag == NULL) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    memset(pTag, 0, offsetof(STagData, data));
B
Bomin Zhang 已提交
817
    tstrncpy(pTag->name, pSTableMeterMetaInfo->name, sizeof(pTag->name));
818 819
    pCmd->pTagData = pTag;

H
hjxilinx 已提交
820
    code = tscGetTableMeta(pSql, pSTableMeterMetaInfo);
H
hzcheng 已提交
821 822 823 824
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }

weixin_48148422's avatar
weixin_48148422 已提交
825
    if (!UTIL_TABLE_IS_SUPER_TABLE(pSTableMeterMetaInfo)) {
H
hjxilinx 已提交
826
      return tscInvalidSQLErrMsg(pCmd->payload, "create table only from super table is allowed", sToken.z);
S
slguan 已提交
827 828
    }

H
hjxilinx 已提交
829
    SSchema *pTagSchema = tscGetTableTagSchema(pSTableMeterMetaInfo->pTableMeta);
H
hjxilinx 已提交
830
    STableComInfo tinfo = tscGetTableInfo(pSTableMeterMetaInfo->pTableMeta);
H
hjxilinx 已提交
831
    
S
slguan 已提交
832 833 834
    index = 0;
    sToken = tStrGetToken(sql, &index, false, 0, NULL);
    sql += index;
L
lihui 已提交
835

836
    SParsedDataColInfo spd = {0};
H
hjxilinx 已提交
837 838
    
    uint8_t numOfTags = tscGetNumOfTags(pSTableMeterMetaInfo->pTableMeta);
L
lihui 已提交
839 840 841 842 843 844
    spd.numOfCols = numOfTags;

    // if specify some tags column
    if (sToken.type != TK_LP) {
      tscSetAssignedColumnInfo(&spd, pTagSchema, numOfTags);
    } else {
845 846
      /* insert into tablename (col1, col2,..., coln) using superTableName (tagName1, tagName2, ..., tagNamen)
       * tags(tagVal1, tagVal2, ..., tagValn) values(v1, v2,... vn); */
L
lihui 已提交
847 848 849 850 851 852 853 854 855 856 857
      int16_t offset[TSDB_MAX_COLUMNS] = {0};
      for (int32_t t = 1; t < numOfTags; ++t) {
        offset[t] = offset[t - 1] + pTagSchema[t - 1].bytes;
      }

      while (1) {
        index = 0;
        sToken = tStrGetToken(sql, &index, false, 0, NULL);
        sql += index;

        if (TK_STRING == sToken.type) {
H
Haojun Liao 已提交
858
          strdequote(sToken.z);
S
TD-1057  
Shengliang Guan 已提交
859
          sToken.n = (uint32_t)strtrim(sToken.z);
L
lihui 已提交
860 861 862 863 864 865 866 867 868 869 870 871
        }

        if (sToken.type == TK_RP) {
          break;
        }

        bool findColumnIndex = false;

        // todo speedup by using hash list
        for (int32_t t = 0; t < numOfTags; ++t) {
          if (strncmp(sToken.z, pTagSchema[t].name, sToken.n) == 0 && strlen(pTagSchema[t].name) == sToken.n) {
            SParsedColElem *pElem = &spd.elems[spd.numOfAssignedCols++];
872
            pElem->offset = offset[t];
L
lihui 已提交
873 874 875 876 877 878
            pElem->colIndex = t;

            if (spd.hasVal[t] == true) {
              return tscInvalidSQLErrMsg(pCmd->payload, "duplicated tag name", sToken.z);
            }

879
            spd.hasVal[t] = true;
L
lihui 已提交
880 881 882 883 884 885 886 887 888 889 890 891 892
            findColumnIndex = true;
            break;
          }
        }

        if (!findColumnIndex) {
          return tscInvalidSQLErrMsg(pCmd->payload, "invalid tag name", sToken.z);
        }
      }

      if (spd.numOfAssignedCols == 0 || spd.numOfAssignedCols > numOfTags) {
        return tscInvalidSQLErrMsg(pCmd->payload, "tag name expected", sToken.z);
      }
L
lihui 已提交
893 894 895 896

      index = 0;
      sToken = tStrGetToken(sql, &index, false, 0, NULL);
      sql += index;
L
lihui 已提交
897
    }
898

S
slguan 已提交
899
    if (sToken.type != TK_TAGS) {
L
lihui 已提交
900
      return tscInvalidSQLErrMsg(pCmd->payload, "keyword TAGS expected", sToken.z);
H
hzcheng 已提交
901 902
    }

B
Bomin Zhang 已提交
903 904 905 906 907
    SKVRowBuilder kvRowBuilder = {0};
    if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }

S
slguan 已提交
908 909
    uint32_t ignoreTokenTypes = TK_LP;
    uint32_t numOfIgnoreToken = 1;
L
lihui 已提交
910
    for (int i = 0; i < spd.numOfAssignedCols; ++i) {
B
Bomin Zhang 已提交
911
      SSchema* pSchema = pTagSchema + spd.elems[i].colIndex;
912

S
slguan 已提交
913 914 915
      index = 0;
      sToken = tStrGetToken(sql, &index, true, numOfIgnoreToken, &ignoreTokenTypes);
      sql += index;
H
Haojun Liao 已提交
916

B
Bomin Zhang 已提交
917 918 919 920 921
      if (TK_ILLEGAL == sToken.type) {
        tdDestroyKVRowBuilder(&kvRowBuilder);
        return TSDB_CODE_TSC_INVALID_SQL;
      }

H
Haojun Liao 已提交
922
      if (sToken.n == 0 || sToken.type == TK_RP) {
S
slguan 已提交
923 924
        break;
      }
H
hzcheng 已提交
925

S
slguan 已提交
926 927 928 929
      // Remove quotation marks
      if (TK_STRING == sToken.type) {
        sToken.z++;
        sToken.n -= 2;
H
hzcheng 已提交
930 931
      }

B
Bomin Zhang 已提交
932 933
      char tagVal[TSDB_MAX_TAGS_LEN];
      code = tsParseOneColumnData(pSchema, &sToken, tagVal, pCmd->payload, &sql, false, tinfo.precision);
H
hzcheng 已提交
934
      if (code != TSDB_CODE_SUCCESS) {
B
Bomin Zhang 已提交
935
        tdDestroyKVRowBuilder(&kvRowBuilder);
H
hjxilinx 已提交
936
        return code;
H
hzcheng 已提交
937
      }
B
Bomin Zhang 已提交
938 939

      tdAddColToKVRow(&kvRowBuilder, pSchema->colId, pSchema->type, tagVal);
L
lihui 已提交
940
    }
S
slguan 已提交
941

B
Bomin Zhang 已提交
942
    SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
B
Bomin Zhang 已提交
943 944 945 946 947
    tdDestroyKVRowBuilder(&kvRowBuilder);
    if (row == NULL) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    tdSortKVRowByColIdx(row);
948 949 950 951 952 953 954

    pTag = (STagData*)realloc(pCmd->pTagData, offsetof(STagData, data) + kvRowLen(row));
    if (pTag == NULL) {
      return TSDB_CODE_TSC_OUT_OF_MEMORY;
    }
    pCmd->pTagData = pTag;
    pTag->dataLen = htonl(kvRowLen(row));
B
Bomin Zhang 已提交
955
    kvRowCpy(pTag->data, row);
B
Bomin Zhang 已提交
956 957
    free(row);

L
lihui 已提交
958 959 960 961
    index = 0;
    sToken = tStrGetToken(sql, &index, false, 0, NULL);
    sql += index;
    if (sToken.n == 0 || sToken.type != TK_RP) {
Y
TD-934  
yihaoDeng 已提交
962
      return tscSQLSyntaxErrMsg(pCmd->payload, ") expected", sToken.z);
H
hzcheng 已提交
963 964 965
    }

    if (tscValidateName(&tableToken) != TSDB_CODE_SUCCESS) {
L
lihui 已提交
966
      return tscInvalidSQLErrMsg(pCmd->payload, "invalid table name", *sqlstr);
H
hzcheng 已提交
967 968
    }

H
Haojun Liao 已提交
969
    int32_t ret = tscSetTableFullName(pTableMetaInfo, &tableToken, pSql);
H
hzcheng 已提交
970 971 972 973 974
    if (ret != TSDB_CODE_SUCCESS) {
      return ret;
    }

    createTable = true;
H
hjxilinx 已提交
975
    code = tscGetMeterMetaEx(pSql, pTableMetaInfo, true);
976
    if (TSDB_CODE_TSC_ACTION_IN_PROGRESS == code) {
H
hjxilinx 已提交
977 978 979
      return code;
    }
    
H
hzcheng 已提交
980 981 982 983
  } else {
    if (cstart != NULL) {
      sql = cstart;
    } else {
S
slguan 已提交
984
      sql = sToken.z;
H
hzcheng 已提交
985
    }
B
Bomin Zhang 已提交
986
    code = tscGetMeterMetaEx(pSql, pTableMetaInfo, false);
H
hjxilinx 已提交
987
    
988
    if (pCmd->curSql == NULL) {
989
      assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS);
H
hjxilinx 已提交
990
    }
H
hzcheng 已提交
991 992
  }

S
TD-1057  
Shengliang Guan 已提交
993
  int32_t len = (int32_t)(cend - cstart + 1);
H
hzcheng 已提交
994 995
  if (cstart != NULL && createTable == true) {
    /* move the column list to start position of the next accessed points */
W
WangXin 已提交
996
    memmove(sql - len, cstart, len);
H
hzcheng 已提交
997 998 999 1000 1001
    *sqlstr = sql - len;
  } else {
    *sqlstr = sql;
  }

1002
  if (*sqlstr == NULL) {
1003
    code = TSDB_CODE_TSC_INVALID_SQL;
1004 1005
  }
  
H
hzcheng 已提交
1006 1007 1008
  return code;
}

H
Haojun Liao 已提交
1009
int validateTableName(char *tblName, int len, SStrToken* psTblToken) {
H
Haojun Liao 已提交
1010
  tstrncpy(psTblToken->z, tblName, TSDB_TABLE_FNAME_LEN);
S
slguan 已提交
1011

H
Hui Li 已提交
1012 1013
  psTblToken->n    = len;
  psTblToken->type = TK_ID;
B
Bomin Zhang 已提交
1014
  tSQLGetToken(psTblToken->z, &psTblToken->type);
S
slguan 已提交
1015

H
Hui Li 已提交
1016
  return tscValidateName(psTblToken);
H
huili 已提交
1017 1018
}

1019 1020 1021 1022 1023 1024 1025 1026 1027
static int32_t validateDataSource(SSqlCmd *pCmd, int8_t type, const char *sql) {
  if (pCmd->dataSourceType != 0 && pCmd->dataSourceType != type) {
    return tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES and FILE are not allowed to mix up", sql);
  }

  pCmd->dataSourceType = type;
  return TSDB_CODE_SUCCESS;
}

H
hzcheng 已提交
1028 1029 1030 1031 1032 1033 1034 1035 1036
/**
 * usage: insert into table1 values() () table2 values()()
 *
 * @param str
 * @param acct
 * @param db
 * @param pSql
 * @return
 */
H
Haojun Liao 已提交
1037
int tsParseInsertSql(SSqlObj *pSql) {
S
slguan 已提交
1038
  SSqlCmd *pCmd = &pSql->cmd;
H
Haojun Liao 已提交
1039
  char* str = pCmd->curSql;
1040

S
slguan 已提交
1041
  int32_t totalNum = 0;
1042 1043 1044 1045 1046
  int32_t code = TSDB_CODE_SUCCESS;

  SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
  assert(pQueryInfo != NULL);

1047 1048 1049 1050 1051
  STableMetaInfo *pTableMetaInfo = (pQueryInfo->numOfTables == 0)? tscAddEmptyMetaInfo(pQueryInfo):tscGetMetaInfo(pQueryInfo, 0);
  if (pTableMetaInfo == NULL) {
    terrno = TSDB_CODE_TSC_OUT_OF_MEMORY;
    code = terrno;
    return code;
1052
  }
H
hzcheng 已提交
1053

H
Haojun Liao 已提交
1054 1055 1056
  if ((code = tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) != TSDB_CODE_SUCCESS) {
    return code;
  }
H
hzcheng 已提交
1057

H
Haojun Liao 已提交
1058
  if (NULL == pCmd->pTableList) {
H
Haojun Liao 已提交
1059
    pCmd->pTableList = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
1060
    if (NULL == pCmd->pTableList) {
1061
      code = TSDB_CODE_TSC_OUT_OF_MEMORY;
1062
      goto _clean;
L
lihui 已提交
1063 1064
    }
  } else {
1065
    str = pCmd->curSql;
L
lihui 已提交
1066 1067
  }
  
1068
  tscDebug("%p create data block list hashList:%p", pSql, pCmd->pTableList);
H
hzcheng 已提交
1069 1070

  while (1) {
1071
    int32_t   index = 0;
H
Haojun Liao 已提交
1072
    SStrToken sToken = tStrGetToken(str, &index, false, 0, NULL);
1073 1074 1075 1076 1077 1078 1079 1080

    // no data in the sql string anymore.
    if (sToken.n == 0) {
      /*
       * if the data is from the data file, no data has been generated yet. So, there no data to
       * merge or submit, save the file path and parse the file in other routines.
       */
      if (pCmd->dataSourceType == DATA_FROM_DATA_FILE) {
S
slguan 已提交
1081 1082 1083
        goto _clean;
      }

1084 1085 1086 1087 1088
      /*
       * if no data has been generated during parsing the sql string, error msg will return
       * Otherwise, create the first submit block and submit to virtual node.
       */
      if (totalNum == 0) {
1089
        code = TSDB_CODE_TSC_INVALID_SQL;
1090
        goto _clean;
1091 1092
      } else {
        break;
H
hzcheng 已提交
1093 1094 1095
      }
    }

1096
    pCmd->curSql = sToken.z;
H
Haojun Liao 已提交
1097
    char buf[TSDB_TABLE_FNAME_LEN];
H
Haojun Liao 已提交
1098
    SStrToken sTblToken;
B
Bomin Zhang 已提交
1099
    sTblToken.z = buf;
S
slguan 已提交
1100
    // Check if the table name available or not
H
Hui Li 已提交
1101
    if (validateTableName(sToken.z, sToken.n, &sTblToken) != TSDB_CODE_SUCCESS) {
H
hjxilinx 已提交
1102
      code = tscInvalidSQLErrMsg(pCmd->payload, "table name invalid", sToken.z);
1103
      goto _clean;
H
huili 已提交
1104 1105
    }

H
Hui Li 已提交
1106
    if ((code = tscSetTableFullName(pTableMetaInfo, &sTblToken, pSql)) != TSDB_CODE_SUCCESS) {
1107
      goto _clean;
H
hzcheng 已提交
1108 1109
    }

1110 1111
    if ((code = tscCheckIfCreateTable(&str, pSql)) != TSDB_CODE_SUCCESS) {
      /*
H
Haojun Liao 已提交
1112 1113
       * After retrieving the table meta from server, the sql string will be parsed from the paused position.
       * And during the getTableMetaCallback function, the sql string will be parsed from the paused position.
1114
       */
1115
      if (TSDB_CODE_TSC_ACTION_IN_PROGRESS == code) {
H
hjxilinx 已提交
1116
        return code;
H
hzcheng 已提交
1117
      }
H
hjxilinx 已提交
1118
      
H
Haojun Liao 已提交
1119
      tscError("%p async insert parse error, code:%s", pSql, tstrerror(code));
1120
      pCmd->curSql = NULL;
1121
      goto _clean;
H
hzcheng 已提交
1122 1123
    }

weixin_48148422's avatar
weixin_48148422 已提交
1124
    if (UTIL_TABLE_IS_SUPER_TABLE(pTableMetaInfo)) {
H
hjxilinx 已提交
1125
      code = tscInvalidSQLErrMsg(pCmd->payload, "insert data into super table is not supported", NULL);
1126
      goto _clean;
H
hzcheng 已提交
1127 1128
    }

S
slguan 已提交
1129 1130 1131
    index = 0;
    sToken = tStrGetToken(str, &index, false, 0, NULL);
    str += index;
1132

S
slguan 已提交
1133
    if (sToken.n == 0) {
1134
      code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES or FILE required", sToken.z);
1135
      goto _clean;
H
hzcheng 已提交
1136
    }
H
hjxilinx 已提交
1137
    
H
hjxilinx 已提交
1138
    STableComInfo tinfo = tscGetTableInfo(pTableMetaInfo->pTableMeta);
H
hjxilinx 已提交
1139
    
S
slguan 已提交
1140
    if (sToken.type == TK_VALUES) {
H
hjxilinx 已提交
1141
      SParsedDataColInfo spd = {.numOfCols = tinfo.numOfColumns};
H
hjxilinx 已提交
1142 1143
      
      SSchema *pSchema = tscGetTableSchema(pTableMetaInfo->pTableMeta);
H
hjxilinx 已提交
1144
      tscSetAssignedColumnInfo(&spd, pSchema, tinfo.numOfColumns);
H
hzcheng 已提交
1145

1146
      if (validateDataSource(pCmd, DATA_FROM_SQL_STRING, sToken.z) != TSDB_CODE_SUCCESS) {
1147
        goto _clean;
H
hzcheng 已提交
1148 1149 1150 1151 1152 1153
      }

      /*
       * app here insert data in different vnodes, so we need to set the following
       * data in another submit procedure using async insert routines
       */
1154
      code = doParseInsertStatement(pCmd, &str, &spd, &totalNum);
H
hzcheng 已提交
1155
      if (code != TSDB_CODE_SUCCESS) {
1156
        goto _clean;
H
hzcheng 已提交
1157
      }
S
slguan 已提交
1158
    } else if (sToken.type == TK_FILE) {
1159
      if (validateDataSource(pCmd, DATA_FROM_DATA_FILE, sToken.z) != TSDB_CODE_SUCCESS) {
1160
        goto _clean;
H
hzcheng 已提交
1161 1162
      }

S
slguan 已提交
1163 1164
      index = 0;
      sToken = tStrGetToken(str, &index, false, 0, NULL);
1165 1166
      if (sToken.type != TK_STRING && sToken.type != TK_ID) {
        code = tscInvalidSQLErrMsg(pCmd->payload, "file path is required following keyword FILE", sToken.z);
1167
        goto _clean;
1168
      }
S
slguan 已提交
1169 1170
      str += index;
      if (sToken.n == 0) {
H
hjxilinx 已提交
1171
        code = tscInvalidSQLErrMsg(pCmd->payload, "file path is required following keyword FILE", sToken.z);
1172
        goto _clean;
H
hzcheng 已提交
1173 1174
      }

H
Haojun Liao 已提交
1175 1176
      strncpy(pCmd->payload, sToken.z, sToken.n);
      strdequote(pCmd->payload);
1177

H
Haojun Liao 已提交
1178
      // todo refactor extract method
H
hzcheng 已提交
1179
      wordexp_t full_path;
H
Haojun Liao 已提交
1180
      if (wordexp(pCmd->payload, &full_path, 0) != 0) {
H
hjxilinx 已提交
1181
        code = tscInvalidSQLErrMsg(pCmd->payload, "invalid filename", sToken.z);
1182
        goto _clean;
H
hzcheng 已提交
1183 1184
      }

H
Haojun Liao 已提交
1185 1186
      tstrncpy(pCmd->payload, full_path.we_wordv[0], pCmd->allocSize);
      wordfree(&full_path);
1187

S
slguan 已提交
1188
    } else if (sToken.type == TK_LP) {
H
hzcheng 已提交
1189
      /* insert into tablename(col1, col2,..., coln) values(v1, v2,... vn); */
1190
      STableMeta *pTableMeta = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0)->pTableMeta;
H
hjxilinx 已提交
1191
      SSchema *   pSchema = tscGetTableSchema(pTableMeta);
H
hzcheng 已提交
1192

1193
      if (validateDataSource(pCmd, DATA_FROM_SQL_STRING, sToken.z) != TSDB_CODE_SUCCESS) {
1194
        goto _clean;
H
hzcheng 已提交
1195 1196
      }

1197
      SParsedDataColInfo spd = {0};
H
hjxilinx 已提交
1198
      spd.numOfCols = tinfo.numOfColumns;
H
hzcheng 已提交
1199 1200

      int16_t offset[TSDB_MAX_COLUMNS] = {0};
H
hjxilinx 已提交
1201
      for (int32_t t = 1; t < tinfo.numOfColumns; ++t) {
H
hzcheng 已提交
1202 1203 1204 1205
        offset[t] = offset[t - 1] + pSchema[t - 1].bytes;
      }

      while (1) {
S
slguan 已提交
1206 1207 1208 1209 1210
        index = 0;
        sToken = tStrGetToken(str, &index, false, 0, NULL);
        str += index;

        if (TK_STRING == sToken.type) {
B
Bomin Zhang 已提交
1211
          tscDequoteAndTrimToken(&sToken);
S
slguan 已提交
1212 1213 1214
        }

        if (sToken.type == TK_RP) {
H
hzcheng 已提交
1215 1216 1217 1218 1219 1220
          break;
        }

        bool findColumnIndex = false;

        // todo speedup by using hash list
H
hjxilinx 已提交
1221
        for (int32_t t = 0; t < tinfo.numOfColumns; ++t) {
S
slguan 已提交
1222
          if (strncmp(sToken.z, pSchema[t].name, sToken.n) == 0 && strlen(pSchema[t].name) == sToken.n) {
S
slguan 已提交
1223
            SParsedColElem *pElem = &spd.elems[spd.numOfAssignedCols++];
H
hzcheng 已提交
1224 1225 1226 1227
            pElem->offset = offset[t];
            pElem->colIndex = t;

            if (spd.hasVal[t] == true) {
H
hjxilinx 已提交
1228
              code = tscInvalidSQLErrMsg(pCmd->payload, "duplicated column name", sToken.z);
1229
              goto _clean;
H
hzcheng 已提交
1230 1231 1232 1233 1234 1235 1236 1237
            }

            spd.hasVal[t] = true;
            findColumnIndex = true;
            break;
          }
        }

S
slguan 已提交
1238
        if (!findColumnIndex) {
H
hjxilinx 已提交
1239
          code = tscInvalidSQLErrMsg(pCmd->payload, "invalid column name", sToken.z);
1240
          goto _clean;
H
hzcheng 已提交
1241 1242 1243
        }
      }

H
hjxilinx 已提交
1244
      if (spd.numOfAssignedCols == 0 || spd.numOfAssignedCols > tinfo.numOfColumns) {
H
hjxilinx 已提交
1245
        code = tscInvalidSQLErrMsg(pCmd->payload, "column name expected", sToken.z);
1246
        goto _clean;
H
hzcheng 已提交
1247 1248
      }

S
slguan 已提交
1249 1250 1251 1252 1253
      index = 0;
      sToken = tStrGetToken(str, &index, false, 0, NULL);
      str += index;

      if (sToken.type != TK_VALUES) {
H
hjxilinx 已提交
1254
        code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES is expected", sToken.z);
1255
        goto _clean;
H
hzcheng 已提交
1256 1257
      }

1258
      code = doParseInsertStatement(pCmd, &str, &spd, &totalNum);
H
hzcheng 已提交
1259
      if (code != TSDB_CODE_SUCCESS) {
1260
        goto _clean;
H
hzcheng 已提交
1261 1262
      }
    } else {
H
hjxilinx 已提交
1263
      code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES or FILE are required", sToken.z);
1264
      goto _clean;
H
hzcheng 已提交
1265 1266 1267
    }
  }

S
slguan 已提交
1268 1269 1270 1271
  // we need to keep the data blocks if there are parameters in the sql
  if (pCmd->numOfParams > 0) {
    goto _clean;
  }
1272

1273 1274 1275
  if (taosHashGetSize(pCmd->pTableList) > 0) { // merge according to vgId
    if ((code = tscMergeTableDataBlocks(pSql)) != TSDB_CODE_SUCCESS) {
      goto _clean;
S
slguan 已提交
1276
    }
H
hzcheng 已提交
1277 1278 1279 1280 1281 1282
  }

  code = TSDB_CODE_SUCCESS;
  goto _clean;

_clean:
1283
  pCmd->curSql     = NULL;
1284
  pCmd->parseFinished  = 1;
H
hzcheng 已提交
1285 1286 1287
  return code;
}

H
Haojun Liao 已提交
1288
int tsInsertInitialCheck(SSqlObj *pSql) {
S
slguan 已提交
1289
  if (!pSql->pTscObj->writeAuth) {
1290
    return TSDB_CODE_TSC_NO_WRITE_AUTH;
S
slguan 已提交
1291
  }
H
hzcheng 已提交
1292

H
hjxilinx 已提交
1293
  int32_t  index = 0;
S
slguan 已提交
1294
  SSqlCmd *pCmd = &pSql->cmd;
1295

H
Haojun Liao 已提交
1296
  SStrToken sToken = tStrGetToken(pSql->sqlstr, &index, false, 0, NULL);
H
hjxilinx 已提交
1297
  assert(sToken.type == TK_INSERT || sToken.type == TK_IMPORT);
1298 1299 1300

  pCmd->count = 0;
  pCmd->command = TSDB_SQL_INSERT;
1301
  pSql->res.numOfRows = 0;
1302

1303
  SQueryInfo *pQueryInfo = tscGetQueryInfoDetailSafely(pCmd, pCmd->clauseIndex);
1304

H
Haojun Liao 已提交
1305
  TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT | pCmd->insertType);
1306 1307

  sToken = tStrGetToken(pSql->sqlstr, &index, false, 0, NULL);
S
slguan 已提交
1308
  if (sToken.type != TK_INTO) {
H
hjxilinx 已提交
1309
    return tscInvalidSQLErrMsg(pCmd->payload, "keyword INTO is expected", sToken.z);
S
slguan 已提交
1310
  }
1311

H
Haojun Liao 已提交
1312 1313
  pCmd->curSql = sToken.z + sToken.n;
  return TSDB_CODE_SUCCESS;
H
hzcheng 已提交
1314 1315
}

H
Haojun Liao 已提交
1316
int tsParseSql(SSqlObj *pSql, bool initial) {
H
hzcheng 已提交
1317
  int32_t ret = TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1318
  SSqlCmd* pCmd = &pSql->cmd;
H
Haojun Liao 已提交
1319

H
Haojun Liao 已提交
1320
  if ((!pCmd->parseFinished) && (!initial)) {
1321
    tscDebug("%p resume to parse sql: %s", pSql, pCmd->curSql);
H
[TD-98]  
hjxilinx 已提交
1322
  }
1323

B
Bomin Zhang 已提交
1324 1325 1326 1327 1328
  ret = tscAllocPayload(&pSql->cmd, TSDB_DEFAULT_PAYLOAD_SIZE);
  if (TSDB_CODE_SUCCESS != ret) {
    return ret;
  }

H
hjxilinx 已提交
1329
  if (tscIsInsertData(pSql->sqlstr)) {
H
hzcheng 已提交
1330
    /*
1331 1332
     * Set the fp before parse the sql string, in case of getTableMeta failed, in which
     * the error handle callback function can rightfully restore the user-defined callback function (fp).
H
hzcheng 已提交
1333
     */
H
Haojun Liao 已提交
1334
    if (initial && (pSql->cmd.insertType != TSDB_QUERY_TYPE_STMT_INSERT)) {
H
[TD-98]  
hjxilinx 已提交
1335
      pSql->fetchFp = pSql->fp;
H
hjxilinx 已提交
1336
      pSql->fp = (void(*)())tscHandleMultivnodeInsert;
H
hzcheng 已提交
1337
    }
B
Bomin Zhang 已提交
1338

B
Bomin Zhang 已提交
1339 1340 1341 1342 1343 1344 1345
    if (initial && ((ret = tsInsertInitialCheck(pSql)) != TSDB_CODE_SUCCESS)) {
      return ret;
    }

    // make a backup as tsParseInsertSql may modify the string
    char* sqlstr = strdup(pSql->sqlstr);
    ret = tsParseInsertSql(pSql);
1346
    if (sqlstr == NULL || pSql->parseRetry >= 1 || ret != TSDB_CODE_TSC_INVALID_SQL) {
B
Bomin Zhang 已提交
1347 1348
      free(sqlstr);
    } else {
B
Bomin Zhang 已提交
1349 1350 1351
      tscResetSqlCmdObj(pCmd, true);
      free(pSql->sqlstr);
      pSql->sqlstr = sqlstr;
1352
      pSql->parseRetry++;
B
Bomin Zhang 已提交
1353 1354 1355
      if ((ret = tsInsertInitialCheck(pSql)) == TSDB_CODE_SUCCESS) {
        ret = tsParseInsertSql(pSql);
      }
H
Haojun Liao 已提交
1356
    }
H
hzcheng 已提交
1357
  } else {
1358
    SSqlInfo SQLInfo = qSQLParse(pSql->sqlstr);
H
hzcheng 已提交
1359
    ret = tscToSQLCmd(pSql, &SQLInfo);
1360
    if (ret == TSDB_CODE_TSC_INVALID_SQL && pSql->parseRetry == 0 && SQLInfo.type == TSDB_SQL_NULL) {
B
Bomin Zhang 已提交
1361
      tscResetSqlCmdObj(pCmd, true);
1362
      pSql->parseRetry++;
B
Bomin Zhang 已提交
1363 1364
      ret = tscToSQLCmd(pSql, &SQLInfo);
    }
1365

H
hzcheng 已提交
1366 1367 1368 1369
    SQLInfoDestroy(&SQLInfo);
  }

  /*
1370
   * the pRes->code may be modified or released by another thread in tscTableMetaCallBack function,
H
Haojun Liao 已提交
1371 1372 1373
   * so do NOT use pRes->code to determine if the getTableMeta function
   * invokes new threads to get data from mgmt node or simply retrieves data from cache.
   * do NOT assign return code to pRes->code for the same reason since it may be released by another thread already.
H
hzcheng 已提交
1374 1375 1376 1377
   */
  return ret;
}

S
slguan 已提交
1378 1379 1380
static int doPackSendDataBlock(SSqlObj *pSql, int32_t numOfRows, STableDataBlocks *pTableDataBlocks) {
  int32_t  code = TSDB_CODE_SUCCESS;
  SSqlCmd *pCmd = &pSql->cmd;
H
Haojun Liao 已提交
1381
  pSql->res.numOfRows = 0;
S
slguan 已提交
1382

1383
  assert(pCmd->numOfClause == 1);
1384
  STableMeta *pTableMeta = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0)->pTableMeta;
S
slguan 已提交
1385

1386
  SSubmitBlk *pBlocks = (SSubmitBlk *)(pTableDataBlocks->pData);
H
Haojun Liao 已提交
1387 1388
  code = tsSetBlockInfo(pBlocks, pTableMeta, numOfRows);
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1389
    return tscInvalidSQLErrMsg(pCmd->payload, "too many rows in sql, total number of rows should be less than 32767", NULL);
H
Haojun Liao 已提交
1390
  }
S
slguan 已提交
1391

1392
  if ((code = tscMergeTableDataBlocks(pSql)) != TSDB_CODE_SUCCESS) {
S
slguan 已提交
1393 1394
    return code;
  }
S
slguan 已提交
1395

1396
  STableDataBlocks *pDataBlock = taosArrayGetP(pCmd->pDataBlocks, 0);
S
slguan 已提交
1397 1398 1399 1400
  if ((code = tscCopyDataBlockToPayload(pSql, pDataBlock)) != TSDB_CODE_SUCCESS) {
    return code;
  }

H
Haojun Liao 已提交
1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423
  return tscProcessSql(pSql);
}

typedef struct SImportFileSupport {
  SSqlObj *pSql;
  FILE    *fp;
} SImportFileSupport;

static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int code) {
  assert(param != NULL && tres != NULL);

  SSqlObj *pSql = tres;
  SSqlCmd *pCmd = &pSql->cmd;

  SImportFileSupport *pSupporter = (SImportFileSupport *) param;

  SSqlObj *pParentSql = pSupporter->pSql;
  FILE    *fp = pSupporter->fp;

  if (taos_errno(pSql) != TSDB_CODE_SUCCESS) {  // handle error
    assert(taos_errno(pSql) == code);

    taos_free_result(pSql);
S
TD-1848  
Shengliang Guan 已提交
1424
    tfree(pSupporter);
H
Haojun Liao 已提交
1425 1426 1427
    fclose(fp);

    pParentSql->res.code = code;
H
Haojun Liao 已提交
1428
    tscQueueAsyncRes(pParentSql);
H
Haojun Liao 已提交
1429
    return;
S
slguan 已提交
1430 1431
  }

H
Haojun Liao 已提交
1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448
  // accumulate the total submit records
  pParentSql->res.numOfRows += pSql->res.numOfRows;

  STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
  STableMeta *    pTableMeta = pTableMetaInfo->pTableMeta;
  SSchema *       pSchema = tscGetTableSchema(pTableMeta);
  STableComInfo   tinfo = tscGetTableInfo(pTableMeta);

  SParsedDataColInfo spd = {.numOfCols = tinfo.numOfColumns};
  tscSetAssignedColumnInfo(&spd, pSchema, tinfo.numOfColumns);

  size_t  n = 0;
  ssize_t readLen = 0;
  char *  line = NULL;
  int32_t count = 0;
  int32_t maxRows = 0;

H
Haojun Liao 已提交
1449 1450
  tscDestroyBlockArrayList(pSql->cmd.pDataBlocks);
  pCmd->pDataBlocks = taosArrayInit(1, POINTER_BYTES);
H
Haojun Liao 已提交
1451

H
Haojun Liao 已提交
1452 1453 1454 1455 1456 1457 1458
  STableDataBlocks *pTableDataBlock = NULL;
  int32_t ret = tscCreateDataBlock(TSDB_PAYLOAD_SIZE, tinfo.rowSize, sizeof(SSubmitBlk), pTableMetaInfo->name, pTableMeta, &pTableDataBlock);
  if (ret != TSDB_CODE_SUCCESS) {
//    return ret;
  }

  taosArrayPush(pCmd->pDataBlocks, &pTableDataBlock);
H
Haojun Liao 已提交
1459
  tscAllocateMemIfNeed(pTableDataBlock, tinfo.rowSize, &maxRows);
H
Haojun Liao 已提交
1460

H
Haojun Liao 已提交
1461 1462
  char *tokenBuf = calloc(1, 4096);

S
TD-1848  
Shengliang Guan 已提交
1463
  while ((readLen = tgetline(&line, &n, fp)) != -1) {
H
Haojun Liao 已提交
1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474
    if (('\r' == line[readLen - 1]) || ('\n' == line[readLen - 1])) {
      line[--readLen] = 0;
    }

    if (readLen == 0) {
      continue;
    }

    char *lineptr = line;
    strtolower(line, line);

H
Haojun Liao 已提交
1475
    int32_t len = tsParseOneRowData(&lineptr, pTableDataBlock, pSchema, &spd, pCmd, tinfo.precision, &code, tokenBuf);
H
Haojun Liao 已提交
1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487
    if (len <= 0 || pTableDataBlock->numOfParams > 0) {
      pSql->res.code = code;
      break;
    }

    pTableDataBlock->size += len;

    if (++count >= maxRows) {
      break;
    }
  }

S
TD-1848  
Shengliang Guan 已提交
1488
  tfree(tokenBuf);
H
Haojun Liao 已提交
1489 1490 1491
  free(line);

  if (count > 0) {
H
Haojun Liao 已提交
1492 1493
    code = doPackSendDataBlock(pSql, count, pTableDataBlock);
    if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1494
      pParentSql->res.code = code;
H
Haojun Liao 已提交
1495 1496
      tscQueueAsyncRes(pParentSql);
      return;
H
Haojun Liao 已提交
1497 1498 1499 1500
    }

  } else {
    taos_free_result(pSql);
S
TD-1848  
Shengliang Guan 已提交
1501
    tfree(pSupporter);
H
Haojun Liao 已提交
1502 1503 1504 1505 1506
    fclose(fp);

    pParentSql->fp = pParentSql->fetchFp;

    // all data has been sent to vnode, call user function
S
TD-1057  
Shengliang Guan 已提交
1507
    int32_t v = (pParentSql->res.code != TSDB_CODE_SUCCESS) ? pParentSql->res.code : (int32_t)pParentSql->res.numOfRows;
H
Haojun Liao 已提交
1508 1509
    (*pParentSql->fp)(pParentSql->param, pParentSql, v);
  }
S
slguan 已提交
1510 1511
}

H
Haojun Liao 已提交
1512
void tscProcessMultiVnodesImportFromFile(SSqlObj *pSql) {
S
slguan 已提交
1513
  SSqlCmd *pCmd = &pSql->cmd;
H
hzcheng 已提交
1514 1515 1516 1517
  if (pCmd->command != TSDB_SQL_INSERT) {
    return;
  }

H
Haojun Liao 已提交
1518
  assert(pCmd->dataSourceType == DATA_FROM_DATA_FILE  && strlen(pCmd->payload) != 0);
H
Haojun Liao 已提交
1519

H
Haojun Liao 已提交
1520
  SImportFileSupport *pSupporter = calloc(1, sizeof(SImportFileSupport));
H
Haojun Liao 已提交
1521 1522 1523
  SSqlObj *pNew = createSubqueryObj(pSql, 0, parseFileSendDataBlock, pSupporter, TSDB_SQL_INSERT, NULL);

  pNew->cmd.pDataBlocks = taosArrayInit(4, POINTER_BYTES);
H
Haojun Liao 已提交
1524
  pCmd->count = 1;
H
Haojun Liao 已提交
1525

H
Haojun Liao 已提交
1526 1527 1528 1529
  FILE *fp = fopen(pCmd->payload, "r");
  if (fp == NULL) {
    pSql->res.code = TAOS_SYSTEM_ERROR(errno);
    tscError("%p failed to open file %s to load data from file, code:%s", pSql, pCmd->payload, tstrerror(pSql->res.code));
H
hzcheng 已提交
1530

S
TD-1848  
Shengliang Guan 已提交
1531
    tfree(pSupporter)
H
Haojun Liao 已提交
1532
    tscQueueAsyncRes(pSql);
S
slguan 已提交
1533

H
Haojun Liao 已提交
1534 1535
    return;
  }
S
slguan 已提交
1536

H
Haojun Liao 已提交
1537 1538
  pSupporter->pSql = pSql;
  pSupporter->fp = fp;
1539

H
Haojun Liao 已提交
1540
  parseFileSendDataBlock(pSupporter, pNew, 0);
H
hzcheng 已提交
1541
}