scalar.c 22.4 KB
Newer Older
D
dapan1121 已提交
1
#include "nodes.h"
D
dapan1121 已提交
2 3 4 5 6 7
#include "common.h"
#include "querynodes.h"
#include "function.h"
#include "functionMgt.h"
#include "sclvector.h"
#include "sclInt.h"
D
dapan 已提交
8
#include "tep.h"
D
dapan1121 已提交
9

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

  return 2;
}

D
dapan1121 已提交
19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53
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;
  SScalarParam in = {.num = 1}, out = {.num = 1, .type = type};
  int8_t dummy = 0;
  int32_t bufLen = 60;
  out.data = malloc(bufLen);
  int32_t len = 0;
  void *buf = NULL;
  
  for (int32_t i = 0; i < nodeList->pNodeList->length; ++i) {
    SValueNode *valueNode = (SValueNode *)cell->pNode;

    if (valueNode->node.resType.type != type) {
      in.type = valueNode->node.resType.type;
      in.bytes = valueNode->node.resType.bytes;
      in.data = nodesGetValueFromNode(valueNode);
    
      code = vectorConvertImpl(&in, &out);
      if (code) {
        sclError("convert from %d to %d failed", in.type, out.type);
        SCL_ERR_JRET(code);
      }

      if (IS_VAR_DATA_TYPE(type)) {
        len = varDataLen(out.data);
D
dapan1121 已提交
54
        buf = varDataVal(out.data);
D
dapan1121 已提交
55 56
      } else {
        len = tDataTypes[type].bytes;
D
dapan1121 已提交
57
        buf = out.data;
D
dapan1121 已提交
58 59 60
      }
    } else {
      buf = nodesGetValueFromNode(valueNode);
D
dapan1121 已提交
61 62 63 64 65 66 67
      if (IS_VAR_DATA_TYPE(type)) {
        len = varDataLen(buf);
        buf = varDataVal(buf);
      } else {
        len = valueNode->node.resType.bytes;
        buf = out.data;
      }      
D
dapan1121 已提交
68 69 70 71 72 73
    }
    
    if (taosHashPut(pObj, buf, (size_t)len, &dummy, sizeof(dummy))) {
      sclError("taosHashPut failed");
      SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
D
dapan1121 已提交
74 75

    cell = cell->pNext;
D
dapan1121 已提交
76 77 78 79 80 81 82 83 84 85 86 87 88 89 90
  }

  tfree(out.data);
  *data = pObj;

  return TSDB_CODE_SUCCESS;

_return:

  tfree(out.data);
  taosHashCleanup(pObj);
  
  SCL_RET(code);
}

D
dapan 已提交
91
FORCE_INLINE bool sclIsNull(SScalarParam* param, int32_t idx) {
D
dapan1121 已提交
92
  if (param->dataInBlock) {
D
dapan 已提交
93
    return colDataIsNull(param->orig.columnData, 0, idx, NULL);
D
dapan1121 已提交
94 95
  }

D
dapan 已提交
96
  return param->bitmap ? colDataIsNull_f(param->bitmap, idx) : false;
D
dapan1121 已提交
97 98
}

D
dapan 已提交
99 100 101 102 103 104 105 106 107
FORCE_INLINE void sclSetNull(SScalarParam* param, int32_t idx) {
  if (NULL == param->bitmap) {
    param->bitmap = calloc(BitmapLen(param->num), sizeof(char));
    if (NULL == param->bitmap) {
      sclError("calloc %d failed", param->num);
      return;
    }
  }
  
D
dapan1121 已提交
108 109 110
  colDataSetNull_f(param->bitmap, idx);
}

D
dapan1121 已提交
111

D
dapan1121 已提交
112 113 114 115 116 117 118
void sclFreeRes(SHashObj *res) {
  SScalarParam *p = NULL;
  void *pIter = taosHashIterate(res, NULL);
  while (pIter) {
    p = (SScalarParam *)pIter;

    if (p) {
D
dapan 已提交
119
      sclFreeParam(p);
D
dapan1121 已提交
120 121 122 123 124 125 126 127
    }
    
    pIter = taosHashIterate(res, pIter);
  }
  
  taosHashCleanup(res);
}

D
dapan1121 已提交
128
void sclFreeParamNoData(SScalarParam *param) {
D
dapan 已提交
129
  tfree(param->bitmap);
D
dapan1121 已提交
130 131 132 133 134
}


void sclFreeParam(SScalarParam *param) {
  sclFreeParamNoData(param);
D
dapan 已提交
135 136 137 138 139 140 141 142
  
  if (!param->dataInBlock) {
    if (SCL_DATA_TYPE_DUMMY_HASH == param->type) {
      taosHashCleanup((SHashObj *)param->orig.data);
    } else {
      tfree(param->orig.data);
    }
  }
D
dapan1121 已提交
143 144
}

D
dapan1121 已提交
145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161

int32_t sclCopyValueNodeValue(SValueNode *pNode, void **res) {
  if (TSDB_DATA_TYPE_NULL == pNode->node.resType.type) {
    return TSDB_CODE_SUCCESS;
  }
  
  *res = malloc(pNode->node.resType.bytes);
  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 已提交
162 163 164 165
int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t *rowNum) {
  switch (nodeType(node)) {
    case QUERY_NODE_VALUE: {
      SValueNode *valueNode = (SValueNode *)node;
D
dapan1121 已提交
166
      //SCL_ERR_RET(sclCopyValueNodeValue(valueNode, &param->data));
D
dapan1121 已提交
167
      param->data = nodesGetValueFromNode(valueNode);
D
dapan1121 已提交
168
      param->orig.data = param->data;
D
dapan1121 已提交
169 170 171
      param->num = 1;
      param->type = valueNode->node.resType.type;
      param->bytes = valueNode->node.resType.bytes;
D
dapan1121 已提交
172 173 174
      if (TSDB_DATA_TYPE_NULL == param->type) {
        sclSetNull(param, 0);
      }
D
dapan1121 已提交
175
      param->dataInBlock = false;
D
dapan1121 已提交
176 177 178
      
      break;
    }
D
dapan1121 已提交
179 180
    case QUERY_NODE_NODE_LIST: {
      SNodeListNode *nodeList = (SNodeListNode *)node;
D
dapan1121 已提交
181 182 183 184 185 186
      if (nodeList->pNodeList->length <= 0) {
        sclError("invalid length in nodeList, length:%d", nodeList->pNodeList->length);
        SCL_RET(TSDB_CODE_QRY_INVALID_INPUT);
      }

      SCL_ERR_RET(scalarGenerateSetFromList(&param->data, node, nodeList->dataType.type));
D
dapan 已提交
187
      param->orig.data = param->data;
D
dapan1121 已提交
188 189
      param->num = 1;
      param->type = SCL_DATA_TYPE_DUMMY_HASH;
D
dapan1121 已提交
190
      param->dataInBlock = false;
D
dapan 已提交
191 192 193 194 195 196 197

      if (taosHashPut(ctx->pRes, &node, POINTER_BYTES, param, sizeof(*param))) {
        taosHashCleanup(param->orig.data);
        param->orig.data = NULL;
        sclError("taosHashPut nodeList failed, size:%d", (int32_t)sizeof(*param));
        return TSDB_CODE_QRY_OUT_OF_MEMORY;
      }   
D
dapan1121 已提交
198 199
      break;
    }
X
Xiaoyu Wang 已提交
200
    case QUERY_NODE_COLUMN: {
D
dapan 已提交
201 202
      if (NULL == ctx->pBlockList) {
        sclError("invalid node type for constant calculating, type:%d, src:%p", nodeType(node), ctx->pBlockList);
D
dapan1121 已提交
203 204 205
        SCL_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
      }
      
X
Xiaoyu Wang 已提交
206
      SColumnNode *ref = (SColumnNode *)node;
D
dapan1121 已提交
207 208
      if (ref->dataBlockId >= taosArrayGetSize(ctx->pBlockList)) {
        sclError("column tupleId is too big, tupleId:%d, dataBlockNum:%d", ref->dataBlockId, (int32_t)taosArrayGetSize(ctx->pBlockList));
D
dapan 已提交
209 210 211
        SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
      }

D
dapan1121 已提交
212
      SSDataBlock *block = *(SSDataBlock **)taosArrayGet(ctx->pBlockList, ref->dataBlockId);
D
dapan 已提交
213 214 215
      
      if (NULL == block || ref->slotId >= taosArrayGetSize(block->pDataBlock)) {
        sclError("column slotId is too big, slodId:%d, dataBlockSize:%d", ref->slotId, (int32_t)taosArrayGetSize(block->pDataBlock));
D
dapan1121 已提交
216 217 218
        SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
      }

D
dapan 已提交
219
      SColumnInfoData *columnData = (SColumnInfoData *)taosArrayGet(block->pDataBlock, ref->slotId);
D
dapan1121 已提交
220
      param->data = NULL;
D
dapan 已提交
221
      param->orig.columnData = columnData;
D
dapan1121 已提交
222
      param->dataInBlock = true;
D
dapan1121 已提交
223
      
D
dapan 已提交
224
      param->num = block->info.rows;
D
dapan1121 已提交
225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244
      param->type = columnData->info.type;
      param->bytes = columnData->info.bytes;
      
      break;
    }
    case QUERY_NODE_LOGIC_CONDITION:
    case QUERY_NODE_OPERATOR: {
      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;
    }
  }

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

  return TSDB_CODE_SUCCESS;
}

D
dapan1121 已提交
256
int32_t sclMoveParamListData(SScalarParam *params, int32_t listNum, int32_t idx) {
D
dapan1121 已提交
257 258
  SScalarParam *param = NULL;
  
D
dapan1121 已提交
259
  for (int32_t i = 0; i < listNum; ++i) {
D
dapan1121 已提交
260 261 262 263 264 265
    param = params + i;
    
    if (1 == param->num) {
      continue;
    }

D
dapan1121 已提交
266
    if (param->dataInBlock) {
D
dapan 已提交
267
      param->data = colDataGet(param->orig.columnData, idx);
D
dapan1121 已提交
268 269 270 271 272 273
    } else if (idx) {
      if (IS_VAR_DATA_TYPE(param->type)) {
        param->data = (char *)(param->data) + varDataTLen(param->data);
      } else {
        param->data = (char *)(param->data) + tDataTypes[param->type].bytes;
      }
D
dapan 已提交
274 275
    } else {
      param->data = param->orig.data;
D
dapan1121 已提交
276 277 278 279 280 281 282 283
    }
  }

  return TSDB_CODE_SUCCESS;
}

int32_t sclInitParamList(SScalarParam **pParams, SNodeList* pParamList, SScalarCtx *ctx, int32_t *rowNum) {
  int32_t code = 0;
D
dapan1121 已提交
284 285 286
  SScalarParam *paramList = calloc(pParamList->length, sizeof(SScalarParam));
  if (NULL == paramList) {
    sclError("calloc %d failed", (int32_t)(pParamList->length * sizeof(SScalarParam)));
D
dapan1121 已提交
287
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
D
dapan1121 已提交
288 289
  }

D
dapan1121 已提交
290 291
  SListCell *cell = pParamList->pHead;
  for (int32_t i = 0; i < pParamList->length; ++i) {
D
dapan1121 已提交
292 293
    if (NULL == cell || NULL == cell->pNode) {
      sclError("invalid cell, cell:%p, pNode:%p", cell, cell->pNode);
D
dapan1121 已提交
294
      SCL_ERR_JRET(TSDB_CODE_QRY_INVALID_INPUT);
D
dapan1121 已提交
295 296
    }

D
dapan1121 已提交
297
    SCL_ERR_JRET(sclInitParam(cell->pNode, &paramList[i], ctx, rowNum));
D
dapan1121 已提交
298 299 300
    
    cell = cell->pNext;
  }
D
dapan1121 已提交
301

D
dapan1121 已提交
302 303
  *pParams = paramList;

D
dapan1121 已提交
304
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
305

D
dapan1121 已提交
306 307
_return:

D
dapan1121 已提交
308
  tfree(paramList);
D
dapan1121 已提交
309 310 311 312 313
  SCL_RET(code);
}

int32_t sclInitOperatorParams(SScalarParam **pParams, SOperatorNode *node, SScalarCtx *ctx, int32_t *rowNum) {
  int32_t code = 0;
D
dapan1121 已提交
314
  int32_t paramNum = scalarGetOperatorParamNum(node->opType);
D
dapan1121 已提交
315 316 317 318 319
  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);
  }
  
D
dapan1121 已提交
320 321 322
  SScalarParam *paramList = calloc(paramNum, sizeof(SScalarParam));
  if (NULL == paramList) {
    sclError("calloc %d failed", (int32_t)(paramNum * sizeof(SScalarParam)));
D
dapan1121 已提交
323 324 325
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

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

D
dapan1121 已提交
331 332
  *pParams = paramList;

D
dapan1121 已提交
333
  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
334 335 336

_return:

D
dapan1121 已提交
337
  tfree(paramList);
D
dapan1121 已提交
338
  SCL_RET(code);
D
dapan1121 已提交
339 340
}

D
dapan1121 已提交
341 342

int32_t sclExecFuncion(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *output) {
D
dapan1121 已提交
343 344
  if (NULL == node->pParameterList || node->pParameterList->length <= 0) {
    sclError("invalid function parameter list, list:%p, paramNum:%d", node->pParameterList, node->pParameterList ? node->pParameterList->length : 0);
D
dapan1121 已提交
345
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
D
dapan1121 已提交
346 347 348 349 350
  }

  SScalarFuncExecFuncs ffpSet = {0};
  int32_t code = fmGetScalarFuncExecFuncs(node->funcId, &ffpSet);
  if (code) {
D
dapan1121 已提交
351
    sclError("fmGetFuncExecFuncs failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
D
dapan1121 已提交
352
    SCL_ERR_RET(code);
D
dapan1121 已提交
353 354
  }

D
dapan1121 已提交
355 356 357 358 359 360 361
  SScalarParam *params = NULL;
  int32_t rowNum = 0;
  SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &rowNum));

  output->type = node->node.resType.type;
  output->data = calloc(rowNum, sizeof(tDataTypes[output->type].bytes));
  if (NULL == output->data) {
D
dapan1121 已提交
362
    sclError("calloc %d failed", (int32_t)(rowNum * sizeof(tDataTypes[output->type].bytes)));
D
dapan1121 已提交
363 364
    SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan 已提交
365
  output->orig.data = output->data;
D
dapan1121 已提交
366 367

  for (int32_t i = 0; i < rowNum; ++i) {
D
dapan1121 已提交
368 369 370
    sclMoveParamListData(output, 1, i);
    sclMoveParamListData(params, node->pParameterList->length, i);

D
dapan1121 已提交
371 372
    code = (*ffpSet.process)(params, node->pParameterList->length, output);
    if (code) {
D
dapan1121 已提交
373
      sclError("scalar function exec failed, funcId:%d, code:%s", node->funcId, tstrerror(code));
D
dapan1121 已提交
374 375 376 377 378 379
      SCL_ERR_JRET(code);    
    }
  }

_return:

D
dapan1121 已提交
380 381 382 383
  for (int32_t i = 0; i < node->pParameterList->length; ++i) {
    sclFreeParamNoData(params + i);
  }

D
dapan1121 已提交
384
  tfree(params);
D
dapan1121 已提交
385
  
D
dapan1121 已提交
386 387 388 389 390 391 392 393 394 395 396 397 398
  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 已提交
399 400
  }

D
dapan1121 已提交
401 402 403 404 405 406 407 408
  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;
  int32_t code = 0;
D
dapan1121 已提交
409
  
D
dapan1121 已提交
410 411 412
  SCL_ERR_RET(sclInitParamList(&params, node->pParameterList, ctx, &rowNum));

  output->type = node->node.resType.type;
D
dapan1121 已提交
413 414
  output->bytes = sizeof(bool);
  output->num = rowNum;
D
dapan1121 已提交
415 416
  output->data = calloc(rowNum, sizeof(bool));
  if (NULL == output->data) {
D
dapan1121 已提交
417
    sclError("calloc %d failed", (int32_t)(rowNum * sizeof(bool)));
D
dapan1121 已提交
418 419
    SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan 已提交
420
  output->orig.data = output->data;
D
dapan1121 已提交
421

D
dapan1121 已提交
422 423 424
  bool value = false;

  for (int32_t i = 0; i < rowNum; ++i) {
D
dapan1121 已提交
425 426 427
    sclMoveParamListData(output, 1, i);
    sclMoveParamListData(params, node->pParameterList->length, i);
  
D
dapan1121 已提交
428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444
    for (int32_t m = 0; m < node->pParameterList->length; ++m) {
      GET_TYPED_DATA(value, bool, params[m].type, params[m].data);
      
      if (LOGIC_COND_TYPE_AND == node->condType && (false == value)) {
        break;
      } else if (LOGIC_COND_TYPE_OR == node->condType && value) {
        break;
      } else if (LOGIC_COND_TYPE_NOT == node->condType) {
        value = !value;
      }
    }

    *(bool *)output->data = value;
  }

_return:

D
dapan1121 已提交
445 446 447 448
  for (int32_t i = 0; i < node->pParameterList->length; ++i) {
    sclFreeParamNoData(params + i);
  }

D
dapan1121 已提交
449
  tfree(params);
D
dapan1121 已提交
450
  SCL_RET(code);
D
dapan1121 已提交
451 452 453 454 455 456 457 458 459 460
}

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

  output->type = node->node.resType.type;
D
dapan1121 已提交
461
  output->num = rowNum;
D
dapan1121 已提交
462
  output->bytes = tDataTypes[output->type].bytes;
D
dapan1121 已提交
463
  output->data = calloc(rowNum, tDataTypes[output->type].bytes);
D
dapan1121 已提交
464
  if (NULL == output->data) {
D
dapan1121 已提交
465
    sclError("calloc %d failed", (int32_t)rowNum * tDataTypes[output->type].bytes);
D
dapan1121 已提交
466 467
    SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan 已提交
468
  output->orig.data = output->data;
D
dapan1121 已提交
469 470 471

  _bin_scalar_fn_t OperatorFn = getBinScalarOperatorFn(node->opType);

D
dapan1121 已提交
472
  int32_t paramNum = scalarGetOperatorParamNum(node->opType);
D
dapan1121 已提交
473 474
  SScalarParam* pLeft = &params[0];
  SScalarParam* pRight = paramNum > 1 ? &params[1] : NULL;
D
dapan1121 已提交
475
  
D
dapan 已提交
476
  OperatorFn(pLeft, pRight, output, TSDB_ORDER_ASC);
D
dapan1121 已提交
477 478 479

_return:

D
dapan1121 已提交
480 481 482 483 484

  for (int32_t i = 0; i < paramNum; ++i) {
    sclFreeParamNoData(params + i);
  }

D
dapan1121 已提交
485
  tfree(params);
D
dapan1121 已提交
486
  
D
dapan1121 已提交
487
  SCL_RET(code);
D
dapan1121 已提交
488 489 490
}


D
dapan 已提交
491
EDealRes sclRewriteFunction(SNode** pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
492 493 494
  SFunctionNode *node = (SFunctionNode *)*pNode;
  SScalarParam output = {0};
  
D
dapan 已提交
495 496
  ctx->code = sclExecFuncion(node, ctx, &output);
  if (ctx->code) {
D
dapan1121 已提交
497 498 499
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
500
  SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
D
dapan1121 已提交
501 502
  if (NULL == res) {
    sclError("make value node failed");
D
dapan1121 已提交
503
    sclFreeParam(&output);
D
dapan 已提交
504
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
505 506 507 508 509
    return DEAL_RES_ERROR;
  }

  res->node.resType = node->node.resType;

D
dapan1121 已提交
510 511 512 513 514 515 516
  if (IS_VAR_DATA_TYPE(output.type)) {
    res->datum.p = output.data;
    output.data = NULL;
  } else {
    memcpy(nodesGetValueFromNode(res), output.data, tDataTypes[output.type].bytes);
  }
  
D
dapan1121 已提交
517 518 519
  nodesDestroyNode(*pNode);
  *pNode = (SNode*)res;

D
dapan1121 已提交
520
  sclFreeParam(&output);
D
dapan1121 已提交
521 522 523 524

  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
525
EDealRes sclRewriteLogic(SNode** pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
526
  SLogicConditionNode *node = (SLogicConditionNode *)*pNode;
D
dapan1121 已提交
527
  SScalarParam output = {0};
D
dapan1121 已提交
528

D
dapan 已提交
529 530
  ctx->code = sclExecLogic(node, ctx, &output);
  if (ctx->code) {
D
dapan1121 已提交
531 532 533
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
534
  SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
D
dapan1121 已提交
535 536
  if (NULL == res) {
    sclError("make value node failed");
D
dapan1121 已提交
537
    sclFreeParam(&output);    
D
dapan 已提交
538
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
539 540 541 542 543
    return DEAL_RES_ERROR;
  }

  res->node.resType = node->node.resType;

D
dapan1121 已提交
544 545 546 547 548 549
  if (IS_VAR_DATA_TYPE(output.type)) {
    res->datum.p = output.data;
    output.data = NULL;
  } else {
    memcpy(nodesGetValueFromNode(res), output.data, tDataTypes[output.type].bytes);
  }
D
dapan1121 已提交
550 551 552 553

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

D
dapan1121 已提交
554 555
  sclFreeParam(&output);

D
dapan1121 已提交
556 557 558
  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
559
EDealRes sclRewriteOperator(SNode** pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
560 561
  SOperatorNode *node = (SOperatorNode *)*pNode;
  SScalarParam output = {0};
D
dapan1121 已提交
562

D
dapan 已提交
563 564
  ctx->code = sclExecOperator(node, ctx, &output);
  if (ctx->code) {
D
dapan1121 已提交
565 566 567
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
568
  SValueNode *res = (SValueNode *)nodesMakeNode(QUERY_NODE_VALUE);
D
dapan1121 已提交
569
  if (NULL == res) {
D
dapan1121 已提交
570 571
    sclError("make value node failed");    
    sclFreeParam(&output);    
D
dapan 已提交
572
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
573 574 575
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
576
  res->node.resType = node->node.resType;
D
dapan1121 已提交
577

D
dapan1121 已提交
578 579 580 581 582 583
  if (IS_VAR_DATA_TYPE(output.type)) {
    res->datum.p = output.data;
    output.data = NULL;
  } else {
    memcpy(nodesGetValueFromNode(res), output.data, tDataTypes[output.type].bytes);
  }
D
dapan1121 已提交
584 585 586 587

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

D
dapan1121 已提交
588 589
  sclFreeParam(&output);    

D
dapan1121 已提交
590 591 592 593 594
  return DEAL_RES_CONTINUE;
}


EDealRes sclConstantsRewriter(SNode** pNode, void* pContext) {
D
dapan1121 已提交
595
  if (QUERY_NODE_VALUE == nodeType(*pNode) || QUERY_NODE_NODE_LIST == nodeType(*pNode)) {
D
dapan1121 已提交
596 597 598
    return DEAL_RES_CONTINUE;
  }

D
dapan 已提交
599 600
  SScalarCtx *ctx = (SScalarCtx *)pContext;

D
dapan1121 已提交
601
  if (QUERY_NODE_FUNCTION == nodeType(*pNode)) {
D
dapan 已提交
602
    return sclRewriteFunction(pNode, ctx);
D
dapan1121 已提交
603 604 605
  }

  if (QUERY_NODE_LOGIC_CONDITION == nodeType(*pNode)) {
D
dapan 已提交
606
    return sclRewriteLogic(pNode, ctx);
D
dapan1121 已提交
607 608
  }

D
dapan1121 已提交
609
  if (QUERY_NODE_OPERATOR == nodeType(*pNode)) {
D
dapan 已提交
610
    return sclRewriteOperator(pNode, ctx);
D
dapan1121 已提交
611 612
  }  
  
D
dapan1121 已提交
613
  sclError("invalid node type for calculating constants, type:%d", nodeType(*pNode));
D
dapan 已提交
614 615

  ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
D
dapan1121 已提交
616 617
  
  return DEAL_RES_ERROR;
D
dapan1121 已提交
618 619
}

D
dapan1121 已提交
620

D
dapan 已提交
621
EDealRes sclWalkFunction(SNode* pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
622
  SFunctionNode *node = (SFunctionNode *)pNode;
D
dapan1121 已提交
623 624 625 626 627
  SScalarParam output = {0};
  
  ctx->code = sclExecFuncion(node, ctx, &output);
  if (ctx->code) {
    return DEAL_RES_ERROR;
D
dapan1121 已提交
628 629
  }

D
dapan1121 已提交
630
  if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
D
dapan1121 已提交
631 632
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
    return DEAL_RES_ERROR;
D
dapan1121 已提交
633 634
  }

D
dapan1121 已提交
635 636 637 638
  return DEAL_RES_CONTINUE;
}


D
dapan 已提交
639
EDealRes sclWalkLogic(SNode* pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
640
  SLogicConditionNode *node = (SLogicConditionNode *)pNode;
D
dapan1121 已提交
641 642 643 644 645
  SScalarParam output = {0};
  
  ctx->code = sclExecLogic(node, ctx, &output);
  if (ctx->code) {
    return DEAL_RES_ERROR;
D
dapan1121 已提交
646 647
  }

D
dapan1121 已提交
648
  if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
D
dapan1121 已提交
649
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
650
    return DEAL_RES_ERROR;
D
dapan1121 已提交
651 652 653 654 655 656
  }

  return DEAL_RES_CONTINUE;
}


D
dapan 已提交
657
EDealRes sclWalkOperator(SNode* pNode, SScalarCtx *ctx) {
D
dapan1121 已提交
658
  SOperatorNode *node = (SOperatorNode *)pNode;
D
dapan1121 已提交
659
  SScalarParam output = {0};
D
dapan1121 已提交
660
  
D
dapan1121 已提交
661 662
  ctx->code = sclExecOperator(node, ctx, &output);
  if (ctx->code) {
D
dapan1121 已提交
663 664
    return DEAL_RES_ERROR;
  }
D
dapan1121 已提交
665

D
dapan1121 已提交
666
  if (taosHashPut(ctx->pRes, &pNode, POINTER_BYTES, &output, sizeof(output))) {
D
dapan1121 已提交
667
    ctx->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
D
dapan1121 已提交
668 669 670
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
671 672 673
  return DEAL_RES_CONTINUE;
}

D
dapan 已提交
674 675 676
EDealRes sclWalkTarget(SNode* pNode, SScalarCtx *ctx) {
  STargetNode *target = (STargetNode *)pNode;
  
D
dapan1121 已提交
677 678
  if (target->dataBlockId >= taosArrayGetSize(ctx->pBlockList)) {
    sclError("target tupleId is too big, tupleId:%d, dataBlockNum:%d", target->dataBlockId, (int32_t)taosArrayGetSize(ctx->pBlockList));
D
dapan 已提交
679 680 681 682
    ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
    return DEAL_RES_ERROR;
  }

D
dapan1121 已提交
683
  SSDataBlock *block = *(SSDataBlock **)taosArrayGet(ctx->pBlockList, target->dataBlockId);
D
dapan 已提交
684
  if (target->slotId >= taosArrayGetSize(block->pDataBlock)) {
D
dapan1121 已提交
685
    sclError("target slot not exist, dataBlockId:%d, slotId:%d, dataBlockNum:%d", target->dataBlockId, target->slotId, (int32_t)taosArrayGetSize(block->pDataBlock));
D
dapan 已提交
686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706
    ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
    return DEAL_RES_ERROR;
  }

  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;
  }
  
  for (int32_t i = 0; i < res->num; ++i) {
    sclMoveParamListData(res, 1, i);

    colDataAppend(col, i, res->data, sclIsNull(res, i));
  }

  sclFreeParam(res);

D
dapan1121 已提交
707 708
  taosHashRemove(ctx->pRes, (void *)&target->pExpr, POINTER_BYTES);

D
dapan 已提交
709 710 711
  return DEAL_RES_CONTINUE;
}

D
dapan1121 已提交
712

D
dapan1121 已提交
713
EDealRes sclCalcWalker(SNode* pNode, void* pContext) {
X
Xiaoyu Wang 已提交
714
  if (QUERY_NODE_VALUE == nodeType(pNode) || QUERY_NODE_NODE_LIST == nodeType(pNode) || QUERY_NODE_COLUMN == nodeType(pNode)) {
D
dapan1121 已提交
715
    return DEAL_RES_CONTINUE;
D
dapan1121 已提交
716
  }
D
dapan 已提交
717 718

  SScalarCtx *ctx = (SScalarCtx *)pContext;
D
dapan1121 已提交
719
    
D
dapan1121 已提交
720
  if (QUERY_NODE_FUNCTION == nodeType(pNode)) {
D
dapan 已提交
721
    return sclWalkFunction(pNode, ctx);
D
dapan1121 已提交
722
  }
D
dapan1121 已提交
723

D
dapan1121 已提交
724
  if (QUERY_NODE_LOGIC_CONDITION == nodeType(pNode)) {
D
dapan 已提交
725
    return sclWalkLogic(pNode, ctx);
D
dapan1121 已提交
726
  }
D
dapan1121 已提交
727

D
dapan1121 已提交
728
  if (QUERY_NODE_OPERATOR == nodeType(pNode)) {
D
dapan 已提交
729
    return sclWalkOperator(pNode, ctx);
D
dapan1121 已提交
730
  }
D
dapan1121 已提交
731

D
dapan 已提交
732 733 734
  if (QUERY_NODE_TARGET == nodeType(pNode)) {
    return sclWalkTarget(pNode, ctx);
  }
D
dapan1121 已提交
735

D
dapan 已提交
736
  sclError("invalid node type for scalar calculating, type:%d", nodeType(pNode));
D
dapan1121 已提交
737 738 739 740
  
  ctx->code = TSDB_CODE_QRY_INVALID_INPUT;
  
  return DEAL_RES_ERROR;
D
dapan1121 已提交
741 742
}

D
dapan1121 已提交
743 744


D
dapan1121 已提交
745 746 747 748 749 750
int32_t scalarCalculateConstants(SNode *pNode, SNode **pRes) {
  if (NULL == pNode) {
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

  int32_t code = 0;
D
dapan 已提交
751 752 753 754 755 756
  SScalarCtx ctx = {0};
  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (NULL == ctx.pRes) {
    sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
757
  
D
dapan 已提交
758
  nodesRewriteNodePostOrder(&pNode, sclConstantsRewriter, (void *)&ctx);
D
dapan1121 已提交
759

D
dapan 已提交
760
  SCL_ERR_JRET(ctx.code);
D
dapan1121 已提交
761 762 763

  *pRes = pNode;

D
dapan 已提交
764 765 766 767 768
_return:
  
  sclFreeRes(ctx.pRes);

  return code;
D
dapan1121 已提交
769 770
}

D
dapan 已提交
771
int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
D
dapan1121 已提交
772
  if (NULL == pNode || NULL == pBlockList) {
D
dapan1121 已提交
773 774 775 776
    SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
  }

  int32_t code = 0;
D
dapan 已提交
777
  SScalarCtx ctx = {.code = 0, .pBlockList = pBlockList};
D
dapan1121 已提交
778 779 780 781 782 783
  
  ctx.pRes = taosHashInit(SCL_DEFAULT_OP_NUM, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (NULL == ctx.pRes) {
    sclError("taosHashInit failed, num:%d", SCL_DEFAULT_OP_NUM);
    SCL_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
  }
D
dapan1121 已提交
784
  
D
dapan1121 已提交
785
  nodesWalkNodePostOrder(pNode, sclCalcWalker, (void *)&ctx);
D
dapan1121 已提交
786

D
dapan 已提交
787
  SCL_ERR_JRET(ctx.code);
D
dapan1121 已提交
788

D
dapan1121 已提交
789 790 791 792 793 794 795
  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);
    }
    
D
dapan1121 已提交
796
    sclMoveParamListData(res, 1, 0);
D
dapan1121 已提交
797 798
    
    *pDst = *res;
D
dapan1121 已提交
799 800
    
    taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES);
D
dapan1121 已提交
801
  }
D
dapan1121 已提交
802

D
dapan 已提交
803 804
_return:
  
D
dapan1121 已提交
805
  //nodesDestroyNode(pNode);
D
dapan 已提交
806
  sclFreeRes(ctx.pRes);
D
dapan1121 已提交
807

D
dapan 已提交
808
  return code;
D
dapan1121 已提交
809 810 811 812
}