scalar.c 27.2 KB
Newer Older
D
dapan1121 已提交
1 2
#include "function.h"
#include "functionMgt.h"
H
Haojun Liao 已提交
3 4
#include "nodes.h"
#include "querynodes.h"
D
dapan1121 已提交
5
#include "sclInt.h"
H
Haojun Liao 已提交
6 7 8
#include "sclvector.h"
#include "tcommon.h"
#include "tdatablock.h"
9
#include "scalar.h"
S
slzhou 已提交
10
#include "tudf.h"
D
dapan1121 已提交
11

D
dapan1121 已提交
12
int32_t scalarGetOperatorParamNum(EOperatorType type) {
D
dapan1121 已提交
13
  if (OP_TYPE_IS_NULL == type || OP_TYPE_IS_NOT_NULL == type || OP_TYPE_IS_TRUE == type || OP_TYPE_IS_NOT_TRUE == type 
D
dapan1121 已提交
14 15
   || OP_TYPE_IS_FALSE == type || OP_TYPE_IS_NOT_FALSE == type || OP_TYPE_IS_UNKNOWN == type || OP_TYPE_IS_NOT_UNKNOWN == type
   || OP_TYPE_MINUS == type) {
D
dapan1121 已提交
16 17 18 19 20 21
    return 1;
  }

  return 2;
}

22
SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows) {
H
Haojun Liao 已提交
23
  SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
24 25 26 27 28 29 30 31 32 33
  if (pColumnData == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  pColumnData->info.type      = pType->type;
  pColumnData->info.bytes     = pType->bytes;
  pColumnData->info.scale     = pType->scale;
  pColumnData->info.precision = pType->precision;

34
  int32_t code = colInfoDataEnsureCapacity(pColumnData, 0, numOfRows);
35 36
  if (code != TSDB_CODE_SUCCESS) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
37
    taosMemoryFree(pColumnData);
38 39 40 41 42 43
    return NULL;
  } else {
    return pColumnData;
  }
}

44 45 46 47
int32_t doConvertDataType(SValueNode* pValueNode, SScalarParam* out) {
  SScalarParam in = {.numOfRows = 1};
  in.columnData = createColumnInfoData(&pValueNode->node.resType, 1);
  colDataAppend(in.columnData, 0, nodesGetValueFromNode(pValueNode), false);
48

49
  colInfoDataEnsureCapacity(out->columnData, 0, 1);
50 51 52 53
  int32_t code = vectorConvertImpl(&in, out);
  sclFreeParam(&in);

  return code;
54 55
}

D
dapan1121 已提交
56 57 58 59 60 61 62 63 64 65 66 67
int32_t scalarGenerateSetFromList(void **data, void *pNode, uint32_t type) {
  SHashObj *pObj = taosHashInit(256, taosGetDefaultHashFunction(type), true, false);
  if (NULL == pObj) {
    sclError("taosHashInit failed, size:%d", 256);
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

  taosHashSetEqualFp(pObj, taosGetDefaultEqualFunction(type)); 

  int32_t code = 0;
  SNodeListNode *nodeList = (SNodeListNode *)pNode;
  SListCell *cell = nodeList->pNodeList->pHead;
H
Haojun Liao 已提交
68
  SScalarParam out = {.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData))};
69

D
dapan1121 已提交
70 71 72 73 74
  int32_t len = 0;
  void *buf = NULL;
  
  for (int32_t i = 0; i < nodeList->pNodeList->length; ++i) {
    SValueNode *valueNode = (SValueNode *)cell->pNode;
D
dapan1121 已提交
75
    
D
dapan1121 已提交
76
    if (valueNode->node.resType.type != type) {
77
      out.columnData->info.type = type;
D
dapan1121 已提交
78 79 80 81 82 83 84 85 86
      if (IS_VAR_DATA_TYPE(type)) {
        if (IS_VAR_DATA_TYPE(valueNode->node.resType.type)) {
          out.columnData->info.bytes = valueNode->node.resType.bytes * TSDB_NCHAR_SIZE;
        } else {
          out.columnData->info.bytes = 64 * TSDB_NCHAR_SIZE;
        }
      } else {
        out.columnData->info.bytes = tDataTypes[type].bytes;
      }
87

88 89
      code = doConvertDataType(valueNode, &out);
      if (code != TSDB_CODE_SUCCESS) {
90
//        sclError("convert data from %d to %d failed", in.type, out.type);
D
dapan1121 已提交
91 92 93 94
        SCL_ERR_JRET(code);
      }

      if (IS_VAR_DATA_TYPE(type)) {
95 96
        len = varDataLen(out.columnData->pData);
        buf = varDataVal(out.columnData->pData);
D
dapan1121 已提交
97 98
      } else {
        len = tDataTypes[type].bytes;
99
        buf = out.columnData->pData;
D
dapan1121 已提交
100 101 102
      }
    } else {
      buf = nodesGetValueFromNode(valueNode);
D
dapan1121 已提交
103 104 105 106 107
      if (IS_VAR_DATA_TYPE(type)) {
        len = varDataLen(buf);
        buf = varDataVal(buf);
      } else {
        len = valueNode->node.resType.bytes;
108
      }
D
dapan1121 已提交
109 110
    }
    
111
    if (taosHashPut(pObj, buf, (size_t)len, NULL, 0)) {
D
dapan1121 已提交
112 113 114
      sclError("taosHashPut failed");
      SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
D
dapan1121 已提交
115 116

    cell = cell->pNext;
D
dapan1121 已提交
117 118 119 120 121 122 123 124 125 126
  }

  *data = pObj;
  return TSDB_CODE_SUCCESS;

_return:
  taosHashCleanup(pObj);
  SCL_RET(code);
}

D
dapan1121 已提交
127 128 129 130 131 132 133
void sclFreeRes(SHashObj *res) {
  SScalarParam *p = NULL;
  void *pIter = taosHashIterate(res, NULL);
  while (pIter) {
    p = (SScalarParam *)pIter;

    if (p) {
D
dapan 已提交
134
      sclFreeParam(p);
D
dapan1121 已提交
135 136 137 138 139 140
    }
    pIter = taosHashIterate(res, pIter);
  }
  taosHashCleanup(res);
}

D
dapan1121 已提交
141
void sclFreeParam(SScalarParam *param) {
142 143
  if (param->columnData != NULL) {
    colDataDestroy(param->columnData);
wmmhello's avatar
wmmhello 已提交
144
    taosMemoryFree(param->columnData);
145 146 147 148 149
  }

  if (param->pHashFilter != NULL) {
    taosHashCleanup(param->pHashFilter);
  }
D
dapan1121 已提交
150 151
}

D
dapan1121 已提交
152 153 154 155 156
int32_t sclCopyValueNodeValue(SValueNode *pNode, void **res) {
  if (TSDB_DATA_TYPE_NULL == pNode->node.resType.type) {
    return TSDB_CODE_SUCCESS;
  }
  
wafwerar's avatar
wafwerar 已提交
157
  *res = taosMemoryMalloc(pNode->node.resType.bytes);
D
dapan1121 已提交
158 159 160 161 162 163 164 165 166
  if (NULL == (*res)) {
    sclError("malloc %d failed", pNode->node.resType.bytes);
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

  memcpy(*res, nodesGetValueFromNode(pNode), pNode->node.resType.bytes);
  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
167 168 169 170
int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t *rowNum) {
  switch (nodeType(node)) {
    case QUERY_NODE_VALUE: {
      SValueNode *valueNode = (SValueNode *)node;
171 172 173 174

      param->numOfRows = 1;
      param->columnData = createColumnInfoData(&valueNode->node.resType, 1);
      if (TSDB_DATA_TYPE_NULL == valueNode->node.resType.type) {
175
        colDataAppendNULL(param->columnData, 0);
176 177
      } else {
        colDataAppend(param->columnData, 0, nodesGetValueFromNode(valueNode), false);
D
dapan1121 已提交
178
      }
D
dapan1121 已提交
179 180
      break;
    }
D
dapan1121 已提交
181 182
    case QUERY_NODE_NODE_LIST: {
      SNodeListNode *nodeList = (SNodeListNode *)node;
183 184
      if (LIST_LENGTH(nodeList->pNodeList) <= 0) {
        sclError("invalid length in nodeList, length:%d", LIST_LENGTH(nodeList->pNodeList));
D
dapan1121 已提交
185 186 187
        SCL_RET(TSDB_CODE_QRY_INVALID_INPUT);
      }

D
dapan1121 已提交
188
      SCL_ERR_RET(scalarGenerateSetFromList((void **)&param->pHashFilter, node, nodeList->dataType.type));
D
dapan 已提交
189
      if (taosHashPut(ctx->pRes, &node, POINTER_BYTES, param, sizeof(*param))) {
190
        taosHashCleanup(param->pHashFilter);
D
dapan 已提交
191 192 193
        sclError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param));
        return TSDB_CODE_QRY_OUT_OF_MEMORY;
      }   
D
dapan1121 已提交
194 195
      break;
    }
X
Xiaoyu Wang 已提交
196
    case QUERY_NODE_COLUMN: {
D
dapan 已提交
197 198
      if (NULL == ctx->pBlockList) {
        sclError("invalid node type for constant calculating, type:%d, src:%p", nodeType(node), ctx->pBlockList);
D
dapan1121 已提交
199 200 201
        SCL_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
      }
      
X
Xiaoyu Wang 已提交
202
      SColumnNode *ref = (SColumnNode *)node;
203 204 205 206 207 208 209 210 211 212 213

      int32_t index = -1;
      for(int32_t i = 0; i < taosArrayGetSize(ctx->pBlockList); ++i) {
        SSDataBlock* pb = taosArrayGetP(ctx->pBlockList, i);
        if (pb->info.blockId == ref->dataBlockId) {
          index = i;
          break;
        }
      }

      if (index == -1) {
D
dapan1121 已提交
214
        sclError("column tupleId is too big, tupleId:%d, dataBlockNum:%d", ref->dataBlockId, (int32_t)taosArrayGetSize(ctx->pBlockList));
D
dapan 已提交
215 216 217
        SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
      }

218 219
      SSDataBlock *block = *(SSDataBlock **)taosArrayGet(ctx->pBlockList, index);
      if (NULL == block || ref->slotId >= block->info.numOfCols) {
D
dapan 已提交
220
        sclError("column slotId is too big, slodId:%d, dataBlockSize:%d", ref->slotId, (int32_t)taosArrayGetSize(block->pDataBlock));
D
dapan1121 已提交
221 222 223
        SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
      }

D
dapan 已提交
224
      SColumnInfoData *columnData = (SColumnInfoData *)taosArrayGet(block->pDataBlock, ref->slotId);
225 226
      param->numOfRows = block->info.rows;
      param->columnData = columnData;
D
dapan1121 已提交
227 228
      break;
    }
229 230 231
    case QUERY_NODE_FUNCTION:
    case QUERY_NODE_OPERATOR:
    case QUERY_NODE_LOGIC_CONDITION: {
D
dapan1121 已提交
232 233 234 235 236 237 238 239
      SScalarParam *res = (SScalarParam *)taosHashGet(ctx->pRes, &node, POINTER_BYTES);
      if (NULL == res) {
        sclError("no result for node, type:%d, node:%p", nodeType(node), node);
        SCL_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
      }
      *param = *res;
      break;
    }
240 241
    default:
      break;
D
dapan1121 已提交
242 243
  }

244 245 246
  if (param->numOfRows > *rowNum) {
    if ((1 != param->numOfRows) && (1 < *rowNum)) {
      sclError("different row nums, rowNum:%d, newRowNum:%d", *rowNum, param->numOfRows);
D
dapan1121 已提交
247 248 249
      SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
    }
    
250
    *rowNum = param->numOfRows;
D
dapan1121 已提交
251 252 253 254 255
  }

  return TSDB_CODE_SUCCESS;
}

D
dapan 已提交
256
int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarCtx *ctx, int32_t *paramNum, int32_t *rowNum) {  
D
dapan1121 已提交
257
  int32_t code = 0;
D
dapan1121 已提交
258 259 260 261 262 263 264 265
  if (NULL == pParamList) {
    if (ctx->pBlockList) {
      SSDataBlock *pBlock = taosArrayGet(ctx->pBlockList, 0);
      *rowNum = pBlock->info.rows;
    } else {
      *rowNum = 1;
    }

D
dapan 已提交
266
    *paramNum = 1;
D
dapan1121 已提交
267
  } else {
D
dapan 已提交
268
    *paramNum = pParamList->length;
D
dapan1121 已提交
269 270
  }

D
dapan 已提交
271
  SScalarParam *paramList = taosMemoryCalloc(*paramNum, sizeof(SScalarParam));
D
dapan1121 已提交
272
  if (NULL == paramList) {
D
dapan 已提交
273
    sclError("calloc %d failed", (int32_t)((*paramNum) * sizeof(SScalarParam)));
D
dapan1121 已提交
274
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
D
dapan1121 已提交
275 276
  }

D
dapan1121 已提交
277 278 279 280 281 282
  if (pParamList) {
    SNode *tnode = NULL;
    int32_t i = 0;
    if (SCL_IS_CONST_CALC(ctx)) {
      WHERE_EACH (tnode, pParamList) { 
        if (!SCL_IS_CONST_NODE(tnode)) {
D
dapan 已提交
283
          WHERE_NEXT;
D
dapan1121 已提交
284 285 286 287 288 289 290 291 292 293 294 295
        } else {
          SCL_ERR_JRET(sclInitParam(tnode, &paramList[i], ctx, rowNum));
          ERASE_NODE(pParamList);
        }
        
        ++i;
      }
    } else {
      FOREACH(tnode, pParamList) { 
        SCL_ERR_JRET(sclInitParam(tnode, &paramList[i], ctx, rowNum));
        ++i;
      }
D
dapan1121 已提交
296
    }
D
dapan1121 已提交
297 298 299
  } else {
    paramList[0].numOfRows = *rowNum;
  }
D
dapan1121 已提交
300

D
dapan1121 已提交
301 302
  if (0 == *rowNum) {
    taosMemoryFreeClear(paramList);    
D
dapan1121 已提交
303
  }
D
dapan1121 已提交
304

D
dapan1121 已提交
305
  *pParams = paramList;
D
dapan1121 已提交
306
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
307

D
dapan1121 已提交
308
_return:
wafwerar's avatar
wafwerar 已提交
309
  taosMemoryFreeClear(paramList);
D
dapan1121 已提交
310 311 312 313 314
  SCL_RET(code);
}

int32_t sclInitOperatorParams(SScalarParam **pParams, SOperatorNode *node, SScalarCtx *ctx, int32_t *rowNum) {
  int32_t code = 0;
D
dapan1121 已提交
315
  int32_t paramNum = scalarGetOperatorParamNum(node->opType);
D
dapan1121 已提交
316 317 318 319 320
  if (NULL == node->pLeft || (paramNum == 2 && NULL == node->pRight)) {
    sclError("invalid operation node, left:%p, right:%p", node->pLeft, node->pRight);
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }
  
wafwerar's avatar
wafwerar 已提交
321
  SScalarParam *paramList = taosMemoryCalloc(paramNum, sizeof(SScalarParam));
D
dapan1121 已提交
322 323
  if (NULL == paramList) {
    sclError("calloc %d failed", (int32_t)(paramNum * sizeof(SScalarParam)));
D
dapan1121 已提交
324 325 326
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

D
dapan1121 已提交
327
  SCL_ERR_JRET(sclInitParam(node->pLeft, &paramList[0], ctx, rowNum));
D
dapan1121 已提交
328
  if (paramNum > 1) {
D
dapan1121 已提交
329
    SCL_ERR_JRET(sclInitParam(node->pRight, &paramList[1], ctx, rowNum));
D
dapan1121 已提交
330 331
  }

D
dapan1121 已提交
332
  *pParams = paramList;
D
dapan1121 已提交
333
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
334 335

_return:
wafwerar's avatar
wafwerar 已提交
336
  taosMemoryFreeClear(paramList);
D
dapan1121 已提交
337
  SCL_RET(code);
D
dapan1121 已提交
338 339
}

340
int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *output) {
D
dapan1121 已提交
341 342
  SScalarParam *params = NULL;
  int32_t rowNum = 0;
D
dapan 已提交
343
  int32_t paramNum = 0;
D
dapan1121 已提交
344
  int32_t code = 0;
D
dapan 已提交
345
  SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &paramNum, &rowNum));
D
dapan1121 已提交
346

D
dapan1121 已提交
347 348 349
  if (fmIsUserDefinedFunc(node->funcId)) {
    UdfcFuncHandle udfHandle = NULL;
    
350 351 352 353 354
    code = setupUdf(node->functionName, &udfHandle);
    if (code != 0) {
      sclError("fmExecFunction error. setupUdf. function name: %s, code:%d", node->functionName, code);
      goto _return;
    }
D
dapan1121 已提交
355
    code = callUdfScalarFunc(udfHandle, params, paramNum, output);
356 357 358 359 360 361 362 363 364
    if (code != 0) {
      sclError("fmExecFunction error. callUdfScalarFunc. function name: %s, udf code:%d", node->functionName, code);
      goto _return;
    }
    code = teardownUdf(udfHandle);
    if (code != 0) {
      sclError("fmExecFunction error. callUdfScalarFunc. function name: %s, udf code:%d", node->functionName, code);
      goto _return;
    }
D
dapan1121 已提交
365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383
  } else {
    SScalarFuncExecFuncs ffpSet = {0};
    code = fmGetScalarFuncExecFuncs(node->funcId, &ffpSet);
    if (code) {
      sclError("fmGetFuncExecFuncs failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
      SCL_ERR_JRET(code);
    }
  
    output->columnData = createColumnInfoData(&node->node.resType, rowNum);
    if (output->columnData == NULL) {
      sclError("calloc %d failed", (int32_t)(rowNum * output->columnData->info.bytes));
      SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
    }

    code = (*ffpSet.process)(params, paramNum, output);
    if (code) {
      sclError("scalar function exec failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
      SCL_ERR_JRET(code);
    }
D
dapan1121 已提交
384 385 386 387
  }

_return:

D
dapan 已提交
388
  for (int32_t i = 0; i < paramNum; ++i) {
H
Haojun Liao 已提交
389
//    sclFreeParamNoData(params + i);
D
dapan1121 已提交
390 391
  }

wafwerar's avatar
wafwerar 已提交
392
  taosMemoryFreeClear(params);
D
dapan1121 已提交
393 394 395 396 397 398 399 400 401 402 403 404
  SCL_RET(code);
}

int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *output) {
  if (NULL == node->pParameterList || node->pParameterList->length <= 0) {
    sclError("invalid logic parameter list, list:%p, paramNum:%d", node->pParameterList, node->pParameterList ? node->pParameterList->length : 0);
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

  if (TSDB_DATA_TYPE_BOOL != node->node.resType.type) {
    sclError("invalid logic resType, type:%d", node->node.resType.type);
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
D
dapan1121 已提交
405 406
  }

D
dapan1121 已提交
407 408 409 410 411 412 413
  if (LOGIC_COND_TYPE_NOT == node->condType && node->pParameterList->length > 1) {
    sclError("invalid NOT operation parameter number, paramNum:%d", node->pParameterList->length);
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

  SScalarParam *params = NULL;
  int32_t rowNum = 0;
D
dapan 已提交
414
  int32_t paramNum = 0;
D
dapan1121 已提交
415
  int32_t code = 0;
D
dapan 已提交
416
  SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &paramNum, &rowNum));
D
dapan1121 已提交
417 418 419 420
  if (NULL == params) {
    output->numOfRows = 0;
    return TSDB_CODE_SUCCESS;
  }
D
dapan1121 已提交
421

422 423 424 425 426 427
  int32_t type = node->node.resType.type;
  output->numOfRows = rowNum;

  SDataType t = {.type = type, .bytes = tDataTypes[type].bytes};
  output->columnData = createColumnInfoData(&t, rowNum);
  if (output->columnData == NULL) {
D
dapan1121 已提交
428
    sclError("calloc %d failed", (int32_t)(rowNum * sizeof(bool)));
D
dapan1121 已提交
429 430
    SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
431

D
dapan1121 已提交
432
  bool value = false;
D
dapan 已提交
433
  bool complete = true;
D
dapan1121 已提交
434
  for (int32_t i = 0; i < rowNum; ++i) {
D
dapan 已提交
435 436
    complete = true;
    for (int32_t m = 0; m < paramNum; ++m) {
D
dapan1121 已提交
437
      if (NULL == params[m].columnData) {
D
dapan 已提交
438
        complete = false;
D
dapan1121 已提交
439 440
        continue;
      }
441 442 443
      char* p = colDataGetData(params[m].columnData, i);
      GET_TYPED_DATA(value, bool, params[m].columnData->info.type, p);

D
dapan1121 已提交
444
      if (LOGIC_COND_TYPE_AND == node->condType && (false == value)) {
D
dapan1121 已提交
445
        complete = true;
D
dapan1121 已提交
446 447
        break;
      } else if (LOGIC_COND_TYPE_OR == node->condType && value) {
D
dapan1121 已提交
448
        complete = true;
D
dapan1121 已提交
449 450 451 452 453 454
        break;
      } else if (LOGIC_COND_TYPE_NOT == node->condType) {
        value = !value;
      }
    }

D
dapan 已提交
455 456 457
    if (complete) {
      colDataAppend(output->columnData, i, (char*) &value, false);
    }
D
dapan1121 已提交
458 459
  }

D
dapan1121 已提交
460 461 462 463 464
  if (SCL_IS_CONST_CALC(ctx) && (false == complete)) {
    sclFreeParam(output);
    output->numOfRows = 0;
  }

D
dapan1121 已提交
465
_return:
D
dapan1121 已提交
466

D
dapan 已提交
467
  for (int32_t i = 0; i < paramNum; ++i) {
H
Haojun Liao 已提交
468
//    sclFreeParamNoData(params + i);
D
dapan1121 已提交
469 470
  }

wafwerar's avatar
wafwerar 已提交
471
  taosMemoryFreeClear(params);
D
dapan1121 已提交
472
  SCL_RET(code);
D
dapan1121 已提交
473 474 475 476 477 478
}

int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *output) {
  SScalarParam *params = NULL;
  int32_t rowNum = 0;
  int32_t code = 0;
479

D
dapan1121 已提交
480
  SCL_ERR_RET(sclInitOperatorParams(&params, node, ctx, &rowNum));
481 482 483
  output->columnData = createColumnInfoData(&node->node.resType, rowNum);
  if (output->columnData == NULL) {
    sclError("calloc failed, size:%d", (int32_t)rowNum * node->node.resType.bytes);
D
dapan1121 已提交
484 485 486 487 488
    SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

  _bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(node->opType);

D
dapan1121 已提交
489
  int32_t paramNum = scalarGetOperatorParamNum(node->opType);
D
dapan1121 已提交
490 491
  SScalarParam* pLeft = &params[0];
  SScalarParam* pRight = paramNum > 1 ? &params[1] : NULL;
492

D
dapan 已提交
493
  OperatorFn(pLeft, pRight, output, TSDB_ORDER_ASC);
D
dapan1121 已提交
494 495

_return:
D
dapan1121 已提交
496
  for (int32_t i = 0; i < paramNum; ++i) {
497
//    sclFreeParam(&params[i]);
D
dapan1121 已提交
498 499
  }

wafwerar's avatar
wafwerar 已提交
500
  taosMemoryFreeClear(params);
D
dapan1121 已提交
501
  SCL_RET(code);
D
dapan1121 已提交
502 503
}

D
dapan1121 已提交
504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525
EDealRes sclRewriteBasedOnOptr(SNode** pNode, SScalarCtx *ctx, EOperatorType opType) {
  if (opType <= OP_TYPE_CALC_MAX) {
    SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
    if (NULL == res) {
      sclError("make value node failed");    
      ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
      return DEAL_RES_ERROR;
    }
    
    res->node.resType.type = TSDB_DATA_TYPE_NULL;
    
    nodesDestroyNode(*pNode);
    *pNode = (SNode*)res;
  } else {
    SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
    if (NULL == res) {
      sclError("make value node failed");    
      ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
      return DEAL_RES_ERROR;
    }
    
    res->node.resType.type = TSDB_DATA_TYPE_BOOL;
D
dapan1121 已提交
526
    res->node.resType.bytes = tDataTypes[TSDB_DATA_TYPE_BOOL].bytes;
D
dapan1121 已提交
527 528 529 530 531 532 533 534 535 536 537 538 539 540 541
    res->datum.b = false;
    
    nodesDestroyNode(*pNode);
    *pNode = (SNode*)res;
  }

  return DEAL_RES_CONTINUE;
}


EDealRes sclRewriteOperatorForNullValue(SNode** pNode, SScalarCtx *ctx) {
  SOperatorNode *node = (SOperatorNode *)*pNode;

  if (node->pLeft && (QUERY_NODE_VALUE == nodeType(node->pLeft))) {
    SValueNode *valueNode = (SValueNode *)node->pLeft;
D
dapan1121 已提交
542
    if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL)) {
D
dapan1121 已提交
543 544 545 546 547 548
      return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
    }
  }

  if (node->pRight && (QUERY_NODE_VALUE == nodeType(node->pRight))) {
    SValueNode *valueNode = (SValueNode *)node->pRight;
D
dapan1121 已提交
549
    if (SCL_IS_NULL_VALUE_NODE(valueNode) && (node->opType != OP_TYPE_IS_NULL && node->opType != OP_TYPE_IS_NOT_NULL)) {
D
dapan1121 已提交
550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577
      return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
    }
  }

  if (node->pRight && (QUERY_NODE_NODE_LIST == nodeType(node->pRight))) {
    SNodeListNode *listNode = (SNodeListNode *)node->pRight;
    SNode* tnode = NULL;
    WHERE_EACH(tnode, listNode->pNodeList) {
      if (SCL_IS_NULL_VALUE_NODE(tnode)) {
        if (node->opType == OP_TYPE_IN) {
          ERASE_NODE(listNode->pNodeList);
          continue;
        } else { //OP_TYPE_NOT_IN
          return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
        }
      }

      WHERE_NEXT;
    }

    if (listNode->pNodeList->length <= 0) {
      return sclRewriteBasedOnOptr(pNode, ctx, node->opType);
    }
  }

  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
578
EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
579
  SFunctionNode *node = (SFunctionNode *)*pNode;
D
dapan1121 已提交
580
  SNode* tnode = NULL;
581
  if (!fmIsScalarFunc(node->funcId)) {
D
dapan1121 已提交
582 583
    return DEAL_RES_CONTINUE;
  }
584

D
dapan1121 已提交
585 586 587 588 589 590
  FOREACH(tnode, node->pParameterList) {
    if (!SCL_IS_CONST_NODE(tnode)) {
      return DEAL_RES_CONTINUE;
    }
  }

D
dapan1121 已提交
591
  SScalarParam output = {0};
592

593
  ctx->code = sclExecFunction(node, ctx, &output);
D
dapan 已提交
594
  if (ctx->code) {
D
dapan1121 已提交
595 596 597
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
598
  SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
D
dapan1121 已提交
599 600
  if (NULL == res) {
    sclError("make value node failed");
D
dapan1121 已提交
601
    sclFreeParam(&output);
D
dapan 已提交
602
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
603 604 605
    return DEAL_RES_ERROR;
  }

606 607
  res->translate = true;

608 609
  if (colDataIsNull_s(output.columnData, 0)) {
    res->node.resType.type = TSDB_DATA_TYPE_NULL;
D
dapan1121 已提交
610
  } else {
D
dapan1121 已提交
611 612 613 614
    res->node.resType.type = output.columnData->info.type;
    res->node.resType.bytes = output.columnData->info.bytes;
    res->node.resType.scale = output.columnData->info.scale;
    res->node.resType.precision = output.columnData->info.precision;
615 616
    int32_t type = output.columnData->info.type;
    if (IS_VAR_DATA_TYPE(type)) {
617
      res->datum.p = taosMemoryCalloc(res->node.resType.bytes + VARSTR_HEADER_SIZE + 1, 1);
618
      memcpy(res->datum.p, output.columnData->pData, varDataTLen(output.columnData->pData));
619
    } else {
D
dapan1121 已提交
620
      nodesSetValueNodeValue(res, output.columnData->pData);
621
    }
D
dapan1121 已提交
622
  }
623

D
dapan1121 已提交
624 625 626
  nodesDestroyNode(*pNode);
  *pNode = (SNode*)res;

D
dapan1121 已提交
627
  sclFreeParam(&output);
D
dapan1121 已提交
628 629 630
  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
631
EDealRes sclRewriteLogic(SNode** pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
632 633
  SLogicConditionNode *node = (SLogicConditionNode *)*pNode;

H
Haojun Liao 已提交
634
  SScalarParam output = {0};
D
dapan 已提交
635 636
  ctx->code = sclExecLogic(node, ctx, &output);
  if (ctx->code) {
D
dapan1121 已提交
637 638 639
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
640 641 642 643
  if (0 == output.numOfRows) {
    return DEAL_RES_CONTINUE;
  }

D
dapan1121 已提交
644
  SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
D
dapan1121 已提交
645 646
  if (NULL == res) {
    sclError("make value node failed");
647
    sclFreeParam(&output);
D
dapan 已提交
648
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
649 650 651 652
    return DEAL_RES_ERROR;
  }

  res->node.resType = node->node.resType;
653
  res->translate = true;
D
dapan1121 已提交
654

655 656 657 658
  int32_t type = output.columnData->info.type;
  if (IS_VAR_DATA_TYPE(type)) {
    res->datum.p = output.columnData->pData;
    output.columnData->pData = NULL;
D
dapan1121 已提交
659
  } else {
D
dapan1121 已提交
660
    nodesSetValueNodeValue(res, output.columnData->pData);
D
dapan1121 已提交
661
  }
D
dapan1121 已提交
662 663 664 665

  nodesDestroyNode(*pNode);
  *pNode = (SNode*)res;

D
dapan1121 已提交
666
  sclFreeParam(&output);
D
dapan1121 已提交
667 668 669
  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
670
EDealRes sclRewriteOperator(SNode** pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
671
  SOperatorNode *node = (SOperatorNode *)*pNode;
D
dapan1121 已提交
672

D
dapan1121 已提交
673 674
  if ((!SCL_IS_CONST_NODE(node->pLeft)) || (!SCL_IS_CONST_NODE(node->pRight))) {
    return sclRewriteOperatorForNullValue(pNode, ctx);
D
dapan1121 已提交
675 676
  }

H
Haojun Liao 已提交
677
  SScalarParam output = {.columnData = taosMemoryCalloc(1, sizeof(SColumnInfoData))};
D
dapan 已提交
678 679
  ctx->code = sclExecOperator(node, ctx, &output);
  if (ctx->code) {
D
dapan1121 已提交
680 681 682
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
683
  SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
D
dapan1121 已提交
684
  if (NULL == res) {
685 686
    sclError("make value node failed");
    sclFreeParam(&output);
D
dapan 已提交
687
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
688 689 690
    return DEAL_RES_ERROR;
  }

691
  res->translate = true;
D
dapan1121 已提交
692

693 694
  if (colDataIsNull_s(output.columnData, 0)) {
    res->node.resType.type = TSDB_DATA_TYPE_NULL;
D
dapan1121 已提交
695
  } else {
696 697 698 699 700 701
    res->node.resType = node->node.resType;
    int32_t type = output.columnData->info.type;
    if (IS_VAR_DATA_TYPE(type)) {  // todo refactor
      res->datum.p = output.columnData->pData;
      output.columnData->pData = NULL;
    } else {
D
dapan1121 已提交
702
      nodesSetValueNodeValue(res, output.columnData->pData);    
703
    }
D
dapan1121 已提交
704
  }
D
dapan1121 已提交
705 706 707 708

  nodesDestroyNode(*pNode);
  *pNode = (SNode*)res;

H
Haojun Liao 已提交
709
  sclFreeParam(&output);
D
dapan1121 已提交
710 711 712 713
  return DEAL_RES_CONTINUE;
}

EDealRes sclConstantsRewriter(SNode** pNode, void* pContext) {
D
dapan 已提交
714 715
  SScalarCtx *ctx = (SScalarCtx *)pContext;

D
dapan1121 已提交
716
  if (QUERY_NODE_FUNCTION == nodeType(*pNode)) {
D
dapan 已提交
717
    return sclRewriteFunction(pNode, ctx);
D
dapan1121 已提交
718 719 720
  }

  if (QUERY_NODE_LOGIC_CONDITION == nodeType(*pNode)) {
D
dapan 已提交
721
    return sclRewriteLogic(pNode, ctx);
D
dapan1121 已提交
722 723
  }

D
dapan1121 已提交
724
  if (QUERY_NODE_OPERATOR == nodeType(*pNode)) {
D
dapan 已提交
725
    return sclRewriteOperator(pNode, ctx);
726
  }
727 728

  return DEAL_RES_CONTINUE;
D
dapan1121 已提交
729 730
}

D
dapan 已提交
731
EDealRes sclWalkFunction(SNode* pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
732
  SFunctionNode *node = (SFunctionNode *)pNode;
D
dapan1121 已提交
733
  SScalarParam output = {0};
734

735
  ctx->code = sclExecFunction(node, ctx, &output);
D
dapan1121 已提交
736 737
  if (ctx->code) {
    return DEAL_RES_ERROR;
D
dapan1121 已提交
738 739
  }

D
dapan1121 已提交
740
  if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
D
dapan1121 已提交
741 742
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
    return DEAL_RES_ERROR;
D
dapan1121 已提交
743 744
  }

D
dapan1121 已提交
745 746 747
  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
748
EDealRes sclWalkLogic(SNode* pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
749
  SLogicConditionNode *node = (SLogicConditionNode *)pNode;
D
dapan1121 已提交
750
  SScalarParam output = {0};
751

D
dapan1121 已提交
752 753 754
  ctx->code = sclExecLogic(node, ctx, &output);
  if (ctx->code) {
    return DEAL_RES_ERROR;
D
dapan1121 已提交
755 756
  }

D
dapan1121 已提交
757
  if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
D
dapan1121 已提交
758
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
759
    return DEAL_RES_ERROR;
D
dapan1121 已提交
760 761 762 763 764
  }

  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
765
EDealRes sclWalkOperator(SNode* pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
766
  SOperatorNode *node = (SOperatorNode *)pNode;
D
dapan1121 已提交
767
  SScalarParam output = {0};
D
dapan1121 已提交
768
  
D
dapan1121 已提交
769 770
  ctx->code = sclExecOperator(node, ctx, &output);
  if (ctx->code) {
D
dapan1121 已提交
771 772
    return DEAL_RES_ERROR;
  }
D
dapan1121 已提交
773

D
dapan1121 已提交
774
  if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
D
dapan1121 已提交
775
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
776 777 778
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
779 780 781
  return DEAL_RES_CONTINUE;
}

782 783 784 785 786 787 788 789 790 791 792 793 794 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
EDealRes sclWalkTarget(SNode* pNode, SScalarCtx *ctx) {
  STargetNode *target = (STargetNode *)pNode;

  if (target->dataBlockId >= taosArrayGetSize(ctx->pBlockList)) {
    sclError("target tupleId is too big, tupleId:%d, dataBlockNum:%d", target->dataBlockId, (int32_t)taosArrayGetSize(ctx->pBlockList));
    ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
    return DEAL_RES_ERROR;
  }

  int32_t index = -1;
  for(int32_t i = 0; i < taosArrayGetSize(ctx->pBlockList); ++i) {
    SSDataBlock* pb = taosArrayGetP(ctx->pBlockList, i);
    if (pb->info.blockId == target->dataBlockId) {
      index = i;
      break;
    }
  }

  if (index == -1) {
    sclError("column tupleId is too big, tupleId:%d, dataBlockNum:%d", target->dataBlockId, (int32_t)taosArrayGetSize(ctx->pBlockList));
    ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
    return DEAL_RES_ERROR;
  }

  SSDataBlock *block = *(SSDataBlock **)taosArrayGet(ctx->pBlockList, index);

  if (target->slotId >= taosArrayGetSize(block->pDataBlock)) {
    sclError("target slot not exist, dataBlockId:%d, slotId:%d, dataBlockNum:%d", target->dataBlockId, target->slotId, (int32_t)taosArrayGetSize(block->pDataBlock));
    ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
    return DEAL_RES_ERROR;
  }

  // if block->pDataBlock is not enough, there are problems if target->slotId bigger than the size of block->pDataBlock,
  SColumnInfoData *col = taosArrayGet(block->pDataBlock, target->slotId);

  SScalarParam *res = (SScalarParam *)taosHashGet(ctx->pRes, (void *)&target->pExpr, POINTER_BYTES);
  if (NULL == res) {
    sclError("no valid res in hash, node:%p, type:%d", target->pExpr, nodeType(target->pExpr));
    ctx->code = TSDB_CODE_QRY_APP_ERROR;
    return DEAL_RES_ERROR;
  }

  colDataAssign(col, res->columnData, res->numOfRows);
  block->info.rows = res->numOfRows;

  sclFreeParam(res);
  taosHashRemove(ctx->pRes, (void *)&target->pExpr, POINTER_BYTES);
  return DEAL_RES_CONTINUE;
}
D
dapan 已提交
831

D
dapan1121 已提交
832
EDealRes sclCalcWalker(SNode* pNode, void* pContext) {
X
Xiaoyu Wang 已提交
833
  if (QUERY_NODE_VALUE == nodeType(pNode) || QUERY_NODE_NODE_LIST == nodeType(pNode) || QUERY_NODE_COLUMN == nodeType(pNode)) {
D
dapan1121 已提交
834
    return DEAL_RES_CONTINUE;
D
dapan1121 已提交
835
  }
D
dapan 已提交
836 837

  SScalarCtx *ctx = (SScalarCtx *)pContext;
D
dapan1121 已提交
838
  if (QUERY_NODE_FUNCTION == nodeType(pNode)) {
D
dapan 已提交
839
    return sclWalkFunction(pNode, ctx);
D
dapan1121 已提交
840
  }
D
dapan1121 已提交
841

D
dapan1121 已提交
842
  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pNode)) {
D
dapan 已提交
843
    return sclWalkLogic(pNode, ctx);
D
dapan1121 已提交
844
  }
D
dapan1121 已提交
845

D
dapan1121 已提交
846
  if (QUERY_NODE_OPERATOR == nodeType(pNode)) {
D
dapan 已提交
847
    return sclWalkOperator(pNode, ctx);
D
dapan1121 已提交
848
  }
D
dapan1121 已提交
849

850 851 852
  if (QUERY_NODE_TARGET == nodeType(pNode)) {
    return sclWalkTarget(pNode, ctx);
  }
D
dapan1121 已提交
853

D
dapan 已提交
854
  sclError("invalid node type for scalar calculating, type:%d", nodeType(pNode));
D
dapan1121 已提交
855 856
  ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
  return DEAL_RES_ERROR;
D
dapan1121 已提交
857 858 859 860 861 862 863 864
}

int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes) {
  if (NULL == pNode) {
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

  int32_t code = 0;
D
dapan 已提交
865
  SScalarCtx ctx = {0};
866
  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
D
dapan 已提交
867 868 869 870
  if (NULL == ctx.pRes) {
    sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
871
  
X
Xiaoyu Wang 已提交
872
  nodesRewriteExprPostOrder(&pNode, sclConstantsRewriter, (void *)&ctx);
D
dapan 已提交
873
  SCL_ERR_JRET(ctx.code);
D
dapan1121 已提交
874 875
  *pRes = pNode;

D
dapan 已提交
876 877 878
_return:
  sclFreeRes(ctx.pRes);
  return code;
D
dapan1121 已提交
879 880
}

D
dapan 已提交
881
int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
D
dapan1121 已提交
882
  if (NULL == pNode || NULL == pBlockList) {
D
dapan1121 已提交
883 884 885 886
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

  int32_t code = 0;
D
dapan 已提交
887
  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList};
888

889
  // TODO: OPT performance
890
  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
D
dapan1121 已提交
891 892 893 894
  if (NULL == ctx.pRes) {
    sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
895
  
X
Xiaoyu Wang 已提交
896
  nodesWalkExprPostOrder(pNode, sclCalcWalker, (void *)&ctx);
D
dapan 已提交
897
  SCL_ERR_JRET(ctx.code);
D
dapan1121 已提交
898

D
dapan1121 已提交
899 900 901 902 903 904 905
  if (pDst) {
    SScalarParam *res = (SScalarParam *)taosHashGet(ctx.pRes, (void *)&pNode, POINTER_BYTES);
    if (NULL == res) {
      sclError("no valid res in hash, node:%p, type:%d", pNode, nodeType(pNode));
      SCL_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
    }
    
906 907
    colDataAssign(pDst->columnData, res->columnData, res->numOfRows);
    pDst->numOfRows = res->numOfRows;
D
dapan1121 已提交
908
    taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES);
D
dapan1121 已提交
909
  }
D
dapan1121 已提交
910

D
dapan 已提交
911
_return:
D
dapan1121 已提交
912
  //nodesDestroyNode(pNode);
D
dapan 已提交
913 914
  sclFreeRes(ctx.pRes);
  return code;
D
dapan1121 已提交
915
}