sclfunc.c 56.9 KB
Newer Older
H
Haojun Liao 已提交
1 2
#include "function.h"
#include "scalar.h"
H
Haojun Liao 已提交
3
#include "sclInt.h"
D
dapan1121 已提交
4
#include "sclvector.h"
5
#include "tdatablock.h"
wmmhello's avatar
wmmhello 已提交
6
#include "tjson.h"
7 8
#include "ttime.h"
#include "vnode.h"
9

G
Ganlin Zhao 已提交
10 11 12 13 14
typedef float (*_float_fn)(float);
typedef double (*_double_fn)(double);
typedef double (*_double_fn_2)(double, double);
typedef int (*_conv_fn)(int);
typedef void (*_trim_fn)(char *, char*, int32_t, int32_t);
G
Ganlin Zhao 已提交
15
typedef int16_t (*_len_fn)(char *, int32_t);
G
Ganlin Zhao 已提交
16

17
/** Math functions **/
18 19 20 21 22
static double tlog(double v) {
  return log(v);
}

static double tlog2(double v, double base) {
23 24 25 26 27 28
  double a = log(v);
  double b = log(base);
  if (isnan(a) || isinf(a)) {
    return a;
  } else if (isnan(b) || isinf(b)) {
    return b;
29 30
  } else {
    return a / b;
31
  }
32 33
}

34
int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
H
Haojun Liao 已提交
35 36 37 38
  SColumnInfoData *pInputData  = pInput->columnData;
  SColumnInfoData *pOutputData = pOutput->columnData;

  int32_t type = GET_PARAM_TYPE(pInput);
39

H
Haojun Liao 已提交
40
  switch (type) {
41
    case TSDB_DATA_TYPE_FLOAT: {
H
Haojun Liao 已提交
42 43
      float *in  = (float *)pInputData->pData;
      float *out = (float *)pOutputData->pData;
44
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
45 46
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
47 48
          continue;
        }
49
        out[i] = (in[i] >= 0)? in[i] : -in[i];
50
      }
H
Haojun Liao 已提交
51
      break;
52 53 54
    }

    case TSDB_DATA_TYPE_DOUBLE: {
H
Haojun Liao 已提交
55 56
      double *in  = (double *)pInputData->pData;
      double *out = (double *)pOutputData->pData;
57
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
58 59
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
60 61
          continue;
        }
62
        out[i] = (in[i] >= 0)? in[i] : -in[i];
H
Haojun Liao 已提交
63 64
      }
      break;
65 66
    }

H
Haojun Liao 已提交
67
    case TSDB_DATA_TYPE_TINYINT: {
H
Haojun Liao 已提交
68 69
      int8_t *in  = (int8_t *)pInputData->pData;
      int8_t *out = (int8_t *)pOutputData->pData;
70
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
71 72
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
73 74
          continue;
        }
75
        out[i] = (in[i] >= 0)? in[i] : -in[i];
76
      }
H
Haojun Liao 已提交
77 78
      break;
    }
79

H
Haojun Liao 已提交
80
    case TSDB_DATA_TYPE_SMALLINT: {
H
Haojun Liao 已提交
81 82
      int16_t *in  = (int16_t *)pInputData->pData;
      int16_t *out = (int16_t *)pOutputData->pData;
83
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
84 85
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
86 87
          continue;
        }
88
        out[i] = (in[i] >= 0)? in[i] : -in[i];
89
      }
H
Haojun Liao 已提交
90 91
      break;
    }
92

H
Haojun Liao 已提交
93
    case TSDB_DATA_TYPE_INT: {
H
Haojun Liao 已提交
94 95
      int32_t *in  = (int32_t *)pInputData->pData;
      int32_t *out = (int32_t *)pOutputData->pData;
96
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
97 98
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
99 100
          continue;
        }
101
        out[i] = (in[i] >= 0)? in[i] : -in[i];
102
      }
H
Haojun Liao 已提交
103 104
      break;
    }
105

H
Haojun Liao 已提交
106
    case TSDB_DATA_TYPE_BIGINT: {
H
Haojun Liao 已提交
107 108
      int64_t *in  = (int64_t *)pInputData->pData;
      int64_t *out = (int64_t *)pOutputData->pData;
109
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
110 111
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
112 113
          continue;
        }
114
        out[i] = (in[i] >= 0)? in[i] : -in[i];
115
      }
H
Haojun Liao 已提交
116
      break;
117 118
    }

119 120 121 122 123 124 125
    case TSDB_DATA_TYPE_NULL: {
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
        colDataAppendNULL(pOutputData, i);
      }
      break;
    }

H
Haojun Liao 已提交
126
    default: {
127
      colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL);
128 129
    }
  }
130

H
Haojun Liao 已提交
131
  pOutput->numOfRows = pInput->numOfRows;
132
  return TSDB_CODE_SUCCESS;
133 134
}

135
static int32_t doScalarFunctionUnique(SScalarParam *pInput, int32_t inputNum, SScalarParam* pOutput, _double_fn valFn) {
H
Haojun Liao 已提交
136
  int32_t type = GET_PARAM_TYPE(pInput);
137

138
  SColumnInfoData *pInputData = pInput->columnData;
H
Haojun Liao 已提交
139
  SColumnInfoData *pOutputData = pOutput->columnData;
140

141
  _getDoubleValue_fn_t getValueFn = getVectorDoubleValueFn(type);
142

143
  double *out = (double *)pOutputData->pData;
144

145
  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
146
    if (colDataIsNull_s(pInputData, i) || IS_NULL_TYPE(type)) {
147
      colDataAppendNULL(pOutputData, i);
148
      continue;
149
    }
150 151 152 153 154 155
    double result = valFn(getValueFn(pInputData->pData, i));
    if (isinf(result) || isnan(result)) {
      colDataAppendNULL(pOutputData, i);
    } else {
      out[i] = result;
    }
156
  }
157

H
Haojun Liao 已提交
158
  pOutput->numOfRows = pInput->numOfRows;
159
  return TSDB_CODE_SUCCESS;
160 161
}

162
static int32_t doScalarFunctionUnique2(SScalarParam *pInput, int32_t inputNum, SScalarParam* pOutput, _double_fn_2 valFn) {
G
Ganlin Zhao 已提交
163
  SColumnInfoData *pInputData[2];
H
Haojun Liao 已提交
164
  SColumnInfoData *pOutputData = pOutput->columnData;
G
Ganlin Zhao 已提交
165
  _getDoubleValue_fn_t getValueFn[2];
166

G
Ganlin Zhao 已提交
167 168 169 170
  for (int32_t i = 0; i < inputNum; ++i) {
    pInputData[i] = pInput[i].columnData;
    getValueFn[i]= getVectorDoubleValueFn(GET_PARAM_TYPE(&pInput[i]));
  }
171

172
  double *out = (double *)pOutputData->pData;
173
  double result;
174

175 176 177
  bool hasNullType = (IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[0])) ||
                      IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[1])));

wafwerar's avatar
wafwerar 已提交
178
  int32_t numOfRows = TMAX(pInput[0].numOfRows, pInput[1].numOfRows);
179 180 181
  if (pInput[0].numOfRows == pInput[1].numOfRows) {
    for (int32_t i = 0; i < numOfRows; ++i) {
      if (colDataIsNull_s(pInputData[0], i) ||
182 183
          colDataIsNull_s(pInputData[1], i) ||
          hasNullType) {
184 185 186 187 188 189 190 191 192
        colDataAppendNULL(pOutputData, i);
        continue;
      }
      result = valFn(getValueFn[0](pInputData[0]->pData, i), getValueFn[1](pInputData[1]->pData, i));
      if (isinf(result) || isnan(result)) {
        colDataAppendNULL(pOutputData, i);
      } else {
        out[i] = result;
      }
193
    }
194
  } else if (pInput[0].numOfRows == 1) { //left operand is constant
195
    if (colDataIsNull_s(pInputData[0], 0) || hasNullType) {
196
      colDataAppendNNULL(pOutputData, 0, pInput[1].numOfRows);
197
    } else {
198 199 200 201 202
      for (int32_t i = 0; i < numOfRows; ++i) {
        if (colDataIsNull_s(pInputData[1], i)) {
          colDataAppendNULL(pOutputData, i);
          continue;
        }
203

204 205 206 207 208 209 210 211
        result = valFn(getValueFn[0](pInputData[0]->pData, 0), getValueFn[1](pInputData[1]->pData, i));
        if (isinf(result) || isnan(result)) {
          colDataAppendNULL(pOutputData, i);
          continue;
        }

        out[i] = result;
      }
212
    }
213
  } else if (pInput[1].numOfRows == 1) {
214
    if (colDataIsNull_s(pInputData[1], 0) || hasNullType) {
215
      colDataAppendNNULL(pOutputData, 0, pInput[0].numOfRows);
216
    } else {
217 218 219 220 221 222 223 224 225 226 227 228 229 230
      for (int32_t i = 0; i < numOfRows; ++i) {
        if (colDataIsNull_s(pInputData[0], i)) {
          colDataAppendNULL(pOutputData, i);
          continue;
        }

        result = valFn(getValueFn[0](pInputData[0]->pData, i), getValueFn[1](pInputData[1]->pData, 0));
        if (isinf(result) || isnan(result)) {
          colDataAppendNULL(pOutputData, i);
          continue;
        }

        out[i] = result;
      }
231
    }
232
  }
233

234
  pOutput->numOfRows = numOfRows;
235 236 237
  return TSDB_CODE_SUCCESS;
}

238
static int32_t doScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam* pOutput, _float_fn f1, _double_fn d1) {
H
Haojun Liao 已提交
239
  int32_t type = GET_PARAM_TYPE(pInput);
240

H
Haojun Liao 已提交
241 242
  SColumnInfoData *pInputData  = pInput->columnData;
  SColumnInfoData *pOutputData = pOutput->columnData;
243

H
Haojun Liao 已提交
244 245 246 247
  switch (type) {
    case TSDB_DATA_TYPE_FLOAT: {
      float *in  = (float *)pInputData->pData;
      float *out = (float *)pOutputData->pData;
248

249
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
250 251
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
252 253
          continue;
        }
254
        out[i] = f1(in[i]);
255
      }
H
Haojun Liao 已提交
256 257
      break;
    }
258

H
Haojun Liao 已提交
259 260 261
    case TSDB_DATA_TYPE_DOUBLE: {
      double *in  = (double *)pInputData->pData;
      double *out = (double *)pOutputData->pData;
262

263
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
264 265
        if (colDataIsNull_s(pInputData, i)) {
          colDataAppendNULL(pOutputData, i);
H
Haojun Liao 已提交
266 267
          continue;
        }
268
        out[i] = d1(in[i]);
269
      }
H
Haojun Liao 已提交
270 271 272
      break;
    }

273 274 275 276 277 278 279
    case TSDB_DATA_TYPE_NULL: {
      for (int32_t i = 0; i < pInput->numOfRows; ++i) {
        colDataAppendNULL(pOutputData, i);
      }
      break;
    }

H
Haojun Liao 已提交
280
    default: {
281
      colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL);
282
    }
283 284
  }

H
Haojun Liao 已提交
285
  pOutput->numOfRows = pInput->numOfRows;
286 287
  return TSDB_CODE_SUCCESS;
}
288

G
Ganlin Zhao 已提交
289
/** String functions **/
290
static int16_t tlength(char *input, int32_t type) {
G
Ganlin Zhao 已提交
291 292 293
  return varDataLen(input);
}

294
static int16_t tcharlength(char *input, int32_t type) {
G
Ganlin Zhao 已提交
295 296 297 298
  if (type == TSDB_DATA_TYPE_VARCHAR) {
    return varDataLen(input);
  } else { //NCHAR
    return varDataLen(input) / TSDB_NCHAR_SIZE;
G
Ganlin Zhao 已提交
299
  }
G
Ganlin Zhao 已提交
300
}
G
Ganlin Zhao 已提交
301

302
static void tltrim(char *input, char *output, int32_t type, int32_t charLen) {
G
Ganlin Zhao 已提交
303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318
  int32_t numOfSpaces = 0;
  if (type == TSDB_DATA_TYPE_VARCHAR) {
    for (int32_t i = 0; i < charLen; ++i) {
      if (!isspace(*(varDataVal(input) + i))) {
        break;
      }
      numOfSpaces++;
    }
  } else { //NCHAR
    for (int32_t i = 0; i < charLen; ++i) {
      if (!iswspace(*((uint32_t *)varDataVal(input) + i))) {
        break;
      }
      numOfSpaces++;
    }
  }
G
Ganlin Zhao 已提交
319

G
Ganlin Zhao 已提交
320 321 322 323 324 325 326 327
  int32_t resLen;
  if (type == TSDB_DATA_TYPE_VARCHAR) {
    resLen = charLen - numOfSpaces;
    memcpy(varDataVal(output), varDataVal(input) + numOfSpaces, resLen);
  } else {
    resLen = (charLen - numOfSpaces) * TSDB_NCHAR_SIZE;
    memcpy(varDataVal(output), varDataVal(input) + numOfSpaces * TSDB_NCHAR_SIZE, resLen);
  }
G
Ganlin Zhao 已提交
328

G
Ganlin Zhao 已提交
329 330 331
  varDataSetLen(output, resLen);
}

332
static void trtrim(char *input, char *output, int32_t type, int32_t charLen) {
G
Ganlin Zhao 已提交
333 334 335 336 337 338 339 340 341
  int32_t numOfSpaces = 0;
  if (type == TSDB_DATA_TYPE_VARCHAR) {
    for (int32_t i = charLen - 1; i >= 0; --i) {
      if (!isspace(*(varDataVal(input) + i))) {
        break;
      }
      numOfSpaces++;
    }
  } else { //NCHAR
342
    for (int32_t i = charLen - 1; i >= 0; --i) {
G
Ganlin Zhao 已提交
343 344 345 346
      if (!iswspace(*((uint32_t *)varDataVal(input) + i))) {
        break;
      }
      numOfSpaces++;
G
Ganlin Zhao 已提交
347
    }
G
Ganlin Zhao 已提交
348
  }
G
Ganlin Zhao 已提交
349

G
Ganlin Zhao 已提交
350 351 352 353 354
  int32_t resLen;
  if (type == TSDB_DATA_TYPE_VARCHAR) {
    resLen = charLen - numOfSpaces;
  } else {
    resLen = (charLen - numOfSpaces) * TSDB_NCHAR_SIZE;
G
Ganlin Zhao 已提交
355
  }
G
Ganlin Zhao 已提交
356
  memcpy(varDataVal(output), varDataVal(input), resLen);
G
Ganlin Zhao 已提交
357

G
Ganlin Zhao 已提交
358
  varDataSetLen(output, resLen);
G
Ganlin Zhao 已提交
359 360
}

361
static int32_t doLengthFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput, _len_fn lenFn) {
G
Ganlin Zhao 已提交
362 363 364 365 366
  int32_t type = GET_PARAM_TYPE(pInput);

  SColumnInfoData *pInputData  = pInput->columnData;
  SColumnInfoData *pOutputData = pOutput->columnData;

367 368
  ASSERT(pOutputData->info.type == TSDB_DATA_TYPE_BIGINT);
  int64_t *out = (int64_t *)pOutputData->pData;
G
Ganlin Zhao 已提交
369 370

  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
371
    if (colDataIsNull_s(pInputData, i)) {
372
      colDataAppendNULL(pOutputData, i);
G
Ganlin Zhao 已提交
373 374 375
      continue;
    }

376
    char *in = colDataGetData(pInputData, i);
G
Ganlin Zhao 已提交
377
    out[i] = lenFn(in, type);
G
Ganlin Zhao 已提交
378 379 380 381 382
  }

  pOutput->numOfRows = pInput->numOfRows;
  return TSDB_CODE_SUCCESS;
}
383

384 385
static int32_t concatCopyHelper(const char *input, char *output, bool hasNchar, int32_t type, int16_t *dataLen) {
  if (hasNchar && type == TSDB_DATA_TYPE_VARCHAR) {
386
    TdUcs4 *newBuf = taosMemoryCalloc((varDataLen(input) + 1) * TSDB_NCHAR_SIZE, 1);
387 388
    int32_t len = varDataLen(input);
    bool ret = taosMbsToUcs4(varDataVal(input), len, newBuf, (varDataLen(input) + 1) * TSDB_NCHAR_SIZE, &len);
389 390 391 392 393 394 395 396 397 398 399 400
    if (!ret) {
      taosMemoryFree(newBuf);
      return TSDB_CODE_FAILED;
    }
    memcpy(varDataVal(output) + *dataLen, newBuf, varDataLen(input) * TSDB_NCHAR_SIZE);
    *dataLen += varDataLen(input) * TSDB_NCHAR_SIZE;
    taosMemoryFree(newBuf);
  } else {
    memcpy(varDataVal(output) + *dataLen, varDataVal(input), varDataLen(input));
    *dataLen += varDataLen(input);
  }
  return TSDB_CODE_SUCCESS;
401
}
G
Ganlin Zhao 已提交
402

403 404 405 406 407 408 409 410 411 412 413 414 415
static int32_t getNumOfNullEntries(SColumnInfoData *pColumnInfoData, int32_t numOfRows) {
  int32_t numOfNulls = 0;
  if (!pColumnInfoData->hasNull) {
    return numOfNulls;
  }
  for (int i = 0; i < numOfRows; ++i) {
    if (pColumnInfoData->varmeta.offset[i] == -1) {
      numOfNulls++;
    }
  }
  return numOfNulls;
}

G
Ganlin Zhao 已提交
416
int32_t concatFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
417
  int32_t ret = TSDB_CODE_SUCCESS;
G
Ganlin Zhao 已提交
418 419
  SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *));
  SColumnInfoData *pOutputData = pOutput->columnData;
420
  char **input = taosMemoryCalloc(inputNum, POINTER_BYTES);
421
  char *outputBuf = NULL;
G
Ganlin Zhao 已提交
422

423
  int32_t inputLen = 0;
424
  int32_t numOfRows = 0;
425
  bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false;
426 427 428 429 430
  for (int32_t i = 0; i < inputNum; ++i) {
    if (pInput[i].numOfRows > numOfRows) {
      numOfRows = pInput[i].numOfRows;
    }
  }
G
Ganlin Zhao 已提交
431 432
  for (int32_t i = 0; i < inputNum; ++i) {
    pInputData[i] = pInput[i].columnData;
433
    int32_t factor = 1;
434
    if (hasNchar && (GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_VARCHAR)) {
435 436
      factor = TSDB_NCHAR_SIZE;
    }
437 438

    int32_t numOfNulls = getNumOfNullEntries(pInputData[i], pInput[i].numOfRows);
439
    if (pInput[i].numOfRows == 1) {
440
      inputLen += (pInputData[i]->varmeta.length - VARSTR_HEADER_SIZE) * factor * (numOfRows - numOfNulls);
441
    } else {
442
      inputLen += (pInputData[i]->varmeta.length - (numOfRows - numOfNulls) * VARSTR_HEADER_SIZE) * factor;
443
    }
444 445
  }

446 447 448
  int32_t outputLen = inputLen + numOfRows * VARSTR_HEADER_SIZE;
  outputBuf = taosMemoryCalloc(outputLen, 1);
  char *output = outputBuf;
G
Ganlin Zhao 已提交
449

450
  for (int32_t k = 0; k < numOfRows; ++k) {
451
    bool hasNull = false;
G
Ganlin Zhao 已提交
452
    for (int32_t i = 0; i < inputNum; ++i) {
453
      if (colDataIsNull_s(pInputData[i], k) ||
454
          IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[i]))) {
455
        colDataAppendNULL(pOutputData, k);
G
Ganlin Zhao 已提交
456 457 458 459 460 461 462 463 464
        hasNull = true;
        break;
      }
    }

    if (hasNull) {
      continue;
    }

465

G
Ganlin Zhao 已提交
466 467
    int16_t dataLen = 0;
    for (int32_t i = 0; i < inputNum; ++i) {
468 469
      int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k;
      input[i] = colDataGetData(pInputData[i], rowIdx);
470 471 472 473

      ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen);
      if (ret != TSDB_CODE_SUCCESS) {
        goto DONE;
474
      }
G
Ganlin Zhao 已提交
475
    }
476
    varDataSetLen(output, dataLen);
477 478
    colDataAppend(pOutputData, k, output, false);
    output += varDataTLen(output);
G
Ganlin Zhao 已提交
479 480
  }

481
  pOutput->numOfRows = numOfRows;
482 483

DONE:
484
  taosMemoryFree(input);
485
  taosMemoryFree(outputBuf);
G
Ganlin Zhao 已提交
486 487
  taosMemoryFree(pInputData);

488
  return ret;
G
Ganlin Zhao 已提交
489 490
}

491

G
Ganlin Zhao 已提交
492
int32_t concatWsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
493
  int32_t ret = TSDB_CODE_SUCCESS;
G
Ganlin Zhao 已提交
494 495
  SColumnInfoData **pInputData = taosMemoryCalloc(inputNum, sizeof(SColumnInfoData *));
  SColumnInfoData *pOutputData = pOutput->columnData;
496
  char **input = taosMemoryCalloc(inputNum, POINTER_BYTES);
497
  char *outputBuf = NULL;
G
Ganlin Zhao 已提交
498

499
  int32_t inputLen = 0;
500
  int32_t numOfRows = 0;
501
  bool hasNchar = (GET_PARAM_TYPE(pOutput) == TSDB_DATA_TYPE_NCHAR) ? true : false;
502
  for (int32_t i = 1; i < inputNum; ++i) {
503 504 505 506
    if (pInput[i].numOfRows > numOfRows) {
      numOfRows = pInput[i].numOfRows;
    }
  }
G
Ganlin Zhao 已提交
507 508
  for (int32_t i = 0; i < inputNum; ++i) {
    pInputData[i] = pInput[i].columnData;
509
    int32_t factor = 1;
510
    if (hasNchar && (GET_PARAM_TYPE(&pInput[i]) == TSDB_DATA_TYPE_VARCHAR)) {
511 512
      factor = TSDB_NCHAR_SIZE;
    }
513 514

    int32_t numOfNulls = getNumOfNullEntries(pInputData[i], pInput[i].numOfRows);
515 516
    if (i == 0) {
      // calculate required separator space
517
      inputLen += (pInputData[0]->varmeta.length - VARSTR_HEADER_SIZE) * (numOfRows - numOfNulls) * (inputNum - 2) * factor;
518
    } else if (pInput[i].numOfRows == 1) {
519
      inputLen += (pInputData[i]->varmeta.length - VARSTR_HEADER_SIZE) * (numOfRows - numOfNulls) * factor;
520
    } else {
521
      inputLen += (pInputData[i]->varmeta.length - (numOfRows - numOfNulls) * VARSTR_HEADER_SIZE) * factor;
522 523 524
    }
  }

525 526 527
  int32_t outputLen = inputLen + numOfRows * VARSTR_HEADER_SIZE;
  outputBuf = taosMemoryCalloc(outputLen, 1);
  char *output = outputBuf;
G
Ganlin Zhao 已提交
528

529
  for (int32_t k = 0; k < numOfRows; ++k) {
530
    if (colDataIsNull_s(pInputData[0], k) ||
531
        IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[0]))) {
532
      colDataAppendNULL(pOutputData, k);
G
Ganlin Zhao 已提交
533 534 535 536
      continue;
    }

    int16_t dataLen = 0;
537
    bool hasNull = false;
G
Ganlin Zhao 已提交
538
    for (int32_t i = 1; i < inputNum; ++i) {
539
      if (colDataIsNull_s(pInputData[i], k) ||
540
          IS_NULL_TYPE(GET_PARAM_TYPE(&pInput[i]))) {
541 542
        hasNull = true;
        break;
G
Ganlin Zhao 已提交
543 544
      }

545 546
      int32_t rowIdx = (pInput[i].numOfRows == 1) ? 0 : k;
      input[i] = colDataGetData(pInputData[i], rowIdx);
547

548 549 550
      ret = concatCopyHelper(input[i], output, hasNchar, GET_PARAM_TYPE(&pInput[i]), &dataLen);
      if (ret != TSDB_CODE_SUCCESS) {
        goto DONE;
551
      }
G
Ganlin Zhao 已提交
552

553

G
Ganlin Zhao 已提交
554 555
      if (i < inputNum - 1) {
        //insert the separator
556 557
        char *sep = (pInput[0].numOfRows == 1) ? colDataGetData(pInputData[0], 0) : colDataGetData(pInputData[0], k);
         ret = concatCopyHelper(sep, output, hasNchar, GET_PARAM_TYPE(&pInput[0]), &dataLen);
558
        if (ret != TSDB_CODE_SUCCESS) {
559
          goto DONE;
560
        }
G
Ganlin Zhao 已提交
561 562
      }
    }
563 564 565 566 567 568 569 570 571

    if (hasNull) {
      colDataAppendNULL(pOutputData, k);
      memset(output, 0, dataLen);
    } else {
      varDataSetLen(output, dataLen);
      colDataAppend(pOutputData, k, output, false);
      output += varDataTLen(output);
    }
G
Ganlin Zhao 已提交
572 573
  }

574
  pOutput->numOfRows = numOfRows;
575 576

DONE:
577
  taosMemoryFree(input);
578
  taosMemoryFree(outputBuf);
G
Ganlin Zhao 已提交
579 580
  taosMemoryFree(pInputData);

581
  return ret;
G
Ganlin Zhao 已提交
582 583
}

584
static int32_t doCaseConvFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput, _conv_fn convFn) {
G
Ganlin Zhao 已提交
585 586 587 588 589
  int32_t type = GET_PARAM_TYPE(pInput);

  SColumnInfoData *pInputData  = pInput->columnData;
  SColumnInfoData *pOutputData = pOutput->columnData;

590 591
  int32_t outputLen = pInputData->varmeta.length;
  char *outputBuf = taosMemoryCalloc(outputLen, 1);
592
  char *output = outputBuf;
593

G
Ganlin Zhao 已提交
594
  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
595 596
    if (colDataIsNull_s(pInputData, i)) {
      colDataAppendNULL(pOutputData, i);
G
Ganlin Zhao 已提交
597 598 599
      continue;
    }

600
    char *input = colDataGetData(pInput[0].columnData, i);
601
    int32_t len = varDataLen(input);
G
Ganlin Zhao 已提交
602 603
    if (type == TSDB_DATA_TYPE_VARCHAR) {
      for (int32_t j = 0; j < len; ++j) {
604
        *(varDataVal(output) + j) = convFn(*(varDataVal(input) + j));
G
Ganlin Zhao 已提交
605 606 607
      }
    } else { //NCHAR
      for (int32_t j = 0; j < len / TSDB_NCHAR_SIZE; ++j) {
608
        *((uint32_t *)varDataVal(output) + j) = convFn(*((uint32_t *)varDataVal(input) + j));
G
Ganlin Zhao 已提交
609 610
      }
    }
611
    varDataSetLen(output, len);
612
    colDataAppend(pOutputData, i, output, false);
613
    output += varDataTLen(output);
G
Ganlin Zhao 已提交
614 615 616
  }

  pOutput->numOfRows = pInput->numOfRows;
617
  taosMemoryFree(outputBuf);
G
Ganlin Zhao 已提交
618 619 620 621 622

  return TSDB_CODE_SUCCESS;
}


623
static int32_t doTrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput, _trim_fn trimFn) {
G
Ganlin Zhao 已提交
624 625 626 627 628
  int32_t type = GET_PARAM_TYPE(pInput);

  SColumnInfoData *pInputData  = pInput->columnData;
  SColumnInfoData *pOutputData = pOutput->columnData;

629 630
  int32_t outputLen = pInputData->varmeta.length;
  char *outputBuf = taosMemoryCalloc(outputLen, 1);
631
  char *output = outputBuf;
632

G
Ganlin Zhao 已提交
633
  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
634 635
    if (colDataIsNull_s(pInputData, i)) {
      colDataAppendNULL(pOutputData, i);
G
Ganlin Zhao 已提交
636 637 638
      continue;
    }

639
    char *input = colDataGetData(pInputData, i);
640
    int32_t len = varDataLen(input);
G
Ganlin Zhao 已提交
641
    int32_t charLen = (type == TSDB_DATA_TYPE_VARCHAR) ? len : len / TSDB_NCHAR_SIZE;
642 643 644
    trimFn(input, output, type, charLen);

    varDataSetLen(output, len);
645
    colDataAppend(pOutputData, i, output, false);
646
    output += varDataTLen(output);
G
Ganlin Zhao 已提交
647 648 649
  }

  pOutput->numOfRows = pInput->numOfRows;
650
  taosMemoryFree(outputBuf);
G
Ganlin Zhao 已提交
651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667

  return TSDB_CODE_SUCCESS;
}

int32_t substrFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  int32_t subPos = 0;
  GET_TYPED_DATA(subPos, int32_t, GET_PARAM_TYPE(&pInput[1]), pInput[1].columnData->pData);

  int32_t subLen = INT16_MAX;
  if (inputNum == 3) {
    GET_TYPED_DATA(subLen, int32_t, GET_PARAM_TYPE(&pInput[2]), pInput[2].columnData->pData);
    subLen = (GET_PARAM_TYPE(pInput) == TSDB_DATA_TYPE_VARCHAR) ? subLen : subLen * TSDB_NCHAR_SIZE;
  }

  SColumnInfoData *pInputData  = pInput->columnData;
  SColumnInfoData *pOutputData = pOutput->columnData;

668
  int32_t outputLen = pInputData->varmeta.length * pInput->numOfRows;
669
  char *outputBuf = taosMemoryCalloc(outputLen, 1);
670
  char *output = outputBuf;
671 672

  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
673 674
    if (colDataIsNull_s(pInputData, i)) {
      colDataAppendNULL(pOutputData, i);
G
Ganlin Zhao 已提交
675 676
      continue;
    }
677
    char *input = colDataGetData(pInput[0].columnData, i);
678
    int32_t len = varDataLen(input);
G
Ganlin Zhao 已提交
679 680 681 682
    int32_t startPosBytes;

    if (subPos > 0) {
      startPosBytes = (GET_PARAM_TYPE(pInput) == TSDB_DATA_TYPE_VARCHAR) ? subPos - 1 : (subPos - 1) * TSDB_NCHAR_SIZE;
wafwerar's avatar
wafwerar 已提交
683
      startPosBytes = TMIN(startPosBytes, len);
G
Ganlin Zhao 已提交
684 685
    } else {
      startPosBytes = (GET_PARAM_TYPE(pInput) == TSDB_DATA_TYPE_VARCHAR) ? len + subPos : len + subPos * TSDB_NCHAR_SIZE;
wafwerar's avatar
wafwerar 已提交
686
      startPosBytes = TMAX(startPosBytes, 0);
G
Ganlin Zhao 已提交
687 688
    }

wafwerar's avatar
wafwerar 已提交
689
    int32_t resLen = TMIN(subLen, len - startPosBytes);
690 691
    if (resLen > 0) {
      memcpy(varDataVal(output), varDataVal(input) + startPosBytes, resLen);
G
Ganlin Zhao 已提交
692 693
    }

694
    varDataSetLen(output, resLen);
695
    colDataAppend(pOutputData, i , output, false);
696
    output += varDataTLen(output);
G
Ganlin Zhao 已提交
697 698 699
  }

  pOutput->numOfRows = pInput->numOfRows;
700
  taosMemoryFree(outputBuf);
G
Ganlin Zhao 已提交
701 702 703 704

  return TSDB_CODE_SUCCESS;
}

G
Ganlin Zhao 已提交
705
int32_t castFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
706
  int16_t inputType  = GET_PARAM_TYPE(&pInput[0]);
707
  int16_t inputLen   = GET_PARAM_BYTES(&pInput[0]);
708 709
  int16_t outputType = GET_PARAM_TYPE(&pOutput[0]);
  int64_t outputLen  = GET_PARAM_BYTES(&pOutput[0]);
G
Ganlin Zhao 已提交
710

wafwerar's avatar
wafwerar 已提交
711
  char *outputBuf = taosMemoryCalloc(outputLen * pInput[0].numOfRows + 1, 1);
G
Ganlin Zhao 已提交
712 713 714 715 716 717 718
  char *output = outputBuf;

  for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
    if (colDataIsNull_s(pInput[0].columnData, i)) {
      colDataAppendNULL(pOutput->columnData, i);
      continue;
    }
719

720
    char *input = colDataGetData(pInput[0].columnData, i);
G
Ganlin Zhao 已提交
721 722

    switch(outputType) {
723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776
      case TSDB_DATA_TYPE_TINYINT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(int8_t *)output = taosStr2Int8(varDataVal(input), NULL, 10);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len  = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(int8_t *)output = taosStr2Int8(newBuf, NULL, 10);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(int8_t *)output, int8_t, inputType, input);
        }
        break;
      }
      case TSDB_DATA_TYPE_SMALLINT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(int16_t *)output = taosStr2Int16(varDataVal(input), NULL, 10);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len  = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(int16_t *)output = taosStr2Int16(newBuf, NULL, 10);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(int16_t *)output, int16_t, inputType, input);
        }
        break;
      }
      case TSDB_DATA_TYPE_INT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(int32_t *)output = taosStr2Int32(varDataVal(input), NULL, 10);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len  = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(int32_t *)output = taosStr2Int32(newBuf, NULL, 10);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(int32_t *)output, int32_t, inputType, input);
        }
        break;
      }
G
Ganlin Zhao 已提交
777 778
      case TSDB_DATA_TYPE_BIGINT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
779
          *(int64_t *)output = taosStr2Int64(varDataVal(input), NULL, 10);
G
Ganlin Zhao 已提交
780
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
781
          char *newBuf = taosMemoryCalloc(1, inputLen);
G
Ganlin Zhao 已提交
782 783 784 785 786 787
          int32_t len  = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
wafwerar's avatar
wafwerar 已提交
788
          *(int64_t *)output = taosStr2Int64(newBuf, NULL, 10);
G
Ganlin Zhao 已提交
789 790 791 792 793 794
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(int64_t *)output, int64_t, inputType, input);
        }
        break;
      }
795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 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
      case TSDB_DATA_TYPE_UTINYINT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(uint8_t *)output = taosStr2UInt8(varDataVal(input), NULL, 10);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(uint8_t *)output = taosStr2UInt8(newBuf, NULL, 10);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(uint8_t *)output, uint8_t, inputType, input);
        }
        break;
      }
      case TSDB_DATA_TYPE_USMALLINT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(uint16_t *)output = taosStr2UInt16(varDataVal(input), NULL, 10);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(uint16_t *)output = taosStr2UInt16(newBuf, NULL, 10);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(uint16_t *)output, uint16_t, inputType, input);
        }
        break;
      }
      case TSDB_DATA_TYPE_UINT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(uint32_t *)output = taosStr2UInt32(varDataVal(input), NULL, 10);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(uint32_t *)output = taosStr2UInt32(newBuf, NULL, 10);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(uint32_t *)output, uint32_t, inputType, input);
        }
        break;
      }
G
Ganlin Zhao 已提交
849 850
      case TSDB_DATA_TYPE_UBIGINT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
851
          *(uint64_t *)output = taosStr2UInt64(varDataVal(input), NULL, 10);
G
Ganlin Zhao 已提交
852
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
853
          char *newBuf = taosMemoryCalloc(1, inputLen);
G
Ganlin Zhao 已提交
854 855 856 857 858 859
          int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
wafwerar's avatar
wafwerar 已提交
860
          *(uint64_t *)output = taosStr2UInt64(newBuf, NULL, 10);
G
Ganlin Zhao 已提交
861 862 863 864 865 866
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(uint64_t *)output, uint64_t, inputType, input);
        }
        break;
      }
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
      case TSDB_DATA_TYPE_FLOAT: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(float *)output = taosStr2Float(varDataVal(input), NULL);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len  = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(float *)output = taosStr2Float(newBuf, NULL);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(float *)output, float, inputType, input);
        }
        break;
      }
      case TSDB_DATA_TYPE_DOUBLE: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(double *)output = taosStr2Double(varDataVal(input), NULL);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len  = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(double *)output = taosStr2Double(newBuf, NULL);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(double *)output, double, inputType, input);
        }
        break;
      }
G
Ganlin Zhao 已提交
903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920
      case TSDB_DATA_TYPE_BOOL: {
        if (inputType == TSDB_DATA_TYPE_BINARY) {
          *(bool *)output = taosStr2Int8(varDataVal(input), NULL, 10);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len  = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          newBuf[len] = 0;
          *(bool *)output = taosStr2Int8(newBuf, NULL, 10);
          taosMemoryFree(newBuf);
        } else {
          GET_TYPED_DATA(*(bool *)output, bool, inputType, input);
        }
        break;
      }
G
Ganlin Zhao 已提交
921
      case TSDB_DATA_TYPE_TIMESTAMP: {
922
        int64_t timeVal;
G
Ganlin Zhao 已提交
923
        if (inputType == TSDB_DATA_TYPE_BINARY || inputType == TSDB_DATA_TYPE_NCHAR) {
924 925 926
          int64_t timePrec;
          GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[1]), pInput[1].columnData->pData);
          int32_t ret = convertStringToTimestamp(inputType, input, timePrec, &timeVal);
927 928 929 930 931
          if (ret != TSDB_CODE_SUCCESS) {
            *(int64_t *)output = 0;
          } else {
            *(int64_t *)output = timeVal;
          }
G
Ganlin Zhao 已提交
932 933 934 935 936 937 938 939 940 941
        } else {
          GET_TYPED_DATA(*(int64_t *)output, int64_t, inputType, input);
        }
        break;
      }
      case TSDB_DATA_TYPE_BINARY: {
        if (inputType == TSDB_DATA_TYPE_BOOL) {
          int32_t len = sprintf(varDataVal(output), "%.*s", (int32_t)(outputLen - VARSTR_HEADER_SIZE), *(int8_t *)input ? "true" : "false");
          varDataSetLen(output, len);
        } else if (inputType == TSDB_DATA_TYPE_BINARY) {
wafwerar's avatar
wafwerar 已提交
942
          int32_t len = TMIN(varDataLen(input), outputLen - VARSTR_HEADER_SIZE);
943
          len = sprintf(varDataVal(output), "%.*s", len, varDataVal(input));
G
Ganlin Zhao 已提交
944
          varDataSetLen(output, len);
945
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
946 947 948 949 950 951 952 953 954 955
          char *newBuf = taosMemoryCalloc(1, inputLen);
          int32_t len = taosUcs4ToMbs((TdUcs4 *)varDataVal(input), varDataLen(input), newBuf);
          if (len < 0) {
            taosMemoryFree(newBuf);
            return TSDB_CODE_FAILED;
          }
          len = TMIN(len, outputLen - VARSTR_HEADER_SIZE);
          memcpy(varDataVal(output), newBuf, len);
          varDataSetLen(output, len);
          taosMemoryFree(newBuf);
G
Ganlin Zhao 已提交
956 957 958 959 960 961 962 963 964 965 966 967
        } else {
          char tmp[400] = {0};
          NUM_TO_STRING(inputType, input, sizeof(tmp), tmp);
          int32_t len = (int32_t)strlen(tmp);
          len = (outputLen - VARSTR_HEADER_SIZE) > len ? len : (outputLen - VARSTR_HEADER_SIZE);
          memcpy(varDataVal(output), tmp, len);
          varDataSetLen(output, len);
        }
        break;
      }
      case TSDB_DATA_TYPE_NCHAR: {
        int32_t outputCharLen = (outputLen - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
968
        int32_t len;
G
Ganlin Zhao 已提交
969 970
        if (inputType == TSDB_DATA_TYPE_BOOL) {
          char tmp[8] = {0};
971
          len = sprintf(tmp, "%.*s", outputCharLen, *(int8_t *)input ? "true" : "false" );
G
Ganlin Zhao 已提交
972 973 974 975 976 977
          bool ret = taosMbsToUcs4(tmp, len, (TdUcs4 *)varDataVal(output), outputLen - VARSTR_HEADER_SIZE, &len);
          if (!ret) {
            return TSDB_CODE_FAILED;
          }
          varDataSetLen(output, len);
        } else if (inputType == TSDB_DATA_TYPE_BINARY) {
978
          len = outputCharLen > varDataLen(input) ? varDataLen(input) : outputCharLen;
G
Ganlin Zhao 已提交
979 980 981 982 983 984
          bool ret = taosMbsToUcs4(input + VARSTR_HEADER_SIZE, len, (TdUcs4 *)varDataVal(output), outputLen - VARSTR_HEADER_SIZE, &len);
          if (!ret) {
            return TSDB_CODE_FAILED;
          }
          varDataSetLen(output, len);
        } else if (inputType == TSDB_DATA_TYPE_NCHAR) {
985 986 987
          len = TMIN(outputLen - VARSTR_HEADER_SIZE, varDataLen(input));
          memcpy(output, input, len + VARSTR_HEADER_SIZE);
          varDataSetLen(output, len);
G
Ganlin Zhao 已提交
988 989 990
        } else {
          char tmp[400] = {0};
          NUM_TO_STRING(inputType, input, sizeof(tmp), tmp);
991
          len = (int32_t)strlen(tmp);
G
Ganlin Zhao 已提交
992 993 994 995 996 997 998
          len = outputCharLen > len ? len : outputCharLen;
          bool ret = taosMbsToUcs4(tmp, len, (TdUcs4 *)varDataVal(output), outputLen - VARSTR_HEADER_SIZE, &len);
          if (!ret) {
            return TSDB_CODE_FAILED;
          }
          varDataSetLen(output, len);
        }
999
        //for constant conversion, need to set proper length of pOutput description
G
Ganlin Zhao 已提交
1000
        if (len < outputLen) {
D
dapan1121 已提交
1001
          pOutput->columnData->info.bytes = len + VARSTR_HEADER_SIZE;
1002
        }
G
Ganlin Zhao 已提交
1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021
        break;
      }
      default: {
        return TSDB_CODE_FAILED;
      }
    }

    colDataAppend(pOutput->columnData, i, output, false);
    if (IS_VAR_DATA_TYPE(outputType)) {
      output += varDataTLen(output);
    } else {
      output += tDataTypes[outputType].bytes;
    }
  }

  pOutput->numOfRows = pInput->numOfRows;
  taosMemoryFree(outputBuf);
  return TSDB_CODE_SUCCESS;
}
G
Ganlin Zhao 已提交
1022

1023 1024 1025
int32_t toISO8601Function(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  int32_t type = GET_PARAM_TYPE(pInput);

1026
  bool tzPresent = (inputNum == 2) ? true : false;
1027 1028
  char* tz;
  int32_t tzLen;
1029 1030 1031 1032
  if (tzPresent) {
    tz = varDataVal(pInput[1].columnData->pData);
    tzLen = varDataLen(pInput[1].columnData->pData);
  }
1033

1034 1035 1036 1037 1038 1039
  for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
    if (colDataIsNull_s(pInput[0].columnData, i)) {
      colDataAppendNULL(pOutput->columnData, i);
      continue;
    }

1040
    char *input = colDataGetData(pInput[0].columnData, i);
1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056
    char fraction[20] = {0};
    bool hasFraction = false;
    NUM_TO_STRING(type, input, sizeof(fraction), fraction);
    int32_t tsDigits = (int32_t)strlen(fraction);

    char buf[64] = {0};
    int64_t timeVal;
    GET_TYPED_DATA(timeVal, int64_t, type, input);
    if (tsDigits > TSDB_TIME_PRECISION_SEC_DIGITS) {
      if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
        timeVal = timeVal / 1000;
      } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
        timeVal = timeVal / (1000 * 1000);
      } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
        timeVal = timeVal / (1000 * 1000 * 1000);
      } else {
1057 1058
        colDataAppendNULL(pOutput->columnData, i);
        continue;
1059 1060 1061 1062 1063
      }
      hasFraction = true;
      memmove(fraction, fraction + TSDB_TIME_PRECISION_SEC_DIGITS, TSDB_TIME_PRECISION_SEC_DIGITS);
    }

1064
    struct tm *tmInfo = taosLocalTime((const time_t *)&timeVal, NULL);
1065
    strftime(buf, sizeof(buf), "%Y-%m-%dT%H:%M:%S", tmInfo);
1066 1067
    int32_t len = (int32_t)strlen(buf);

1068 1069 1070 1071
    //add timezone string
    snprintf(buf + len, tzLen + 1, "%s", tz);
    len += tzLen;

1072 1073
    if (hasFraction) {
      int32_t fracLen = (int32_t)strlen(fraction) + 1;
1074 1075 1076 1077

      char *tzInfo;
      if (buf[len - 1] == 'z' || buf[len - 1] == 'Z') {
        tzInfo = &buf[len - 1];
1078 1079
        memmove(tzInfo + fracLen, tzInfo, strlen(tzInfo));
      } else {
1080 1081 1082 1083 1084 1085 1086 1087 1088 1089
        tzInfo = strchr(buf, '+');
        if (tzInfo) {
          memmove(tzInfo + fracLen, tzInfo, strlen(tzInfo));
        } else {
          //search '-' backwards
          tzInfo = strrchr(buf, '-');
          if (tzInfo) {
            memmove(tzInfo + fracLen, tzInfo, strlen(tzInfo));
          }
        }
1090 1091
      }

1092
      char tmp[32] = {0};
1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108
      sprintf(tmp, ".%s", fraction);
      memcpy(tzInfo, tmp, fracLen);
      len += fracLen;
    }

    memmove(buf + VARSTR_HEADER_SIZE, buf, len);
    varDataSetLen(buf, len);

    colDataAppend(pOutput->columnData, i, buf, false);
  }

  pOutput->numOfRows = pInput->numOfRows;

  return TSDB_CODE_SUCCESS;
}

1109 1110
int32_t toUnixtimestampFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  int32_t type = GET_PARAM_TYPE(pInput);
1111 1112
  int64_t timePrec;
  GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[1]), pInput[1].columnData->pData);
1113 1114 1115 1116 1117 1118

  for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
    if (colDataIsNull_s(pInput[0].columnData, i)) {
      colDataAppendNULL(pOutput->columnData, i);
      continue;
    }
1119
    char *input = colDataGetData(pInput[0].columnData, i);
1120 1121

    int64_t timeVal = 0;
1122 1123 1124
    int32_t ret = convertStringToTimestamp(type, input, timePrec, &timeVal);
    if (ret != TSDB_CODE_SUCCESS) {
      colDataAppendNULL(pOutput->columnData, i);
1125 1126
    } else {
      colDataAppend(pOutput->columnData, i, (char *)&timeVal, false);
1127
    }
1128 1129 1130 1131 1132 1133 1134
  }

  pOutput->numOfRows = pInput->numOfRows;

  return TSDB_CODE_SUCCESS;
}

wmmhello's avatar
wmmhello 已提交
1135 1136 1137
int32_t toJsonFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  int32_t type = GET_PARAM_TYPE(pInput);

wmmhello's avatar
wmmhello 已提交
1138
  char tmp[TSDB_MAX_JSON_TAG_LEN] = {0};
wmmhello's avatar
wmmhello 已提交
1139
  for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
wmmhello's avatar
wmmhello 已提交
1140 1141
    SArray* pTagVals = taosArrayInit(8, sizeof(STagVal));
    STag*   pTag = NULL;
wmmhello's avatar
wmmhello 已提交
1142

wmmhello's avatar
wmmhello 已提交
1143 1144
    if (colDataIsNull_s(pInput[0].columnData, i)) {
      tTagNew(pTagVals, 1, true, &pTag);
wmmhello's avatar
wmmhello 已提交
1145
    }else{
wmmhello's avatar
wmmhello 已提交
1146
      char *input = pInput[0].columnData->pData + pInput[0].columnData->varmeta.offset[i];
wmmhello's avatar
wmmhello 已提交
1147
      if (varDataLen(input) > (TSDB_MAX_JSON_TAG_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE){
wmmhello's avatar
wmmhello 已提交
1148 1149
        taosArrayDestroy(pTagVals);
        return TSDB_CODE_FAILED;
wmmhello's avatar
wmmhello 已提交
1150 1151
      }
      memcpy(tmp, varDataVal(input), varDataLen(input));
wmmhello's avatar
wmmhello 已提交
1152
      tmp[varDataLen(input)] = 0;
wmmhello's avatar
wmmhello 已提交
1153 1154 1155
      if(parseJsontoTagData(tmp, pTagVals, &pTag, NULL)){
        tTagNew(pTagVals, 1, true, &pTag);
      }
wmmhello's avatar
wmmhello 已提交
1156 1157
    }

wmmhello's avatar
wmmhello 已提交
1158 1159 1160
    colDataAppend(pOutput->columnData, i, (const char*)pTag, false);
    tTagFree(pTag);
    taosArrayDestroy(pTagVals);
wmmhello's avatar
wmmhello 已提交
1161 1162 1163 1164 1165 1166
  }

  pOutput->numOfRows = pInput->numOfRows;
  return TSDB_CODE_SUCCESS;
}

1167 1168 1169
int32_t timeTruncateFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  int32_t type = GET_PARAM_TYPE(&pInput[0]);

1170
  int64_t timeUnit, timePrec, timeVal = 0;
1171
  GET_TYPED_DATA(timeUnit, int64_t, GET_PARAM_TYPE(&pInput[1]), pInput[1].columnData->pData);
1172
  GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[2]), pInput[2].columnData->pData);
1173 1174 1175 1176

  int64_t factor = (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 :
                   (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);

1177 1178
  timeUnit = timeUnit * 1000 / factor;

1179 1180 1181 1182 1183 1184
  for (int32_t i = 0; i < pInput[0].numOfRows; ++i) {
    if (colDataIsNull_s(pInput[0].columnData, i)) {
      colDataAppendNULL(pOutput->columnData, i);
      continue;
    }

1185 1186
    char *input = colDataGetData(pInput[0].columnData, i);

1187
    if (IS_VAR_DATA_TYPE(type)) { /* datetime format strings */
1188 1189 1190 1191 1192
      int32_t ret = convertStringToTimestamp(type, input, TSDB_TIME_PRECISION_NANO, &timeVal);
      if (ret != TSDB_CODE_SUCCESS) {
        colDataAppendNULL(pOutput->columnData, i);
        continue;
      }
1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234
      //If converted value is less than 10digits in second, use value in second instead
      int64_t timeValSec = timeVal / 1000000000;
      if (timeValSec < 1000000000) {
        timeVal = timeValSec;
      }
    } else if (type == TSDB_DATA_TYPE_BIGINT) { /* unix timestamp */
      GET_TYPED_DATA(timeVal, int64_t, type, input);
    } else if (type == TSDB_DATA_TYPE_TIMESTAMP) { /* timestamp column*/
      GET_TYPED_DATA(timeVal, int64_t, type, input);
      int64_t timeValSec = timeVal / factor;
      if (timeValSec < 1000000000) {
        timeVal = timeValSec;
      }
    }

    char buf[20] = {0};
    NUM_TO_STRING(TSDB_DATA_TYPE_BIGINT, &timeVal, sizeof(buf), buf);
    int32_t tsDigits = (int32_t)strlen(buf);

    switch (timeUnit) {
      case 0: { /* 1u */
        if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000 * 1000;
        //} else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
        //  //timeVal = timeVal / 1000;
        } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
          timeVal = timeVal * factor;
        } else {
          timeVal = timeVal * 1;
        }
        break;
      }
      case 1: { /* 1a */
        if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal * 1;
        } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal / 1000 * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000000 * 1000000;
        } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS){
          timeVal = timeVal * factor;
        } else {
1235 1236
          colDataAppendNULL(pOutput->columnData, i);
          continue;
1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249
        }
        break;
      }
      case 1000: { /* 1s */
        if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal / 1000 * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal / 1000000 * 1000000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000000000 * 1000000000;
        } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
          timeVal = timeVal * factor;
        } else {
1250 1251
          colDataAppendNULL(pOutput->columnData, i);
          continue;
1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264
        }
        break;
      }
      case 60000: { /* 1m */
        if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal / 1000 / 60 * 60 * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal / 1000000 / 60 * 60 * 1000000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000000000 / 60 * 60 * 1000000000;
        } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
          timeVal = timeVal * factor / factor / 60 * 60 * factor;
        } else {
1265 1266
          colDataAppendNULL(pOutput->columnData, i);
          continue;
1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279
        }
        break;
      }
      case 3600000: { /* 1h */
        if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal / 1000 / 3600 * 3600 * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal / 1000000 / 3600 * 3600 * 1000000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000000000 / 3600 * 3600 * 1000000000;
        } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
          timeVal = timeVal * factor / factor / 3600 * 3600 * factor;
        } else {
1280 1281
          colDataAppendNULL(pOutput->columnData, i);
          continue;
1282 1283 1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294
        }
        break;
      }
      case 86400000: { /* 1d */
        if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal / 1000 / 86400 * 86400 * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal / 1000000 / 86400 * 86400 * 1000000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000000000 / 86400 * 86400 * 1000000000;
        } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
          timeVal = timeVal * factor / factor / 86400* 86400 * factor;
        } else {
1295 1296
          colDataAppendNULL(pOutput->columnData, i);
          continue;
1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309
        }
        break;
      }
      case 604800000: { /* 1w */
        if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal / 1000 / 604800 * 604800 * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal / 1000000 / 604800 * 604800 * 1000000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000000000 / 604800 * 604800 * 1000000000;
        } else if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
          timeVal = timeVal * factor / factor / 604800 * 604800* factor;
        } else {
1310 1311
          colDataAppendNULL(pOutput->columnData, i);
          continue;
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 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356
        }
        break;
      }
      default: {
        timeVal = timeVal * 1;
        break;
      }
    }

    //truncate the timestamp to db precision
    switch (timePrec) {
      case TSDB_TIME_PRECISION_MILLI: {
        if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal / 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000000;
        }
        break;
      }
      case TSDB_TIME_PRECISION_MICRO: {
        if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal = timeVal / 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal * 1000;
        }
        break;
      }
      case TSDB_TIME_PRECISION_NANO: {
        if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal = timeVal * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal = timeVal * 1000000;
        }
        break;
      }
    }

    colDataAppend(pOutput->columnData, i, (char *)&timeVal, false);
  }

  pOutput->numOfRows = pInput->numOfRows;

  return TSDB_CODE_SUCCESS;
}

G
Ganlin Zhao 已提交
1357
int32_t timeDiffFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
1358 1359
  int64_t timeUnit = -1, timePrec, timeVal[2] = {0};
  if (inputNum == 4) {
G
Ganlin Zhao 已提交
1360
    GET_TYPED_DATA(timeUnit, int64_t, GET_PARAM_TYPE(&pInput[2]), pInput[2].columnData->pData);
1361 1362 1363
    GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[3]), pInput[3].columnData->pData);
  } else {
    GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[2]), pInput[2].columnData->pData);
G
Ganlin Zhao 已提交
1364 1365
  }

1366 1367 1368 1369 1370
  int64_t factor = (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 :
                   (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);

  timeUnit = timeUnit * 1000 / factor;

G
Ganlin Zhao 已提交
1371 1372 1373 1374
  int32_t numOfRows = 0;
  for (int32_t i = 0; i < inputNum; ++i) {
    if (pInput[i].numOfRows > numOfRows) {
      numOfRows = pInput[i].numOfRows;
G
Ganlin Zhao 已提交
1375 1376 1377
    }
  }

G
Ganlin Zhao 已提交
1378 1379 1380
  char *input[2];
  for (int32_t i = 0; i < numOfRows; ++i) {
    bool hasNull = false;
G
Ganlin Zhao 已提交
1381
    for (int32_t k = 0; k < 2; ++k) {
G
Ganlin Zhao 已提交
1382 1383 1384
      if (colDataIsNull_s(pInput[k].columnData, i)) {
        hasNull = true;
        break;
G
Ganlin Zhao 已提交
1385 1386
      }

1387 1388 1389
      int32_t rowIdx = (pInput[k].numOfRows == 1) ? 0 : i;
      input[k] = colDataGetData(pInput[k].columnData, rowIdx);

G
Ganlin Zhao 已提交
1390 1391
      int32_t type = GET_PARAM_TYPE(&pInput[k]);
      if (IS_VAR_DATA_TYPE(type)) { /* datetime format strings */
1392 1393
        int32_t ret = convertStringToTimestamp(type, input[k], TSDB_TIME_PRECISION_NANO, &timeVal[k]);
        if (ret != TSDB_CODE_SUCCESS) {
1394 1395
          hasNull = true;
          break;
1396
        }
G
Ganlin Zhao 已提交
1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418
      } else if (type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_TIMESTAMP) { /* unix timestamp or ts column*/
        GET_TYPED_DATA(timeVal[k], int64_t, type, input[k]);
        if (type == TSDB_DATA_TYPE_TIMESTAMP) {
          int64_t factor = (timePrec == TSDB_TIME_PRECISION_MILLI) ? 1000 :
                           (timePrec == TSDB_TIME_PRECISION_MICRO ? 1000000 : 1000000000);
          int64_t timeValSec = timeVal[k] / factor;
          if (timeValSec < 1000000000) {
            timeVal[k] = timeValSec;
          }
        }

        char buf[20] = {0};
        NUM_TO_STRING(TSDB_DATA_TYPE_BIGINT, &timeVal[k], sizeof(buf), buf);
        int32_t tsDigits = (int32_t)strlen(buf);
        if (tsDigits <= TSDB_TIME_PRECISION_SEC_DIGITS) {
          timeVal[k] = timeVal[k] * 1000000000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MILLI_DIGITS) {
          timeVal[k] = timeVal[k] * 1000000;
        } else if (tsDigits == TSDB_TIME_PRECISION_MICRO_DIGITS) {
          timeVal[k] = timeVal[k] * 1000;
        } else if (tsDigits == TSDB_TIME_PRECISION_NANO_DIGITS) {
          timeVal[k] = timeVal[k];
G
Ganlin Zhao 已提交
1419
        } else {
1420 1421
          hasNull = true;
          break;
G
Ganlin Zhao 已提交
1422
        }
G
Ganlin Zhao 已提交
1423 1424 1425
      }
    }

G
Ganlin Zhao 已提交
1426 1427 1428 1429 1430
    if (hasNull) {
      colDataAppendNULL(pOutput->columnData, i);
      continue;
    }

G
Ganlin Zhao 已提交
1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487
    int64_t result = (timeVal[0] >= timeVal[1]) ? (timeVal[0] - timeVal[1]) :
                                                  (timeVal[1] - timeVal[0]);

    if (timeUnit < 0) { // if no time unit given use db precision
      switch(timePrec) {
        case TSDB_TIME_PRECISION_MILLI: {
          result = result / 1000000;
          break;
        }
        case TSDB_TIME_PRECISION_MICRO: {
          result = result / 1000;
          break;
        }
        case TSDB_TIME_PRECISION_NANO: {
          result = result / 1;
          break;
        }
      }
    } else {
      switch(timeUnit) {
        case 0: { /* 1u */
          result = result / 1000;
          break;
        }
        case 1: { /* 1a */
          result = result / 1000000;
          break;
        }
        case 1000: { /* 1s */
          result = result / 1000000000;
          break;
        }
        case 60000: { /* 1m */
          result = result / 1000000000 / 60;
          break;
        }
        case 3600000: { /* 1h */
          result = result / 1000000000 / 3600;
          break;
        }
        case 86400000: { /* 1d */
          result = result / 1000000000 / 86400;
          break;
        }
        case 604800000: { /* 1w */
          result = result / 1000000000 / 604800;
          break;
        }
        default: {
          break;
        }
      }
    }

    colDataAppend(pOutput->columnData, i, (char *)&result, false);
  }

G
Ganlin Zhao 已提交
1488
  pOutput->numOfRows = numOfRows;
G
Ganlin Zhao 已提交
1489 1490 1491 1492

  return TSDB_CODE_SUCCESS;
}

1493
int32_t nowFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
1494 1495 1496 1497
  int64_t timePrec;
  GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[0]), pInput[0].columnData->pData);

  int64_t ts = taosGetTimestamp(timePrec);
X
Xiaoyu Wang 已提交
1498 1499
  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
    colDataAppendInt64(pOutput->columnData, i, &ts);
1500
  }
X
Xiaoyu Wang 已提交
1501
  pOutput->numOfRows = pInput->numOfRows;
1502 1503 1504 1505
  return TSDB_CODE_SUCCESS;
}

int32_t todayFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
1506
  int64_t timePrec;
1507
  GET_TYPED_DATA(timePrec, int64_t, GET_PARAM_TYPE(&pInput[0]), pInput[0].columnData->pData);
1508 1509

  int64_t ts = taosGetTimestampToday(timePrec);
X
Xiaoyu Wang 已提交
1510 1511
  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
    colDataAppendInt64(pOutput->columnData, i, &ts);
1512
  }
X
Xiaoyu Wang 已提交
1513
  pOutput->numOfRows = pInput->numOfRows;
1514 1515 1516
  return TSDB_CODE_SUCCESS;
}

1517
int32_t timezoneFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
1518 1519 1520
  char output[TD_TIMEZONE_LEN + VARSTR_HEADER_SIZE] = {0};
  memcpy(varDataVal(output), tsTimezoneStr, TD_TIMEZONE_LEN);
  varDataSetLen(output, strlen(tsTimezoneStr));
X
Xiaoyu Wang 已提交
1521
  for (int32_t i = 0; i < pInput->numOfRows; ++i) {
1522
    colDataAppend(pOutput->columnData, i, output, false);
1523
  }
X
Xiaoyu Wang 已提交
1524
  pOutput->numOfRows = pInput->numOfRows;
1525 1526 1527
  return TSDB_CODE_SUCCESS;
}

1528 1529 1530
int32_t atanFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique(pInput, inputNum, pOutput, atan);
}
1531

1532 1533 1534
int32_t sinFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique(pInput, inputNum, pOutput, sin);
}
1535

1536 1537 1538
int32_t cosFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique(pInput, inputNum, pOutput, cos);
}
1539

1540 1541 1542
int32_t tanFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique(pInput, inputNum, pOutput, tan);
}
1543

1544 1545 1546
int32_t asinFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique(pInput, inputNum, pOutput, asin);
}
1547

1548 1549 1550
int32_t acosFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique(pInput, inputNum, pOutput, acos);
}
H
Haojun Liao 已提交
1551

G
Ganlin Zhao 已提交
1552 1553 1554 1555 1556
int32_t powFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique2(pInput, inputNum, pOutput, pow);
}

int32_t logFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
1557 1558 1559 1560 1561
  if (inputNum == 1) {
    return doScalarFunctionUnique(pInput, inputNum, pOutput, tlog);
  } else {
    return doScalarFunctionUnique2(pInput, inputNum, pOutput, tlog2);
  }
G
Ganlin Zhao 已提交
1562 1563
}

1564 1565 1566
int32_t sqrtFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunctionUnique(pInput, inputNum, pOutput, sqrt);
}
1567

1568 1569 1570 1571 1572 1573 1574 1575 1576 1577
int32_t ceilFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunction(pInput, inputNum, pOutput, ceilf, ceil);
}

int32_t floorFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunction(pInput, inputNum, pOutput, floorf, floor);
}

int32_t roundFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doScalarFunction(pInput, inputNum, pOutput, roundf, round);
1578 1579
}

G
Ganlin Zhao 已提交
1580
int32_t lowerFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
wafwerar's avatar
wafwerar 已提交
1581 1582 1583
#ifdef WINDOWS
  return doCaseConvFunction(pInput, inputNum, pOutput, towlower);
#else
G
Ganlin Zhao 已提交
1584
  return doCaseConvFunction(pInput, inputNum, pOutput, tolower);
wafwerar's avatar
wafwerar 已提交
1585
#endif
1586 1587
}

G
Ganlin Zhao 已提交
1588
int32_t upperFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
wafwerar's avatar
wafwerar 已提交
1589 1590 1591
#ifdef WINDOWS
  return doCaseConvFunction(pInput, inputNum, pOutput, towupper);
#else
G
Ganlin Zhao 已提交
1592
  return doCaseConvFunction(pInput, inputNum, pOutput, toupper);
wafwerar's avatar
wafwerar 已提交
1593
#endif
1594 1595
}

G
Ganlin Zhao 已提交
1596 1597
int32_t ltrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doTrimFunction(pInput, inputNum, pOutput, tltrim);
1598 1599
}

G
Ganlin Zhao 已提交
1600 1601
int32_t rtrimFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doTrimFunction(pInput, inputNum, pOutput, trtrim);
1602 1603
}

G
Ganlin Zhao 已提交
1604 1605 1606
int32_t lengthFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doLengthFunction(pInput, inputNum, pOutput, tlength);
}
1607

G
Ganlin Zhao 已提交
1608 1609
int32_t charLengthFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  return doLengthFunction(pInput, inputNum, pOutput, tcharlength);
1610 1611
}

H
Haojun Liao 已提交
1612
#if 0
1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644 1645 1646 1647 1648 1649 1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664 1665 1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676
static void reverseCopy(char* dest, const char* src, int16_t type, int32_t numOfRows) {
  switch(type) {
    case TSDB_DATA_TYPE_TINYINT:
    case TSDB_DATA_TYPE_UTINYINT:{
      int8_t* p = (int8_t*) dest;
      int8_t* pSrc = (int8_t*) src;

      for(int32_t i = 0; i < numOfRows; ++i) {
        p[i] = pSrc[numOfRows - i - 1];
      }
      return;
    }

    case TSDB_DATA_TYPE_SMALLINT:
    case TSDB_DATA_TYPE_USMALLINT:{
      int16_t* p = (int16_t*) dest;
      int16_t* pSrc = (int16_t*) src;

      for(int32_t i = 0; i < numOfRows; ++i) {
        p[i] = pSrc[numOfRows - i - 1];
      }
      return;
    }
    case TSDB_DATA_TYPE_INT:
    case TSDB_DATA_TYPE_UINT: {
      int32_t* p = (int32_t*) dest;
      int32_t* pSrc = (int32_t*) src;

      for(int32_t i = 0; i < numOfRows; ++i) {
        p[i] = pSrc[numOfRows - i - 1];
      }
      return;
    }
    case TSDB_DATA_TYPE_BIGINT:
    case TSDB_DATA_TYPE_UBIGINT: {
      int64_t* p = (int64_t*) dest;
      int64_t* pSrc = (int64_t*) src;

      for(int32_t i = 0; i < numOfRows; ++i) {
        p[i] = pSrc[numOfRows - i - 1];
      }
      return;
    }
    case TSDB_DATA_TYPE_FLOAT: {
      float* p = (float*) dest;
      float* pSrc = (float*) src;

      for(int32_t i = 0; i < numOfRows; ++i) {
        p[i] = pSrc[numOfRows - i - 1];
      }
      return;
    }
    case TSDB_DATA_TYPE_DOUBLE: {
      double* p = (double*) dest;
      double* pSrc = (double*) src;

      for(int32_t i = 0; i < numOfRows; ++i) {
        p[i] = pSrc[numOfRows - i - 1];
      }
      return;
    }
    default: assert(0);
  }
}
H
Haojun Liao 已提交
1677
#endif
1678

1679 1680 1681 1682 1683 1684 1685 1686
bool getTimePseudoFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv) {
  pEnv->calcMemSize = sizeof(int64_t);
  return true;
}

int32_t qStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  ASSERT(inputNum == 1);
  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 0));
1687
  return TSDB_CODE_SUCCESS;
1688 1689 1690 1691 1692
}

int32_t qEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  ASSERT(inputNum == 1);
  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 1));
1693
  return TSDB_CODE_SUCCESS;
1694 1695 1696 1697 1698
}

int32_t winDurFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  ASSERT(inputNum == 1);
  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t *)colDataGetData(pInput->columnData, 2));
1699
  return TSDB_CODE_SUCCESS;
1700 1701 1702 1703 1704 1705 1706 1707 1708 1709 1710 1711
}

int32_t winStartTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  ASSERT(inputNum == 1);
  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t*) colDataGetData(pInput->columnData, 3));
  return TSDB_CODE_SUCCESS;
}

int32_t winEndTsFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  ASSERT(inputNum == 1);
  colDataAppendInt64(pOutput->columnData, pOutput->numOfRows, (int64_t*) colDataGetData(pInput->columnData, 4));
  return TSDB_CODE_SUCCESS;
1712
}
1713 1714 1715

int32_t qTbnameFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
  ASSERT(inputNum == 1);
1716 1717 1718 1719 1720 1721 1722 1723 1724 1725 1726 1727 1728 1729 1730 1731

  SMetaReader mr = {0};
  metaReaderInit(&mr, pInput->param, 0);

  uint64_t uid = *(uint64_t *)colDataGetData(pInput->columnData, 0);
  metaGetTableEntryByUid(&mr, uid);

  char str[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
  STR_TO_VARSTR(str, mr.me.name);
  metaReaderClear(&mr);

  for(int32_t i = 0; i < pInput->numOfRows; ++i) {
    colDataAppend(pOutput->columnData, pOutput->numOfRows + i, str, false);
  }

  pOutput->numOfRows += pInput->numOfRows;
1732 1733
  return TSDB_CODE_SUCCESS;
}