executorimpl.c 126.9 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
15

H
Haojun Liao 已提交
16 17
#include "filter.h"
#include "function.h"
18 19
#include "functionMgt.h"
#include "os.h"
H
Haojun Liao 已提交
20
#include "querynodes.h"
21
#include "tfill.h"
dengyihao's avatar
dengyihao 已提交
22
#include "tname.h"
X
Xiaoyu Wang 已提交
23
#include "tref.h"
24

H
Haojun Liao 已提交
25
#include "tdatablock.h"
26
#include "tglobal.h"
H
Haojun Liao 已提交
27
#include "tmsg.h"
H
Haojun Liao 已提交
28
#include "tsort.h"
29
#include "ttime.h"
H
Haojun Liao 已提交
30

31
#include "executorimpl.h"
dengyihao's avatar
dengyihao 已提交
32
#include "index.h"
33
#include "query.h"
34
#include "tcompare.h"
H
Haojun Liao 已提交
35
#include "thash.h"
36
#include "ttypes.h"
dengyihao's avatar
dengyihao 已提交
37
#include "vnode.h"
38

H
Haojun Liao 已提交
39
#define IS_MAIN_SCAN(runtime)          ((runtime)->scanFlag == MAIN_SCAN)
40 41 42 43 44 45
#define SET_REVERSE_SCAN_FLAG(runtime) ((runtime)->scanFlag = REVERSE_SCAN)

#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)

#if 0
static UNUSED_FUNC void *u_malloc (size_t __size) {
wafwerar's avatar
wafwerar 已提交
46
  uint32_t v = taosRand();
47 48 49 50

  if (v % 1000 <= 0) {
    return NULL;
  } else {
wafwerar's avatar
wafwerar 已提交
51
    return taosMemoryMalloc(__size);
52 53 54 55
  }
}

static UNUSED_FUNC void* u_calloc(size_t num, size_t __size) {
wafwerar's avatar
wafwerar 已提交
56
  uint32_t v = taosRand();
57 58 59
  if (v % 1000 <= 0) {
    return NULL;
  } else {
wafwerar's avatar
wafwerar 已提交
60
    return taosMemoryCalloc(num, __size);
61 62 63 64
  }
}

static UNUSED_FUNC void* u_realloc(void* p, size_t __size) {
wafwerar's avatar
wafwerar 已提交
65
  uint32_t v = taosRand();
66 67 68
  if (v % 5 <= 1) {
    return NULL;
  } else {
wafwerar's avatar
wafwerar 已提交
69
    return taosMemoryRealloc(p, __size);
70 71 72 73 74 75 76 77
  }
}

#define calloc  u_calloc
#define malloc  u_malloc
#define realloc u_realloc
#endif

X
Xiaoyu Wang 已提交
78
#define CLEAR_QUERY_STATUS(q, st)   ((q)->status &= (~(st)))
79 80
#define QUERY_IS_INTERVAL_QUERY(_q) ((_q)->interval.interval > 0)

L
Liu Jicong 已提交
81 82
int32_t getMaximumIdleDurationSec() { return tsShellActivityTimer * 2; }

83
static void setBlockSMAInfo(SqlFunctionCtx* pCtx, SExprInfo* pExpr, SSDataBlock* pBlock);
84

X
Xiaoyu Wang 已提交
85
static void releaseQueryBuf(size_t numOfTables);
86

87 88
static void destroyFillOperatorInfo(void* param);
static void destroyProjectOperatorInfo(void* param);
H
Haojun Liao 已提交
89
static void destroySortOperatorInfo(void* param);
90
static void destroyAggOperatorInfo(void* param);
X
Xiaoyu Wang 已提交
91

92
static void destroyIntervalOperatorInfo(void* param);
H
Haojun Liao 已提交
93

H
Haojun Liao 已提交
94
void setOperatorCompleted(SOperatorInfo* pOperator) {
95
  pOperator->status = OP_EXEC_DONE;
H
Haojun Liao 已提交
96
  ASSERT(pOperator->pTaskInfo != NULL);
97

98
  pOperator->cost.totalCost = (taosGetTimestampUs() - pOperator->pTaskInfo->cost.start * 1000) / 1000.0;
H
Haojun Liao 已提交
99
  setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
100
}
101

H
Haojun Liao 已提交
102 103 104 105 106 107 108 109 110 111
void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32_t type, bool blocking, int32_t status,
                     void* pInfo, SExecTaskInfo* pTaskInfo) {
  pOperator->name = (char*)name;
  pOperator->operatorType = type;
  pOperator->blocking = blocking;
  pOperator->status = status;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
}

H
Haojun Liao 已提交
112
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator) {
113
  OPTR_SET_OPENED(pOperator);
114
  pOperator->cost.openCost = 0;
H
Haojun Liao 已提交
115
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
116 117
}

H
Haojun Liao 已提交
118 119
SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t cleanup,
                                   __optr_close_fn_t closeFn, __optr_explain_fn_t explain) {
120 121 122 123 124 125 126 127 128 129 130
  SOperatorFpSet fpSet = {
      ._openFn = openFn,
      .getNextFn = nextFn,
      .cleanupFn = cleanup,
      .closeFn = closeFn,
      .getExplainFn = explain,
  };

  return fpSet;
}

131 132
static int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, SDiskbasedBuf* pBuf,
                                  SGroupResInfo* pGroupResInfo);
H
Haojun Liao 已提交
133

134
static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size);
135
static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId);
136

137
#if 0
L
Liu Jicong 已提交
138 139
static bool chkResultRowFromKey(STaskRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo, char* pData,
                                int16_t bytes, bool masterscan, uint64_t uid) {
140 141 142
  bool existed = false;
  SET_RES_WINDOW_KEY(pRuntimeEnv->keyBuf, pData, bytes, uid);

L
Liu Jicong 已提交
143 144
  SResultRow** p1 =
      (SResultRow**)taosHashGet(pRuntimeEnv->pResultRowHashTable, pRuntimeEnv->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
145 146 147 148 149 150 151 152 153 154 155

  // in case of repeat scan/reverse scan, no new time window added.
  if (QUERY_IS_INTERVAL_QUERY(pRuntimeEnv->pQueryAttr)) {
    if (!masterscan) {  // the *p1 may be NULL in case of sliding+offset exists.
      return p1 != NULL;
    }

    if (p1 != NULL) {
      if (pResultRowInfo->size == 0) {
        existed = false;
      } else if (pResultRowInfo->size == 1) {
dengyihao's avatar
dengyihao 已提交
156
        //        existed = (pResultRowInfo->pResult[0] == (*p1));
157 158
      } else {  // check if current pResultRowInfo contains the existed pResultRow
        SET_RES_EXT_WINDOW_KEY(pRuntimeEnv->keyBuf, pData, bytes, uid, pResultRowInfo);
L
Liu Jicong 已提交
159 160
        int64_t* index =
            taosHashGet(pRuntimeEnv->pResultRowListSet, pRuntimeEnv->keyBuf, GET_RES_EXT_WINDOW_KEY_LEN(bytes));
161 162 163 164 165 166 167 168 169 170 171 172 173
        if (index != NULL) {
          existed = true;
        } else {
          existed = false;
        }
      }
    }

    return existed;
  }

  return p1 != NULL;
}
174
#endif
175

176
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize) {
L
Liu Jicong 已提交
177
  SFilePage* pData = NULL;
178 179 180

  // in the first scan, new space needed for results
  int32_t pageId = -1;
181
  if (*currentPageId == -1) {
182
    pData = getNewBufPage(pResultBuf, &pageId);
183 184
    pData->num = sizeof(SFilePage);
  } else {
185 186
    pData = getBufPage(pResultBuf, *currentPageId);
    pageId = *currentPageId;
187

wmmhello's avatar
wmmhello 已提交
188
    if (pData->num + interBufSize > getBufPageSize(pResultBuf)) {
189
      // release current page first, and prepare the next one
190
      releaseBufPage(pResultBuf, pData);
191

192
      pData = getNewBufPage(pResultBuf, &pageId);
193 194 195 196 197 198 199 200 201 202
      if (pData != NULL) {
        pData->num = sizeof(SFilePage);
      }
    }
  }

  if (pData == NULL) {
    return NULL;
  }

203 204
  setBufPageDirty(pData, true);

205 206 207 208
  // set the number of rows in current disk page
  SResultRow* pResultRow = (SResultRow*)((char*)pData + pData->num);
  pResultRow->pageId = pageId;
  pResultRow->offset = (int32_t)pData->num;
209
  *currentPageId = pageId;
210

wmmhello's avatar
wmmhello 已提交
211
  pData->num += interBufSize;
212 213 214
  return pResultRow;
}

215 216 217 218 219 220 221
/**
 * the struct of key in hash table
 * +----------+---------------+
 * | group id |   key data    |
 * | 8 bytes  | actual length |
 * +----------+---------------+
 */
222 223 224
SResultRow* doSetResultOutBufByKey(SDiskbasedBuf* pResultBuf, SResultRowInfo* pResultRowInfo, char* pData,
                                   int16_t bytes, bool masterscan, uint64_t groupId, SExecTaskInfo* pTaskInfo,
                                   bool isIntervalQuery, SAggSupporter* pSup) {
225
  SET_RES_WINDOW_KEY(pSup->keyBuf, pData, bytes, groupId);
H
Haojun Liao 已提交
226

dengyihao's avatar
dengyihao 已提交
227
  SResultRowPosition* p1 =
228
      (SResultRowPosition*)tSimpleHashGet(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
H
Haojun Liao 已提交
229

230 231
  SResultRow* pResult = NULL;

H
Haojun Liao 已提交
232 233
  // in case of repeat scan/reverse scan, no new time window added.
  if (isIntervalQuery) {
234
    if (masterscan && p1 != NULL) {  // the *p1 may be NULL in case of sliding+offset exists.
235
      pResult = getResultRowByPos(pResultBuf, p1, true);
236
      ASSERT(pResult->pageId == p1->pageId && pResult->offset == p1->offset);
H
Haojun Liao 已提交
237 238
    }
  } else {
dengyihao's avatar
dengyihao 已提交
239 240
    // In case of group by column query, the required SResultRow object must be existInCurrentResusltRowInfo in the
    // pResultRowInfo object.
H
Haojun Liao 已提交
241
    if (p1 != NULL) {
242
      // todo
243
      pResult = getResultRowByPos(pResultBuf, p1, true);
244
      ASSERT(pResult->pageId == p1->pageId && pResult->offset == p1->offset);
H
Haojun Liao 已提交
245 246 247
    }
  }

L
Liu Jicong 已提交
248
  // 1. close current opened time window
249
  if (pResultRowInfo->cur.pageId != -1 && ((pResult == NULL) || (pResult->pageId != pResultRowInfo->cur.pageId))) {
250
    SResultRowPosition pos = pResultRowInfo->cur;
X
Xiaoyu Wang 已提交
251
    SFilePage*         pPage = getBufPage(pResultBuf, pos.pageId);
252 253 254 255 256
    releaseBufPage(pResultBuf, pPage);
  }

  // allocate a new buffer page
  if (pResult == NULL) {
H
Haojun Liao 已提交
257
    ASSERT(pSup->resultRowSize > 0);
258
    pResult = getNewResultRow(pResultBuf, &pSup->currentPageId, pSup->resultRowSize);
259

260 261
    // add a new result set for a new group
    SResultRowPosition pos = {.pageId = pResult->pageId, .offset = pResult->offset};
262
    tSimpleHashPut(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes), &pos,
L
Liu Jicong 已提交
263
                   sizeof(SResultRowPosition));
H
Haojun Liao 已提交
264 265
  }

266 267 268
  // 2. set the new time window to be the new active time window
  pResultRowInfo->cur = (SResultRowPosition){.pageId = pResult->pageId, .offset = pResult->offset};

H
Haojun Liao 已提交
269
  // too many time window in query
270
  if (pTaskInfo->execModel == OPTR_EXEC_MODEL_BATCH &&
271
      tSimpleHashGetSize(pSup->pResultRowHashTable) > MAX_INTERVAL_TIME_WINDOW) {
272
    T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW);
H
Haojun Liao 已提交
273 274
  }

H
Haojun Liao 已提交
275
  return pResult;
H
Haojun Liao 已提交
276 277
}

278
// a new buffer page for each table. Needs to opt this design
L
Liu Jicong 已提交
279
static int32_t addNewWindowResultBuf(SResultRow* pWindowRes, SDiskbasedBuf* pResultBuf, int32_t tid, uint32_t size) {
280 281 282 283
  if (pWindowRes->pageId != -1) {
    return 0;
  }

L
Liu Jicong 已提交
284
  SFilePage* pData = NULL;
285 286 287

  // in the first scan, new space needed for results
  int32_t pageId = -1;
288
  SIDList list = getDataBufPagesIdList(pResultBuf);
289 290

  if (taosArrayGetSize(list) == 0) {
291
    pData = getNewBufPage(pResultBuf, &pageId);
292
    pData->num = sizeof(SFilePage);
293 294
  } else {
    SPageInfo* pi = getLastPageInfo(list);
295
    pData = getBufPage(pResultBuf, getPageId(pi));
296
    pageId = getPageId(pi);
297

298
    if (pData->num + size > getBufPageSize(pResultBuf)) {
299
      // release current page first, and prepare the next one
300
      releaseBufPageInfo(pResultBuf, pi);
301

302
      pData = getNewBufPage(pResultBuf, &pageId);
303
      if (pData != NULL) {
304
        pData->num = sizeof(SFilePage);
305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324
      }
    }
  }

  if (pData == NULL) {
    return -1;
  }

  // set the number of rows in current disk page
  if (pWindowRes->pageId == -1) {  // not allocated yet, allocate new buffer
    pWindowRes->pageId = pageId;
    pWindowRes->offset = (int32_t)pData->num;

    pData->num += size;
    assert(pWindowRes->pageId >= 0);
  }

  return 0;
}

325
//  query_range_start, query_range_end, window_duration, window_start, window_end
326
void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow) {
327 328 329
  pColData->info.type = TSDB_DATA_TYPE_TIMESTAMP;
  pColData->info.bytes = sizeof(int64_t);

H
Haojun Liao 已提交
330
  colInfoDataEnsureCapacity(pColData, 5, false);
331 332 333 334 335 336 337 338 339
  colDataAppendInt64(pColData, 0, &pQueryWindow->skey);
  colDataAppendInt64(pColData, 1, &pQueryWindow->ekey);

  int64_t interval = 0;
  colDataAppendInt64(pColData, 2, &interval);  // this value may be variable in case of 'n' and 'y'.
  colDataAppendInt64(pColData, 3, &pQueryWindow->skey);
  colDataAppendInt64(pColData, 4, &pQueryWindow->ekey);
}

L
Liu Jicong 已提交
340
void cleanupExecTimeWindowInfo(SColumnInfoData* pColData) { colDataDestroy(pColData); }
H
Haojun Liao 已提交
341

342 343 344 345 346 347 348 349 350 351 352 353 354 355
typedef struct {
  bool    hasAgg;
  int32_t numOfRows;
  int32_t startOffset;
} SFunctionCtxStatus;

static void functionCtxSave(SqlFunctionCtx* pCtx, SFunctionCtxStatus* pStatus) {
  pStatus->hasAgg = pCtx->input.colDataAggIsSet;
  pStatus->numOfRows = pCtx->input.numOfRows;
  pStatus->startOffset = pCtx->input.startRowIndex;
}

static void functionCtxRestore(SqlFunctionCtx* pCtx, SFunctionCtxStatus* pStatus) {
  pCtx->input.colDataAggIsSet = pStatus->hasAgg;
H
Haojun Liao 已提交
356
  pCtx->input.numOfRows = pStatus->numOfRows;
357 358 359 360 361
  pCtx->input.startRowIndex = pStatus->startOffset;
}

void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData, int32_t offset,
                      int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput) {
362
  for (int32_t k = 0; k < numOfOutput; ++k) {
H
Haojun Liao 已提交
363
    // keep it temporarily
364 365
    SFunctionCtxStatus status = {0};
    functionCtxSave(&pCtx[k], &status);
366

367
    pCtx[k].input.startRowIndex = offset;
368
    pCtx[k].input.numOfRows = forwardStep;
369 370 371

    // not a whole block involved in query processing, statistics data can not be used
    // NOTE: the original value of isSet have been changed here
372 373
    if (pCtx[k].input.colDataAggIsSet && forwardStep < numOfTotal) {
      pCtx[k].input.colDataAggIsSet = false;
374 375
    }

376 377
    if (fmIsWindowPseudoColumnFunc(pCtx[k].functionId)) {
      SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pCtx[k]);
378 379

      char* p = GET_ROWCELL_INTERBUF(pEntryInfo);
380

381
      SColumnInfoData idata = {0};
dengyihao's avatar
dengyihao 已提交
382
      idata.info.type = TSDB_DATA_TYPE_BIGINT;
383
      idata.info.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes;
dengyihao's avatar
dengyihao 已提交
384
      idata.pData = p;
385 386 387 388

      SScalarParam out = {.columnData = &idata};
      SScalarParam tw = {.numOfRows = 5, .columnData = pTimeWindowData};
      pCtx[k].sfp.process(&tw, 1, &out);
389
      pEntryInfo->numOfRes = 1;
390 391 392 393 394 395 396 397
    } else {
      int32_t code = TSDB_CODE_SUCCESS;
      if (functionNeedToExecute(&pCtx[k]) && pCtx[k].fpSet.process != NULL) {
        code = pCtx[k].fpSet.process(&pCtx[k]);

        if (code != TSDB_CODE_SUCCESS) {
          qError("%s apply functions error, code: %s", GET_TASKID(taskInfo), tstrerror(code));
          taskInfo->code = code;
398
          T_LONG_JMP(taskInfo->env, code);
399
        }
400
      }
401

402
      // restore it
403
      functionCtxRestore(&pCtx[k], &status);
404
    }
405 406 407
  }
}

408 409
static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
                                   bool createDummyCol);
410

411 412 413
static void doSetInputDataBlockInfo(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order) {
  SqlFunctionCtx* pCtx = pExprSup->pCtx;
  for (int32_t i = 0; i < pExprSup->numOfExprs; ++i) {
414
    pCtx[i].order = order;
415
    pCtx[i].input.numOfRows = pBlock->info.rows;
416
    setBlockSMAInfo(&pCtx[i], &pExprSup->pExprInfo[i], pBlock);
417
    pCtx[i].pSrcBlock = pBlock;
418 419 420
  }
}

421
void setInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol) {
422
  if (pBlock->pBlockAgg != NULL) {
423
    doSetInputDataBlockInfo(pExprSup, pBlock, order);
424
  } else {
425
    doSetInputDataBlock(pExprSup, pBlock, order, scanFlag, createDummyCol);
H
Haojun Liao 已提交
426
  }
427 428
}

L
Liu Jicong 已提交
429 430
static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunctParam* pFuncParam, int32_t paramIndex,
                                             int32_t numOfRows) {
431 432 433 434 435 436 437 438
  SColumnInfoData* pColInfo = NULL;
  if (pInput->pData[paramIndex] == NULL) {
    pColInfo = taosMemoryCalloc(1, sizeof(SColumnInfoData));
    if (pColInfo == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }

    // Set the correct column info (data type and bytes)
439 440
    pColInfo->info.type = pFuncParam->param.nType;
    pColInfo->info.bytes = pFuncParam->param.nLen;
441 442

    pInput->pData[paramIndex] = pColInfo;
443 444
  } else {
    pColInfo = pInput->pData[paramIndex];
445 446
  }

H
Haojun Liao 已提交
447
  colInfoDataEnsureCapacity(pColInfo, numOfRows, false);
448

449
  int8_t type = pFuncParam->param.nType;
450 451
  if (type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_UBIGINT) {
    int64_t v = pFuncParam->param.i;
dengyihao's avatar
dengyihao 已提交
452
    for (int32_t i = 0; i < numOfRows; ++i) {
453 454 455 456
      colDataAppendInt64(pColInfo, i, &v);
    }
  } else if (type == TSDB_DATA_TYPE_DOUBLE) {
    double v = pFuncParam->param.d;
dengyihao's avatar
dengyihao 已提交
457
    for (int32_t i = 0; i < numOfRows; ++i) {
458 459
      colDataAppendDouble(pColInfo, i, &v);
    }
460
  } else if (type == TSDB_DATA_TYPE_VARCHAR) {
L
Liu Jicong 已提交
461
    char* tmp = taosMemoryMalloc(pFuncParam->param.nLen + VARSTR_HEADER_SIZE);
462
    STR_WITH_SIZE_TO_VARSTR(tmp, pFuncParam->param.pz, pFuncParam->param.nLen);
L
Liu Jicong 已提交
463
    for (int32_t i = 0; i < numOfRows; ++i) {
464 465
      colDataAppend(pColInfo, i, tmp, false);
    }
H
Haojun Liao 已提交
466
    taosMemoryFree(tmp);
467 468 469 470 471
  }

  return TSDB_CODE_SUCCESS;
}

472 473
static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
                                   bool createDummyCol) {
474
  int32_t         code = TSDB_CODE_SUCCESS;
475
  SqlFunctionCtx* pCtx = pExprSup->pCtx;
476

477
  for (int32_t i = 0; i < pExprSup->numOfExprs; ++i) {
L
Liu Jicong 已提交
478
    pCtx[i].order = order;
479 480
    pCtx[i].input.numOfRows = pBlock->info.rows;

L
Liu Jicong 已提交
481
    pCtx[i].pSrcBlock = pBlock;
X
Xiaoyu Wang 已提交
482
    pCtx[i].scanFlag = scanFlag;
H
Haojun Liao 已提交
483

484
    SInputColumnInfoData* pInput = &pCtx[i].input;
485
    pInput->uid = pBlock->info.uid;
C
Cary Xu 已提交
486
    pInput->colDataAggIsSet = false;
487

488
    SExprInfo* pOneExpr = &pExprSup->pExprInfo[i];
489
    for (int32_t j = 0; j < pOneExpr->base.numOfParams; ++j) {
dengyihao's avatar
dengyihao 已提交
490
      SFunctParam* pFuncParam = &pOneExpr->base.pParam[j];
G
Ganlin Zhao 已提交
491 492
      if (pFuncParam->type == FUNC_PARAM_TYPE_COLUMN) {
        int32_t slotId = pFuncParam->pCol->slotId;
dengyihao's avatar
dengyihao 已提交
493
        pInput->pData[j] = taosArrayGet(pBlock->pDataBlock, slotId);
494 495 496
        pInput->totalRows = pBlock->info.rows;
        pInput->numOfRows = pBlock->info.rows;
        pInput->startRowIndex = 0;
497

498
        // NOTE: the last parameter is the primary timestamp column
H
Haojun Liao 已提交
499
        // todo: refactor this
500
        if (fmIsImplicitTsFunc(pCtx[i].functionId) && (j == pOneExpr->base.numOfParams - 1)) {
L
Liu Jicong 已提交
501
          pInput->pPTS = pInput->pData[j];  // in case of merge function, this is not always the ts column data.
502
          //          ASSERT(pInput->pPTS->info.type == TSDB_DATA_TYPE_TIMESTAMP);
503
        }
504 505
        ASSERT(pInput->pData[j] != NULL);
      } else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) {
506 507 508
        // todo avoid case: top(k, 12), 12 is the value parameter.
        // sum(11), 11 is also the value parameter.
        if (createDummyCol && pOneExpr->base.numOfParams == 1) {
509 510 511 512
          pInput->totalRows = pBlock->info.rows;
          pInput->numOfRows = pBlock->info.rows;
          pInput->startRowIndex = 0;

513
          code = doCreateConstantValColumnInfo(pInput, pFuncParam, j, pBlock->info.rows);
514 515 516
          if (code != TSDB_CODE_SUCCESS) {
            return code;
          }
517
        }
G
Ganlin Zhao 已提交
518 519
      }
    }
H
Haojun Liao 已提交
520
  }
521 522

  return code;
H
Haojun Liao 已提交
523 524
}

525
static int32_t doAggregateImpl(SOperatorInfo* pOperator, SqlFunctionCtx* pCtx) {
526
  for (int32_t k = 0; k < pOperator->exprSupp.numOfExprs; ++k) {
H
Haojun Liao 已提交
527
    if (functionNeedToExecute(&pCtx[k])) {
528
      // todo add a dummy funtion to avoid process check
529 530 531
      if (pCtx[k].fpSet.process == NULL) {
        continue;
      }
H
Haojun Liao 已提交
532

533 534 535 536
      int32_t code = pCtx[k].fpSet.process(&pCtx[k]);
      if (code != TSDB_CODE_SUCCESS) {
        qError("%s aggregate function error happens, code: %s", GET_TASKID(pOperator->pTaskInfo), tstrerror(code));
        return code;
537
      }
538 539
    }
  }
540 541

  return TSDB_CODE_SUCCESS;
542 543
}

H
Haojun Liao 已提交
544
static void setPseudoOutputColInfo(SSDataBlock* pResult, SqlFunctionCtx* pCtx, SArray* pPseudoList) {
dengyihao's avatar
dengyihao 已提交
545
  size_t num = (pPseudoList != NULL) ? taosArrayGetSize(pPseudoList) : 0;
H
Haojun Liao 已提交
546 547 548 549 550
  for (int32_t i = 0; i < num; ++i) {
    pCtx[i].pOutput = taosArrayGet(pResult->pDataBlock, i);
  }
}

551
int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBlock* pSrcBlock, SqlFunctionCtx* pCtx,
X
Xiaoyu Wang 已提交
552
                              int32_t numOfOutput, SArray* pPseudoList) {
H
Haojun Liao 已提交
553
  setPseudoOutputColInfo(pResult, pCtx, pPseudoList);
554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573

  if (pSrcBlock == NULL) {
    for (int32_t k = 0; k < numOfOutput; ++k) {
      int32_t outputSlotId = pExpr[k].base.resSchema.slotId;

      ASSERT(pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE);
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);

      int32_t type = pExpr[k].base.pParam[0].param.nType;
      if (TSDB_DATA_TYPE_NULL == type) {
        colDataAppendNNULL(pColInfoData, 0, 1);
      } else {
        colDataAppend(pColInfoData, 0, taosVariantGet(&pExpr[k].base.pParam[0].param, type), false);
      }
    }

    pResult->info.rows = 1;
    return TSDB_CODE_SUCCESS;
  }

L
Liu Jicong 已提交
574 575 576 577
  if (pResult != pSrcBlock) {
    pResult->info.groupId = pSrcBlock->info.groupId;
    memcpy(pResult->info.parTbName, pSrcBlock->info.parTbName, TSDB_TABLE_NAME_LEN);
  }
H
Haojun Liao 已提交
578

579 580
  // if the source equals to the destination, it is to create a new column as the result of scalar
  // function or some operators.
581
  bool createNewColModel = (pResult == pSrcBlock);
582 583 584
  if (createNewColModel) {
    blockDataEnsureCapacity(pResult, pResult->info.rows);
  }
585

586 587
  int32_t numOfRows = 0;

588
  for (int32_t k = 0; k < numOfOutput; ++k) {
589 590
    int32_t               outputSlotId = pExpr[k].base.resSchema.slotId;
    SqlFunctionCtx*       pfCtx = &pCtx[k];
591
    SInputColumnInfoData* pInputData = &pfCtx->input;
592

L
Liu Jicong 已提交
593
    if (pExpr[k].pExpr->nodeType == QUERY_NODE_COLUMN) {  // it is a project query
594
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
595
      if (pResult->info.rows > 0 && !createNewColModel) {
596
        colDataMergeCol(pColInfoData, pResult->info.rows, (int32_t*)&pResult->info.capacity, pInputData->pData[0],
597
                        pInputData->numOfRows);
598
      } else {
599
        colDataAssign(pColInfoData, pInputData->pData[0], pInputData->numOfRows, &pResult->info);
600
      }
601

602
      numOfRows = pInputData->numOfRows;
603
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE) {
604
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
605

dengyihao's avatar
dengyihao 已提交
606
      int32_t offset = createNewColModel ? 0 : pResult->info.rows;
607 608 609 610 611 612 613 614

      int32_t type = pExpr[k].base.pParam[0].param.nType;
      if (TSDB_DATA_TYPE_NULL == type) {
        colDataAppendNNULL(pColInfoData, offset, pSrcBlock->info.rows);
      } else {
        for (int32_t i = 0; i < pSrcBlock->info.rows; ++i) {
          colDataAppend(pColInfoData, i + offset, taosVariantGet(&pExpr[k].base.pParam[0].param, type), false);
        }
615
      }
616 617

      numOfRows = pSrcBlock->info.rows;
H
Haojun Liao 已提交
618
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_OPERATOR) {
619 620 621
      SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
      taosArrayPush(pBlockList, &pSrcBlock);

622
      SColumnInfoData* pResColData = taosArrayGet(pResult->pDataBlock, outputSlotId);
623
      SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
624

625
      SScalarParam dest = {.columnData = &idata};
X
Xiaoyu Wang 已提交
626
      int32_t      code = scalarCalculate(pExpr[k].pExpr->_optrRoot.pRootNode, pBlockList, &dest);
627 628 629 630
      if (code != TSDB_CODE_SUCCESS) {
        taosArrayDestroy(pBlockList);
        return code;
      }
631

dengyihao's avatar
dengyihao 已提交
632
      int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
633
      ASSERT(pResult->info.capacity > 0);
634

635
      colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
D
dapan1121 已提交
636
      colDataDestroy(&idata);
L
Liu Jicong 已提交
637

638
      numOfRows = dest.numOfRows;
639 640
      taosArrayDestroy(pBlockList);
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
641 642
      // _rowts/_c0, not tbname column
      if (fmIsPseudoColumnFunc(pfCtx->functionId) && (!fmIsScanPseudoColumnFunc(pfCtx->functionId))) {
H
Haojun Liao 已提交
643
        // do nothing
644
      } else if (fmIsIndefiniteRowsFunc(pfCtx->functionId)) {
645 646
        SResultRowEntryInfo* pResInfo = GET_RES_INFO(pfCtx);
        pfCtx->fpSet.init(pfCtx, pResInfo);
647 648 649 650 651 652 653 654 655 656

        pfCtx->pOutput = taosArrayGet(pResult->pDataBlock, outputSlotId);
        pfCtx->offset = createNewColModel ? 0 : pResult->info.rows;  // set the start offset

        // set the timestamp(_rowts) output buffer
        if (taosArrayGetSize(pPseudoList) > 0) {
          int32_t* outputColIndex = taosArrayGet(pPseudoList, 0);
          pfCtx->pTsOutput = (SColumnInfoData*)pCtx[*outputColIndex].pOutput;
        }

657 658 659 660 661
        // link pDstBlock to set selectivity value
        if (pfCtx->subsidiaries.num > 0) {
          pfCtx->pDstBlock = pResult;
        }

662 663 664 665 666
        int32_t code = pfCtx->fpSet.process(pfCtx);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
        numOfRows = pResInfo->numOfRes;
H
Haojun Liao 已提交
667
      } else if (fmIsAggFunc(pfCtx->functionId)) {
G
Ganlin Zhao 已提交
668
        // selective value output should be set during corresponding function execution
669 670 671
        if (fmIsSelectValueFunc(pfCtx->functionId)) {
          continue;
        }
672 673
        // _group_key function for "partition by tbname" + csum(col_name) query
        SColumnInfoData* pOutput = taosArrayGet(pResult->pDataBlock, outputSlotId);
674
        int32_t          slotId = pfCtx->param[0].pCol->slotId;
675 676 677

        // todo handle the json tag
        SColumnInfoData* pInput = taosArrayGet(pSrcBlock->pDataBlock, slotId);
678
        for (int32_t f = 0; f < pSrcBlock->info.rows; ++f) {
679 680 681 682 683 684 685 686 687
          bool isNull = colDataIsNull_s(pInput, f);
          if (isNull) {
            colDataAppendNULL(pOutput, pResult->info.rows + f);
          } else {
            char* data = colDataGetData(pInput, f);
            colDataAppend(pOutput, pResult->info.rows + f, data, isNull);
          }
        }

H
Haojun Liao 已提交
688 689 690
      } else {
        SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
        taosArrayPush(pBlockList, &pSrcBlock);
G
Ganlin Zhao 已提交
691

692
        SColumnInfoData* pResColData = taosArrayGet(pResult->pDataBlock, outputSlotId);
693
        SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
H
Haojun Liao 已提交
694

695
        SScalarParam dest = {.columnData = &idata};
X
Xiaoyu Wang 已提交
696
        int32_t      code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
697 698 699 700
        if (code != TSDB_CODE_SUCCESS) {
          taosArrayDestroy(pBlockList);
          return code;
        }
701

dengyihao's avatar
dengyihao 已提交
702
        int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
703
        ASSERT(pResult->info.capacity > 0);
704
        colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
D
dapan1121 已提交
705
        colDataDestroy(&idata);
706 707

        numOfRows = dest.numOfRows;
H
Haojun Liao 已提交
708 709
        taosArrayDestroy(pBlockList);
      }
710
    } else {
711
      return TSDB_CODE_OPS_NOT_SUPPORT;
712 713
    }
  }
714

715 716 717
  if (!createNewColModel) {
    pResult->info.rows += numOfRows;
  }
718 719

  return TSDB_CODE_SUCCESS;
720 721
}

5
54liuyao 已提交
722
bool functionNeedToExecute(SqlFunctionCtx* pCtx) {
723
  struct SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
724

725 726 727 728 729
  // in case of timestamp column, always generated results.
  int32_t functionId = pCtx->functionId;
  if (functionId == -1) {
    return false;
  }
730

731 732
  if (pCtx->scanFlag == REPEAT_SCAN) {
    return fmIsRepeatScanFunc(pCtx->functionId);
733 734
  }

735 736
  if (isRowEntryCompleted(pResInfo)) {
    return false;
737 738
  }

739 740 741
  return true;
}

742 743 744 745 746 747 748
static int32_t doCreateConstantValColumnAggInfo(SInputColumnInfoData* pInput, SFunctParam* pFuncParam, int32_t type,
                                                int32_t paramIndex, int32_t numOfRows) {
  if (pInput->pData[paramIndex] == NULL) {
    pInput->pData[paramIndex] = taosMemoryCalloc(1, sizeof(SColumnInfoData));
    if (pInput->pData[paramIndex] == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
749

750 751 752
    // Set the correct column info (data type and bytes)
    pInput->pData[paramIndex]->info.type = type;
    pInput->pData[paramIndex]->info.bytes = tDataTypes[type].bytes;
753
  }
H
Haojun Liao 已提交
754

755 756 757 758 759 760
  SColumnDataAgg* da = NULL;
  if (pInput->pColumnDataAgg[paramIndex] == NULL) {
    da = taosMemoryCalloc(1, sizeof(SColumnDataAgg));
    pInput->pColumnDataAgg[paramIndex] = da;
    if (da == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
761 762
    }
  } else {
763
    da = pInput->pColumnDataAgg[paramIndex];
764 765
  }

766
  ASSERT(!IS_VAR_DATA_TYPE(type));
767

768 769
  if (type == TSDB_DATA_TYPE_BIGINT) {
    int64_t v = pFuncParam->param.i;
770
    *da = (SColumnDataAgg){.numOfNull = 0, .min = v, .max = v, .sum = v * numOfRows};
771 772
  } else if (type == TSDB_DATA_TYPE_DOUBLE) {
    double v = pFuncParam->param.d;
773
    *da = (SColumnDataAgg){.numOfNull = 0};
774

775 776 777 778 779 780
    *(double*)&da->min = v;
    *(double*)&da->max = v;
    *(double*)&da->sum = v * numOfRows;
  } else if (type == TSDB_DATA_TYPE_BOOL) {  // todo validate this data type
    bool v = pFuncParam->param.i;

781
    *da = (SColumnDataAgg){.numOfNull = 0};
782 783 784 785 786
    *(bool*)&da->min = 0;
    *(bool*)&da->max = v;
    *(bool*)&da->sum = v * numOfRows;
  } else if (type == TSDB_DATA_TYPE_TIMESTAMP) {
    // do nothing
787
  } else {
788
    ASSERT(0);
789 790
  }

791 792
  return TSDB_CODE_SUCCESS;
}
793

794
void setBlockSMAInfo(SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, SSDataBlock* pBlock) {
795 796 797 798 799 800 801 802 803
  int32_t numOfRows = pBlock->info.rows;

  SInputColumnInfoData* pInput = &pCtx->input;
  pInput->numOfRows = numOfRows;
  pInput->totalRows = numOfRows;

  if (pBlock->pBlockAgg != NULL) {
    pInput->colDataAggIsSet = true;

804 805
    for (int32_t j = 0; j < pExprInfo->base.numOfParams; ++j) {
      SFunctParam* pFuncParam = &pExprInfo->base.pParam[j];
806

807 808
      if (pFuncParam->type == FUNC_PARAM_TYPE_COLUMN) {
        int32_t slotId = pFuncParam->pCol->slotId;
809 810 811 812
        pInput->pColumnDataAgg[j] = pBlock->pBlockAgg[slotId];
        if (pInput->pColumnDataAgg[j] == NULL) {
          pInput->colDataAggIsSet = false;
        }
813 814 815 816

        // Here we set the column info data since the data type for each column data is required, but
        // the data in the corresponding SColumnInfoData will not be used.
        pInput->pData[j] = taosArrayGet(pBlock->pDataBlock, slotId);
817 818
      } else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) {
        doCreateConstantValColumnAggInfo(pInput, pFuncParam, pFuncParam->param.nType, j, pBlock->info.rows);
819 820
      }
    }
821
  } else {
822
    pInput->colDataAggIsSet = false;
823 824 825
  }
}

L
Liu Jicong 已提交
826
bool isTaskKilled(SExecTaskInfo* pTaskInfo) {
827 828
  // query has been executed more than tsShellActivityTimer, and the retrieve has not arrived
  // abort current query execution.
L
Liu Jicong 已提交
829 830
  if (pTaskInfo->owner != 0 &&
      ((taosGetTimestampSec() - pTaskInfo->cost.start / 1000) > 10 * getMaximumIdleDurationSec())
831
      /*(!needBuildResAfterQueryComplete(pTaskInfo))*/) {
832
    assert(pTaskInfo->cost.start != 0);
L
Liu Jicong 已提交
833 834 835
    //    qDebug("QInfo:%" PRIu64 " retrieve not arrive beyond %d ms, abort current query execution, start:%" PRId64
    //           ", current:%d", pQInfo->qId, 1, pQInfo->startExecTs, taosGetTimestampSec());
    //    return true;
836 837 838 839 840
  }

  return false;
}

L
Liu Jicong 已提交
841
void setTaskKilled(SExecTaskInfo* pTaskInfo) { pTaskInfo->code = TSDB_CODE_TSC_QUERY_CANCELLED; }
842 843

/////////////////////////////////////////////////////////////////////////////////////////////
844
STimeWindow getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key) {
L
Liu Jicong 已提交
845
  STimeWindow win = {0};
846
  win.skey = taosTimeTruncate(key, pInterval, precision);
847 848

  /*
H
Haojun Liao 已提交
849
   * if the realSkey > INT64_MAX - pInterval->interval, the query duration between
850 851
   * realSkey and realEkey must be less than one interval.Therefore, no need to adjust the query ranges.
   */
852 853 854
  win.ekey = taosTimeAdd(win.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
  if (win.ekey < win.skey) {
    win.ekey = INT64_MAX;
855
  }
856 857

  return win;
858 859
}

L
Liu Jicong 已提交
860 861
int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
                              uint32_t* status) {
862
  *status = BLK_DATA_NOT_LOAD;
863

H
Haojun Liao 已提交
864
  pBlock->pDataBlock = NULL;
L
Liu Jicong 已提交
865
  pBlock->pBlockAgg = NULL;
H
Haojun Liao 已提交
866

L
Liu Jicong 已提交
867 868
  //  int64_t groupId = pRuntimeEnv->current->groupIndex;
  //  bool    ascQuery = QUERY_IS_ASC_QUERY(pQueryAttr);
869

H
Haojun Liao 已提交
870
  STaskCostInfo* pCost = &pTaskInfo->cost;
871

872 873
//  pCost->totalBlocks += 1;
//  pCost->totalRows += pBlock->info.rows;
H
Haojun Liao 已提交
874
#if 0
875 876 877
  // Calculate all time windows that are overlapping or contain current data block.
  // If current data block is contained by all possible time window, do not load current data block.
  if (/*pQueryAttr->pFilters || */pQueryAttr->groupbyColumn || pQueryAttr->sw.gap > 0 ||
H
Haojun Liao 已提交
878
      (QUERY_IS_INTERVAL_QUERY(pQueryAttr) && overlapWithTimeWindow(pTaskInfo, &pBlock->info))) {
879
    (*status) = BLK_DATA_DATA_LOAD;
880 881 882
  }

  // check if this data block is required to load
883
  if ((*status) != BLK_DATA_DATA_LOAD) {
884 885 886 887 888 889 890
    bool needFilter = true;

    // the pCtx[i] result is belonged to previous time window since the outputBuf has not been set yet,
    // the filter result may be incorrect. So in case of interval query, we need to set the correct time output buffer
    if (QUERY_IS_INTERVAL_QUERY(pQueryAttr)) {
      SResultRow* pResult = NULL;

H
Haojun Liao 已提交
891
      bool  masterScan = IS_MAIN_SCAN(pRuntimeEnv);
892 893 894 895 896 897
      TSKEY k = ascQuery? pBlock->info.window.skey : pBlock->info.window.ekey;

      STimeWindow win = getActiveTimeWindow(pTableScanInfo->pResultRowInfo, k, pQueryAttr);
      if (pQueryAttr->pointInterpQuery) {
        needFilter = chkWindowOutputBufByKey(pRuntimeEnv, pTableScanInfo->pResultRowInfo, &win, masterScan, &pResult, groupId,
                                    pTableScanInfo->pCtx, pTableScanInfo->numOfOutput,
898
                                    pTableScanInfo->rowEntryInfoOffset);
899 900 901
      } else {
        if (setResultOutputBufByKey(pRuntimeEnv, pTableScanInfo->pResultRowInfo, pBlock->info.uid, &win, masterScan, &pResult, groupId,
                                    pTableScanInfo->pCtx, pTableScanInfo->numOfOutput,
902
                                    pTableScanInfo->rowEntryInfoOffset) != TSDB_CODE_SUCCESS) {
903
          T_LONG_JMP(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
904 905 906 907
        }
      }
    } else if (pQueryAttr->stableQuery && (!pQueryAttr->tsCompQuery) && (!pQueryAttr->diffQuery)) { // stable aggregate, not interval aggregate or normal column aggregate
      doSetTableGroupOutputBuf(pRuntimeEnv, pTableScanInfo->pResultRowInfo, pTableScanInfo->pCtx,
908
                               pTableScanInfo->rowEntryInfoOffset, pTableScanInfo->numOfOutput,
909 910 911 912 913 914
                               pRuntimeEnv->current->groupIndex);
    }

    if (needFilter) {
      (*status) = doFilterByBlockTimeWindow(pTableScanInfo, pBlock);
    } else {
915
      (*status) = BLK_DATA_DATA_LOAD;
916 917 918 919
    }
  }

  SDataBlockInfo* pBlockInfo = &pBlock->info;
H
Haojun Liao 已提交
920
//  *status = updateBlockLoadStatus(pRuntimeEnv->pQueryAttr, *status);
921

922
  if ((*status) == BLK_DATA_NOT_LOAD || (*status) == BLK_DATA_FILTEROUT) {
923 924
    //qDebug("QInfo:0x%"PRIx64" data block discard, brange:%" PRId64 "-%" PRId64 ", rows:%d", pQInfo->qId, pBlockInfo->window.skey,
//           pBlockInfo->window.ekey, pBlockInfo->rows);
925
    pCost->skipBlocks += 1;
926
  } else if ((*status) == BLK_DATA_SMA_LOAD) {
927 928
    // this function never returns error?
    pCost->loadBlockStatis += 1;
929
//    tsdbRetrieveDatablockSMA(pTableScanInfo->pTsdbReadHandle, &pBlock->pBlockAgg);
930 931

    if (pBlock->pBlockAgg == NULL) {  // data block statistics does not exist, load data block
932
//      pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pTsdbReadHandle, NULL);
933 934 935
      pCost->totalCheckedRows += pBlock->info.rows;
    }
  } else {
936
    assert((*status) == BLK_DATA_DATA_LOAD);
937 938 939

    // load the data block statistics to perform further filter
    pCost->loadBlockStatis += 1;
940
//    tsdbRetrieveDatablockSMA(pTableScanInfo->pTsdbReadHandle, &pBlock->pBlockAgg);
941 942 943 944 945 946

    if (pQueryAttr->topBotQuery && pBlock->pBlockAgg != NULL) {
      { // set previous window
        if (QUERY_IS_INTERVAL_QUERY(pQueryAttr)) {
          SResultRow* pResult = NULL;

H
Haojun Liao 已提交
947
          bool  masterScan = IS_MAIN_SCAN(pRuntimeEnv);
948 949 950 951 952
          TSKEY k = ascQuery? pBlock->info.window.skey : pBlock->info.window.ekey;

          STimeWindow win = getActiveTimeWindow(pTableScanInfo->pResultRowInfo, k, pQueryAttr);
          if (setResultOutputBufByKey(pRuntimeEnv, pTableScanInfo->pResultRowInfo, pBlock->info.uid, &win, masterScan, &pResult, groupId,
                                      pTableScanInfo->pCtx, pTableScanInfo->numOfOutput,
953
                                      pTableScanInfo->rowEntryInfoOffset) != TSDB_CODE_SUCCESS) {
954
            T_LONG_JMP(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
955 956 957 958 959 960 961 962 963 964
          }
        }
      }
      bool load = false;
      for (int32_t i = 0; i < pQueryAttr->numOfOutput; ++i) {
        int32_t functionId = pTableScanInfo->pCtx[i].functionId;
        if (functionId == FUNCTION_TOP || functionId == FUNCTION_BOTTOM) {
//          load = topbot_datablock_filter(&pTableScanInfo->pCtx[i], (char*)&(pBlock->pBlockAgg[i].min),
//                                         (char*)&(pBlock->pBlockAgg[i].max));
          if (!load) { // current block has been discard due to filter applied
965
            pCost->skipBlocks += 1;
966 967
            //qDebug("QInfo:0x%"PRIx64" data block discard, brange:%" PRId64 "-%" PRId64 ", rows:%d", pQInfo->qId,
//                   pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
968
            (*status) = BLK_DATA_FILTEROUT;
969 970 971 972 973 974 975
            return TSDB_CODE_SUCCESS;
          }
        }
      }
    }

    // current block has been discard due to filter applied
H
Haojun Liao 已提交
976
//    if (!doFilterByBlockSMA(pRuntimeEnv, pBlock->pBlockAgg, pTableScanInfo->pCtx, pBlockInfo->rows)) {
977
//      pCost->skipBlocks += 1;
978 979
//      qDebug("QInfo:0x%"PRIx64" data block discard, brange:%" PRId64 "-%" PRId64 ", rows:%d", pQInfo->qId, pBlockInfo->window.skey,
//             pBlockInfo->window.ekey, pBlockInfo->rows);
980
//      (*status) = BLK_DATA_FILTEROUT;
981 982 983 984 985
//      return TSDB_CODE_SUCCESS;
//    }

    pCost->totalCheckedRows += pBlockInfo->rows;
    pCost->loadBlocks += 1;
986
//    pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pTsdbReadHandle, NULL);
987 988 989 990 991
//    if (pBlock->pDataBlock == NULL) {
//      return terrno;
//    }

//    if (pQueryAttr->pFilters != NULL) {
992
//      filterSetColFieldData(pQueryAttr->pFilters, taosArrayGetSize(pBlock->pDataBlock), pBlock->pDataBlock);
993
//    }
994

995 996 997 998
//    if (pQueryAttr->pFilters != NULL || pRuntimeEnv->pTsBuf != NULL) {
//      filterColRowsInDataBlock(pRuntimeEnv, pBlock, ascQuery);
//    }
  }
H
Haojun Liao 已提交
999
#endif
1000 1001 1002
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
1003
static void updateTableQueryInfoForReverseScan(STableQueryInfo* pTableQueryInfo) {
1004 1005 1006 1007 1008
  if (pTableQueryInfo == NULL) {
    return;
  }
}

L
Liu Jicong 已提交
1009
void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status) {
1010
  if (status == TASK_NOT_COMPLETED) {
H
Haojun Liao 已提交
1011
    pTaskInfo->status = status;
1012 1013
  } else {
    // QUERY_NOT_COMPLETED is not compatible with any other status, so clear its position first
1014
    CLEAR_QUERY_STATUS(pTaskInfo, TASK_NOT_COMPLETED);
H
Haojun Liao 已提交
1015
    pTaskInfo->status |= status;
1016 1017 1018
  }
}

1019
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset) {
5
54liuyao 已提交
1020
  bool init = false;
1021
  for (int32_t i = 0; i < numOfOutput; ++i) {
1022
    pCtx[i].resultInfo = getResultEntryInfo(pResult, i, rowEntryInfoOffset);
5
54liuyao 已提交
1023 1024 1025
    if (init) {
      continue;
    }
1026 1027 1028 1029 1030

    struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo;
    if (isRowEntryCompleted(pResInfo) && isRowEntryInitialized(pResInfo)) {
      continue;
    }
1031 1032 1033 1034 1035

    if (fmIsWindowPseudoColumnFunc(pCtx[i].functionId)) {
      continue;
    }

1036 1037 1038 1039 1040 1041
    if (!pResInfo->initialized) {
      if (pCtx[i].functionId != -1) {
        pCtx[i].fpSet.init(&pCtx[i], pResInfo);
      } else {
        pResInfo->initialized = true;
      }
5
54liuyao 已提交
1042 1043
    } else {
      init = true;
1044 1045 1046 1047
    }
  }
}

1048 1049
static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep,
                                                int32_t status);
1050

H
Haojun Liao 已提交
1051 1052
void doFilter(SSDataBlock* pBlock, SFilterInfo* pFilterInfo, SColMatchInfo* pColMatchInfo) {
  if (pFilterInfo == NULL || pBlock->info.rows == 0) {
S
shenglian zhou 已提交
1053 1054
    return;
  }
1055

1056
  SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};
H
Haojun Liao 已提交
1057
  int32_t code = filterSetDataFromSlotId(pFilterInfo, &param1);
1058

1059
  SColumnInfoData* p = NULL;
1060
  int32_t          status = 0;
H
Haojun Liao 已提交
1061

1062
  // todo the keep seems never to be True??
H
Haojun Liao 已提交
1063
  bool keep = filterExecute(pFilterInfo, pBlock, &p, NULL, param1.numOfCols, &status);
1064
  extractQualifiedTupleByFilterResult(pBlock, p, keep, status);
H
Haojun Liao 已提交
1065

1066
  if (pColMatchInfo != NULL) {
H
Haojun Liao 已提交
1067 1068
    size_t  size = taosArrayGetSize(pColMatchInfo->pList);
    for (int32_t i = 0; i < size; ++i) {
H
Haojun Liao 已提交
1069
      SColMatchItem* pInfo = taosArrayGet(pColMatchInfo->pList, i);
1070
      if (pInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
1071
        SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, pInfo->dstSlotId);
1072
        if (pColData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
H
Haojun Liao 已提交
1073
          blockDataUpdateTsWindow(pBlock, pInfo->dstSlotId);
1074 1075 1076 1077 1078 1079
          break;
        }
      }
    }
  }

1080 1081
  colDataDestroy(p);
  taosMemoryFree(p);
1082 1083
}

1084
void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep, int32_t status) {
1085 1086 1087 1088
  if (keep) {
    return;
  }

H
Haojun Liao 已提交
1089 1090 1091
  int32_t totalRows = pBlock->info.rows;

  if (status == FILTER_RESULT_ALL_QUALIFIED) {
1092
    // here nothing needs to be done
H
Haojun Liao 已提交
1093
  } else if (status == FILTER_RESULT_NONE_QUALIFIED) {
1094
    pBlock->info.rows = 0;
H
Haojun Liao 已提交
1095
  } else {
1096
    SSDataBlock* px = createOneDataBlock(pBlock, true);
1097

1098 1099
    size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
    for (int32_t i = 0; i < numOfCols; ++i) {
1100 1101
      SColumnInfoData* pSrc = taosArrayGet(px->pDataBlock, i);
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
1102
      // it is a reserved column for scalar function, and no data in this column yet.
1103
      if (pDst->pData == NULL || pSrc->pData == NULL) {
1104 1105 1106
        continue;
      }

1107 1108
      colInfoDataCleanup(pDst, pBlock->info.rows);

1109
      int32_t numOfRows = 0;
1110
      for (int32_t j = 0; j < totalRows; ++j) {
1111
        if (((int8_t*)p->pData)[j] == 0) {
D
dapan1121 已提交
1112 1113
          continue;
        }
1114

D
dapan1121 已提交
1115
        if (colDataIsNull_s(pSrc, j)) {
1116
          colDataAppendNULL(pDst, numOfRows);
D
dapan1121 已提交
1117
        } else {
1118
          colDataAppend(pDst, numOfRows, colDataGetData(pSrc, j), false);
D
dapan1121 已提交
1119
        }
1120
        numOfRows += 1;
H
Haojun Liao 已提交
1121
      }
1122

1123
      // todo this value can be assigned directly
1124 1125 1126 1127 1128
      if (pBlock->info.rows == totalRows) {
        pBlock->info.rows = numOfRows;
      } else {
        ASSERT(pBlock->info.rows == numOfRows);
      }
1129
    }
1130

dengyihao's avatar
dengyihao 已提交
1131
    blockDataDestroy(px);  // fix memory leak
1132 1133 1134
  }
}

1135
void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId) {
1136
  // for simple group by query without interval, all the tables belong to one group result.
1137 1138 1139
  SExecTaskInfo*    pTaskInfo = pOperator->pTaskInfo;
  SAggOperatorInfo* pAggInfo = pOperator->info;

1140
  SResultRowInfo* pResultRowInfo = &pAggInfo->binfo.resultRowInfo;
1141 1142
  SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;
  int32_t*        rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset;
1143

1144
  SResultRow* pResultRow = doSetResultOutBufByKey(pAggInfo->aggSup.pResultBuf, pResultRowInfo, (char*)&groupId,
L
Liu Jicong 已提交
1145
                                                  sizeof(groupId), true, groupId, pTaskInfo, false, &pAggInfo->aggSup);
L
Liu Jicong 已提交
1146
  assert(pResultRow != NULL);
1147 1148 1149 1150 1151 1152

  /*
   * not assign result buffer yet, add new result buffer
   * all group belong to one result set, and each group result has different group id so set the id to be one
   */
  if (pResultRow->pageId == -1) {
dengyihao's avatar
dengyihao 已提交
1153 1154
    int32_t ret =
        addNewWindowResultBuf(pResultRow, pAggInfo->aggSup.pResultBuf, groupId, pAggInfo->binfo.pRes->info.rowSize);
1155 1156 1157 1158 1159
    if (ret != TSDB_CODE_SUCCESS) {
      return;
    }
  }

1160
  setResultRowInitCtx(pResultRow, pCtx, numOfOutput, rowEntryInfoOffset);
1161 1162
}

1163 1164 1165
static void setExecutionContext(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId) {
  SAggOperatorInfo* pAggInfo = pOperator->info;
  if (pAggInfo->groupId != UINT64_MAX && pAggInfo->groupId == groupId) {
1166 1167
    return;
  }
1168 1169

  doSetTableGroupOutputBuf(pOperator, numOfOutput, groupId);
1170 1171

  // record the current active group id
H
Haojun Liao 已提交
1172
  pAggInfo->groupId = groupId;
1173 1174
}

dengyihao's avatar
dengyihao 已提交
1175 1176
static void doUpdateNumOfRows(SqlFunctionCtx* pCtx, SResultRow* pRow, int32_t numOfExprs,
                              const int32_t* rowCellOffset) {
1177
  bool returnNotNull = false;
1178
  for (int32_t j = 0; j < numOfExprs; ++j) {
1179
    struct SResultRowEntryInfo* pResInfo = getResultEntryInfo(pRow, j, rowCellOffset);
1180 1181 1182 1183 1184 1185 1186
    if (!isRowEntryInitialized(pResInfo)) {
      continue;
    }

    if (pRow->numOfRows < pResInfo->numOfRes) {
      pRow->numOfRows = pResInfo->numOfRes;
    }
1187

1188
    if (fmIsNotNullOutputFunc(pCtx[j].functionId)) {
1189 1190
      returnNotNull = true;
    }
1191
  }
S
shenglian zhou 已提交
1192 1193
  // if all expr skips all blocks, e.g. all null inputs for max function, output one row in final result.
  //  except for first/last, which require not null output, output no rows
1194
  if (pRow->numOfRows == 0 && !returnNotNull) {
1195
    pRow->numOfRows = 1;
1196 1197 1198
  }
}

1199 1200
static void doCopyResultToDataBlock(SExprInfo* pExprInfo, int32_t numOfExprs, SResultRow* pRow, SqlFunctionCtx* pCtx,
                                    SSDataBlock* pBlock, const int32_t* rowEntryOffset, SExecTaskInfo* pTaskInfo) {
1201 1202 1203
  for (int32_t j = 0; j < numOfExprs; ++j) {
    int32_t slotId = pExprInfo[j].base.resSchema.slotId;

1204
    pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowEntryOffset);
1205
    if (pCtx[j].fpSet.finalize) {
1206
      if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_group_key") == 0) {
1207 1208
        // for groupkey along with functions that output multiple lines(e.g. Histogram)
        // need to match groupkey result for each output row of that function.
1209 1210 1211 1212 1213
        if (pCtx[j].resultInfo->numOfRes != 0) {
          pCtx[j].resultInfo->numOfRes = pRow->numOfRows;
        }
      }

1214 1215 1216
      int32_t code = pCtx[j].fpSet.finalize(&pCtx[j], pBlock);
      if (TAOS_FAILED(code)) {
        qError("%s build result data block error, code %s", GET_TASKID(pTaskInfo), tstrerror(code));
1217
        T_LONG_JMP(pTaskInfo->env, code);
1218 1219
      }
    } else if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_select_value") == 0) {
1220
      // do nothing
1221
    } else {
1222 1223
      // expand the result into multiple rows. E.g., _wstart, top(k, 20)
      // the _wstart needs to copy to 20 following rows, since the results of top-k expands to 20 different rows.
1224 1225 1226 1227 1228 1229 1230
      SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId);
      char*            in = GET_ROWCELL_INTERBUF(pCtx[j].resultInfo);
      for (int32_t k = 0; k < pRow->numOfRows; ++k) {
        colDataAppend(pColInfoData, pBlock->info.rows + k, in, pCtx[j].resultInfo->isNullRes);
      }
    }
  }
1231 1232
}

1233 1234 1235
// todo refactor. SResultRow has direct pointer in miainfo
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
                           SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {
1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261
  SFilePage*  page = getBufPage(pBuf, resultRowPosition->pageId);
  SResultRow* pRow = (SResultRow*)((char*)page + resultRowPosition->offset);

  SqlFunctionCtx* pCtx = pSup->pCtx;
  SExprInfo*      pExprInfo = pSup->pExprInfo;
  const int32_t*  rowEntryOffset = pSup->rowEntryInfoOffset;

  doUpdateNumOfRows(pCtx, pRow, pSup->numOfExprs, rowEntryOffset);
  if (pRow->numOfRows == 0) {
    releaseBufPage(pBuf, page);
    return 0;
  }

  int32_t size = pBlock->info.capacity;
  while (pBlock->info.rows + pRow->numOfRows > size) {
    size = size * 1.25;
  }

  int32_t code = blockDataEnsureCapacity(pBlock, size);
  if (TAOS_FAILED(code)) {
    releaseBufPage(pBuf, page);
    qError("%s ensure result data capacity failed, code %s", GET_TASKID(pTaskInfo), tstrerror(code));
    T_LONG_JMP(pTaskInfo->env, code);
  }

  doCopyResultToDataBlock(pExprInfo, pSup->numOfExprs, pRow, pCtx, pBlock, rowEntryOffset, pTaskInfo);
1262 1263

  releaseBufPage(pBuf, page);
1264
  pBlock->info.rows += pRow->numOfRows;
1265 1266 1267
  return 0;
}

1268 1269 1270 1271 1272 1273 1274
int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, SExprSupp* pSup, SDiskbasedBuf* pBuf,
                           SGroupResInfo* pGroupResInfo) {
  SExprInfo*      pExprInfo = pSup->pExprInfo;
  int32_t         numOfExprs = pSup->numOfExprs;
  int32_t*        rowEntryOffset = pSup->rowEntryInfoOffset;
  SqlFunctionCtx* pCtx = pSup->pCtx;

1275
  int32_t numOfRows = getNumOfTotalRes(pGroupResInfo);
1276

1277
  for (int32_t i = pGroupResInfo->index; i < numOfRows; i += 1) {
L
Liu Jicong 已提交
1278 1279
    SResKeyPos* pPos = taosArrayGetP(pGroupResInfo->pRows, i);
    SFilePage*  page = getBufPage(pBuf, pPos->pos.pageId);
1280

1281
    SResultRow* pRow = (SResultRow*)((char*)page + pPos->pos.offset);
1282

H
Haojun Liao 已提交
1283
    doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset);
1284 1285

    // no results, continue to check the next one
1286 1287
    if (pRow->numOfRows == 0) {
      pGroupResInfo->index += 1;
1288
      releaseBufPage(pBuf, page);
1289 1290 1291
      continue;
    }

1292 1293 1294 1295 1296
    if (pBlock->info.groupId == 0) {
      pBlock->info.groupId = pPos->groupId;
    } else {
      // current value belongs to different group, it can't be packed into one datablock
      if (pBlock->info.groupId != pPos->groupId) {
1297
        releaseBufPage(pBuf, page);
1298 1299 1300 1301
        break;
      }
    }

1302
    if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
1303
      ASSERT(pBlock->info.rows > 0);
1304
      releaseBufPage(pBuf, page);
1305 1306 1307 1308
      break;
    }

    pGroupResInfo->index += 1;
1309
    doCopyResultToDataBlock(pExprInfo, numOfExprs, pRow, pCtx, pBlock, rowEntryOffset, pTaskInfo);
1310

1311
    releaseBufPage(pBuf, page);
1312
    pBlock->info.rows += pRow->numOfRows;
1313 1314
  }

X
Xiaoyu Wang 已提交
1315 1316
  qDebug("%s result generated, rows:%d, groupId:%" PRIu64, GET_TASKID(pTaskInfo), pBlock->info.rows,
         pBlock->info.groupId);
1317

1318
  blockDataUpdateTsWindow(pBlock, 0);
1319 1320 1321
  return 0;
}

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 1357 1358 1359 1360 1361
void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                           SDiskbasedBuf* pBuf) {
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
  SSDataBlock*   pBlock = pbInfo->pRes;

  // set output datablock version
  pBlock->info.version = pTaskInfo->version;

  blockDataCleanup(pBlock);
  if (!hasRemainResults(pGroupResInfo)) {
    return;
  }

  // clear the existed group id
  pBlock->info.groupId = 0;
  ASSERT(!pbInfo->mergeResultBlock);
  doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo);
  if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
    SStreamStateAggOperatorInfo* pInfo = pOperator->info;

    char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
    if (tbname != NULL) {
      memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
    } else {
      pBlock->info.parTbName[0] = 0;
    }
  } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
             pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
             pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
    SStreamSessionAggOperatorInfo* pInfo = pOperator->info;

    char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
    if (tbname != NULL) {
      memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
    } else {
      pBlock->info.parTbName[0] = 0;
    }
  }
}

X
Xiaoyu Wang 已提交
1362 1363
void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                            SDiskbasedBuf* pBuf) {
1364
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1365
  SSDataBlock*   pBlock = pbInfo->pRes;
1366

1367 1368 1369
  // set output datablock version
  pBlock->info.version = pTaskInfo->version;

1370
  blockDataCleanup(pBlock);
1371
  if (!hasRemainResults(pGroupResInfo)) {
1372 1373 1374
    return;
  }

1375 1376
  // clear the existed group id
  pBlock->info.groupId = 0;
1377 1378 1379
  if (!pbInfo->mergeResultBlock) {
    doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo);
  } else {
dengyihao's avatar
dengyihao 已提交
1380
    while (hasRemainResults(pGroupResInfo)) {
1381 1382 1383
      doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo);
      if (pBlock->info.rows >= pOperator->resultInfo.threshold) {
        break;
1384 1385
      }

1386 1387
      // clearing group id to continue to merge data that belong to different groups
      pBlock->info.groupId = 0;
1388
    }
1389 1390 1391

    // clear the group id info in SSDataBlock, since the client does not need it
    pBlock->info.groupId = 0;
1392 1393 1394
  }
}

L
Liu Jicong 已提交
1395 1396
void queryCostStatis(SExecTaskInfo* pTaskInfo) {
  STaskCostInfo* pSummary = &pTaskInfo->cost;
1397

1398 1399
  SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;
  if (pSummary->pRecoder != NULL) {
1400
    qDebug(
H
Haojun Liao 已提交
1401 1402 1403 1404 1405
        "%s :cost summary: elapsed time:%.2f ms, extract tableList:%.2f ms, createGroupIdMap:%.2f ms, total blocks:%d, "
        "load block SMA:%d, load data block:%d, total rows:%" PRId64 ", check rows:%" PRId64,
        GET_TASKID(pTaskInfo), pSummary->elapsedTime / 1000.0, pSummary->extractListTime, pSummary->groupIdMapTime,
        pRecorder->totalBlocks, pRecorder->loadBlockStatis, pRecorder->loadBlocks, pRecorder->totalRows,
        pRecorder->totalCheckedRows);
1406
  }
1407 1408
}

L
Liu Jicong 已提交
1409 1410 1411
// static void updateOffsetVal(STaskRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo) {
//   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
//   STableQueryInfo* pTableQueryInfo = pRuntimeEnv->current;
1412
//
L
Liu Jicong 已提交
1413
//   int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQueryAttr->order.order);
1414
//
L
Liu Jicong 已提交
1415 1416 1417 1418
//   if (pQueryAttr->limit.offset == pBlockInfo->rows) {  // current block will ignore completed
//     pTableQueryInfo->lastKey = QUERY_IS_ASC_QUERY(pQueryAttr) ? pBlockInfo->window.ekey + step :
//     pBlockInfo->window.skey + step; pQueryAttr->limit.offset = 0; return;
//   }
1419
//
L
Liu Jicong 已提交
1420 1421 1422 1423 1424
//   if (QUERY_IS_ASC_QUERY(pQueryAttr)) {
//     pQueryAttr->pos = (int32_t)pQueryAttr->limit.offset;
//   } else {
//     pQueryAttr->pos = pBlockInfo->rows - (int32_t)pQueryAttr->limit.offset - 1;
//   }
1425
//
L
Liu Jicong 已提交
1426
//   assert(pQueryAttr->pos >= 0 && pQueryAttr->pos <= pBlockInfo->rows - 1);
1427
//
L
Liu Jicong 已提交
1428 1429
//   SArray *         pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
//   SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
1430
//
L
Liu Jicong 已提交
1431 1432
//   // update the pQueryAttr->limit.offset value, and pQueryAttr->pos value
//   TSKEY *keys = (TSKEY *) pColInfoData->pData;
1433
//
L
Liu Jicong 已提交
1434 1435 1436
//   // update the offset value
//   pTableQueryInfo->lastKey = keys[pQueryAttr->pos];
//   pQueryAttr->limit.offset = 0;
1437
//
L
Liu Jicong 已提交
1438
//   int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
1439
//
L
Liu Jicong 已提交
1440 1441 1442 1443
//   //qDebug("QInfo:0x%"PRIx64" check data block, brange:%" PRId64 "-%" PRId64 ", numBlocksOfStep:%d, numOfRes:%d,
//   lastKey:%"PRId64, GET_TASKID(pRuntimeEnv),
//          pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes, pQuery->current->lastKey);
// }
1444

L
Liu Jicong 已提交
1445 1446
// void skipBlocks(STaskRuntimeEnv *pRuntimeEnv) {
//   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
1447
//
L
Liu Jicong 已提交
1448 1449 1450
//   if (pQueryAttr->limit.offset <= 0 || pQueryAttr->numOfFilterCols > 0) {
//     return;
//   }
1451
//
L
Liu Jicong 已提交
1452 1453
//   pQueryAttr->pos = 0;
//   int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQueryAttr->order.order);
1454
//
L
Liu Jicong 已提交
1455 1456
//   STableQueryInfo* pTableQueryInfo = pRuntimeEnv->current;
//   TsdbQueryHandleT pTsdbReadHandle = pRuntimeEnv->pTsdbReadHandle;
1457
//
L
Liu Jicong 已提交
1458 1459 1460
//   SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER;
//   while (tsdbNextDataBlock(pTsdbReadHandle)) {
//     if (isTaskKilled(pRuntimeEnv->qinfo)) {
1461
//       T_LONG_JMP(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED);
L
Liu Jicong 已提交
1462
//     }
1463
//
L
Liu Jicong 已提交
1464
//     tsdbRetrieveDataBlockInfo(pTsdbReadHandle, &blockInfo);
1465
//
L
Liu Jicong 已提交
1466 1467 1468 1469
//     if (pQueryAttr->limit.offset > blockInfo.rows) {
//       pQueryAttr->limit.offset -= blockInfo.rows;
//       pTableQueryInfo->lastKey = (QUERY_IS_ASC_QUERY(pQueryAttr)) ? blockInfo.window.ekey : blockInfo.window.skey;
//       pTableQueryInfo->lastKey += step;
1470
//
L
Liu Jicong 已提交
1471 1472 1473 1474 1475 1476 1477
//       //qDebug("QInfo:0x%"PRIx64" skip rows:%d, offset:%" PRId64, GET_TASKID(pRuntimeEnv), blockInfo.rows,
//              pQuery->limit.offset);
//     } else {  // find the appropriated start position in current block
//       updateOffsetVal(pRuntimeEnv, &blockInfo);
//       break;
//     }
//   }
1478
//
L
Liu Jicong 已提交
1479
//   if (terrno != TSDB_CODE_SUCCESS) {
1480
//     T_LONG_JMP(pRuntimeEnv->env, terrno);
L
Liu Jicong 已提交
1481 1482 1483 1484 1485 1486 1487
//   }
// }

// static TSKEY doSkipIntervalProcess(STaskRuntimeEnv* pRuntimeEnv, STimeWindow* win, SDataBlockInfo* pBlockInfo,
// STableQueryInfo* pTableQueryInfo) {
//   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
//   SResultRowInfo *pWindowResInfo = &pRuntimeEnv->resultRowInfo;
1488
//
L
Liu Jicong 已提交
1489 1490 1491
//   assert(pQueryAttr->limit.offset == 0);
//   STimeWindow tw = *win;
//   getNextTimeWindow(pQueryAttr, &tw);
1492
//
L
Liu Jicong 已提交
1493 1494
//   if ((tw.skey <= pBlockInfo->window.ekey && QUERY_IS_ASC_QUERY(pQueryAttr)) ||
//       (tw.ekey >= pBlockInfo->window.skey && !QUERY_IS_ASC_QUERY(pQueryAttr))) {
1495
//
L
Liu Jicong 已提交
1496 1497 1498 1499
//     // load the data block and check data remaining in current data block
//     // TODO optimize performance
//     SArray *         pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
//     SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
1500
//
L
Liu Jicong 已提交
1501 1502 1503 1504
//     tw = *win;
//     int32_t startPos =
//         getNextQualifiedWindow(pQueryAttr, &tw, pBlockInfo, pColInfoData->pData, binarySearchForKey, -1);
//     assert(startPos >= 0);
1505
//
L
Liu Jicong 已提交
1506 1507
//     // set the abort info
//     pQueryAttr->pos = startPos;
1508
//
L
Liu Jicong 已提交
1509 1510 1511 1512
//     // reset the query start timestamp
//     pTableQueryInfo->win.skey = ((TSKEY *)pColInfoData->pData)[startPos];
//     pQueryAttr->window.skey = pTableQueryInfo->win.skey;
//     TSKEY key = pTableQueryInfo->win.skey;
1513
//
L
Liu Jicong 已提交
1514 1515
//     pWindowResInfo->prevSKey = tw.skey;
//     int32_t index = pRuntimeEnv->resultRowInfo.curIndex;
1516
//
L
Liu Jicong 已提交
1517 1518
//     int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
//     pRuntimeEnv->resultRowInfo.curIndex = index;  // restore the window index
1519
//
L
Liu Jicong 已提交
1520 1521 1522 1523
//     //qDebug("QInfo:0x%"PRIx64" check data block, brange:%" PRId64 "-%" PRId64 ", numOfRows:%d, numOfRes:%d,
//     lastKey:%" PRId64,
//            GET_TASKID(pRuntimeEnv), pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes,
//            pQueryAttr->current->lastKey);
1524
//
L
Liu Jicong 已提交
1525 1526 1527 1528 1529
//     return key;
//   } else {  // do nothing
//     pQueryAttr->window.skey      = tw.skey;
//     pWindowResInfo->prevSKey = tw.skey;
//     pTableQueryInfo->lastKey = tw.skey;
1530
//
L
Liu Jicong 已提交
1531 1532
//     return tw.skey;
//   }
1533
//
L
Liu Jicong 已提交
1534 1535 1536 1537 1538 1539 1540 1541 1542 1543
//   return true;
// }

// static bool skipTimeInterval(STaskRuntimeEnv *pRuntimeEnv, TSKEY* start) {
//   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
//   if (QUERY_IS_ASC_QUERY(pQueryAttr)) {
//     assert(*start <= pRuntimeEnv->current->lastKey);
//   } else {
//     assert(*start >= pRuntimeEnv->current->lastKey);
//   }
1544
//
L
Liu Jicong 已提交
1545 1546 1547 1548 1549
//   // if queried with value filter, do NOT forward query start position
//   if (pQueryAttr->limit.offset <= 0 || pQueryAttr->numOfFilterCols > 0 || pRuntimeEnv->pTsBuf != NULL ||
//   pRuntimeEnv->pFillInfo != NULL) {
//     return true;
//   }
1550
//
L
Liu Jicong 已提交
1551 1552 1553 1554 1555 1556 1557
//   /*
//    * 1. for interval without interpolation query we forward pQueryAttr->interval.interval at a time for
//    *    pQueryAttr->limit.offset times. Since hole exists, pQueryAttr->interval.interval*pQueryAttr->limit.offset
//    value is
//    *    not valid. otherwise, we only forward pQueryAttr->limit.offset number of points
//    */
//   assert(pRuntimeEnv->resultRowInfo.prevSKey == TSKEY_INITIAL_VAL);
1558
//
L
Liu Jicong 已提交
1559 1560
//   STimeWindow w = TSWINDOW_INITIALIZER;
//   bool ascQuery = QUERY_IS_ASC_QUERY(pQueryAttr);
1561
//
L
Liu Jicong 已提交
1562 1563
//   SResultRowInfo *pWindowResInfo = &pRuntimeEnv->resultRowInfo;
//   STableQueryInfo *pTableQueryInfo = pRuntimeEnv->current;
1564
//
L
Liu Jicong 已提交
1565 1566 1567
//   SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER;
//   while (tsdbNextDataBlock(pRuntimeEnv->pTsdbReadHandle)) {
//     tsdbRetrieveDataBlockInfo(pRuntimeEnv->pTsdbReadHandle, &blockInfo);
1568
//
L
Liu Jicong 已提交
1569 1570 1571 1572 1573 1574 1575 1576 1577
//     if (QUERY_IS_ASC_QUERY(pQueryAttr)) {
//       if (pWindowResInfo->prevSKey == TSKEY_INITIAL_VAL) {
//         getAlignQueryTimeWindow(pQueryAttr, blockInfo.window.skey, blockInfo.window.skey, pQueryAttr->window.ekey,
//         &w); pWindowResInfo->prevSKey = w.skey;
//       }
//     } else {
//       getAlignQueryTimeWindow(pQueryAttr, blockInfo.window.ekey, pQueryAttr->window.ekey, blockInfo.window.ekey, &w);
//       pWindowResInfo->prevSKey = w.skey;
//     }
1578
//
L
Liu Jicong 已提交
1579 1580
//     // the first time window
//     STimeWindow win = getActiveTimeWindow(pWindowResInfo, pWindowResInfo->prevSKey, pQueryAttr);
1581
//
L
Liu Jicong 已提交
1582 1583
//     while (pQueryAttr->limit.offset > 0) {
//       STimeWindow tw = win;
1584
//
L
Liu Jicong 已提交
1585 1586 1587
//       if ((win.ekey <= blockInfo.window.ekey && ascQuery) || (win.ekey >= blockInfo.window.skey && !ascQuery)) {
//         pQueryAttr->limit.offset -= 1;
//         pWindowResInfo->prevSKey = win.skey;
1588
//
L
Liu Jicong 已提交
1589 1590 1591 1592 1593 1594
//         // current time window is aligned with blockInfo.window.ekey
//         // restart it from next data block by set prevSKey to be TSKEY_INITIAL_VAL;
//         if ((win.ekey == blockInfo.window.ekey && ascQuery) || (win.ekey == blockInfo.window.skey && !ascQuery)) {
//           pWindowResInfo->prevSKey = TSKEY_INITIAL_VAL;
//         }
//       }
1595
//
L
Liu Jicong 已提交
1596 1597 1598 1599
//       if (pQueryAttr->limit.offset == 0) {
//         *start = doSkipIntervalProcess(pRuntimeEnv, &win, &blockInfo, pTableQueryInfo);
//         return true;
//       }
1600
//
L
Liu Jicong 已提交
1601 1602
//       // current window does not ended in current data block, try next data block
//       getNextTimeWindow(pQueryAttr, &tw);
1603
//
L
Liu Jicong 已提交
1604 1605 1606 1607 1608 1609 1610 1611 1612
//       /*
//        * If the next time window still starts from current data block,
//        * load the primary timestamp column first, and then find the start position for the next queried time window.
//        * Note that only the primary timestamp column is required.
//        * TODO: Optimize for this cases. All data blocks are not needed to be loaded, only if the first actually
//        required
//        * time window resides in current data block.
//        */
//       if ((tw.skey <= blockInfo.window.ekey && ascQuery) || (tw.ekey >= blockInfo.window.skey && !ascQuery)) {
1613
//
L
Liu Jicong 已提交
1614 1615
//         SArray *pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
//         SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
1616
//
L
Liu Jicong 已提交
1617 1618 1619
//         if ((win.ekey > blockInfo.window.ekey && ascQuery) || (win.ekey < blockInfo.window.skey && !ascQuery)) {
//           pQueryAttr->limit.offset -= 1;
//         }
1620
//
L
Liu Jicong 已提交
1621 1622 1623 1624 1625 1626 1627 1628
//         if (pQueryAttr->limit.offset == 0) {
//           *start = doSkipIntervalProcess(pRuntimeEnv, &win, &blockInfo, pTableQueryInfo);
//           return true;
//         } else {
//           tw = win;
//           int32_t startPos =
//               getNextQualifiedWindow(pQueryAttr, &tw, &blockInfo, pColInfoData->pData, binarySearchForKey, -1);
//           assert(startPos >= 0);
1629
//
L
Liu Jicong 已提交
1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640
//           // set the abort info
//           pQueryAttr->pos = startPos;
//           pTableQueryInfo->lastKey = ((TSKEY *)pColInfoData->pData)[startPos];
//           pWindowResInfo->prevSKey = tw.skey;
//           win = tw;
//         }
//       } else {
//         break;  // offset is not 0, and next time window begins or ends in the next block.
//       }
//     }
//   }
1641
//
L
Liu Jicong 已提交
1642 1643
//   // check for error
//   if (terrno != TSDB_CODE_SUCCESS) {
1644
//     T_LONG_JMP(pRuntimeEnv->env, terrno);
L
Liu Jicong 已提交
1645
//   }
1646
//
L
Liu Jicong 已提交
1647 1648
//   return true;
// }
1649

1650
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num) {
H
Haojun Liao 已提交
1651
  if (p->pDownstream == NULL) {
H
Haojun Liao 已提交
1652
    assert(p->numOfDownstream == 0);
1653 1654
  }

wafwerar's avatar
wafwerar 已提交
1655
  p->pDownstream = taosMemoryCalloc(1, num * POINTER_BYTES);
1656 1657 1658 1659 1660 1661 1662
  if (p->pDownstream == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  memcpy(p->pDownstream, pDownstream, num * POINTER_BYTES);
  p->numOfDownstream = num;
  return TSDB_CODE_SUCCESS;
1663 1664
}

dengyihao's avatar
dengyihao 已提交
1665 1666
static int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t keyBufSize,
                                const char* pKey);
1667

L
Liu Jicong 已提交
1668
static bool needToMerge(SSDataBlock* pBlock, SArray* groupInfo, char** buf, int32_t rowIndex) {
1669 1670 1671 1672
  size_t size = taosArrayGetSize(groupInfo);
  if (size == 0) {
    return true;
  }
1673

1674 1675
  for (int32_t i = 0; i < size; ++i) {
    int32_t* index = taosArrayGet(groupInfo, i);
1676

1677
    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, *index);
L
Liu Jicong 已提交
1678
    bool             isNull = colDataIsNull(pColInfo, rowIndex, pBlock->info.rows, NULL);
1679

1680 1681 1682
    if ((isNull && buf[i] != NULL) || (!isNull && buf[i] == NULL)) {
      return false;
    }
1683

1684 1685 1686 1687 1688 1689 1690 1691 1692 1693 1694 1695 1696
    char* pCell = colDataGetData(pColInfo, rowIndex);
    if (IS_VAR_DATA_TYPE(pColInfo->info.type)) {
      if (varDataLen(pCell) != varDataLen(buf[i])) {
        return false;
      } else {
        if (memcmp(varDataVal(pCell), varDataVal(buf[i]), varDataLen(pCell)) != 0) {
          return false;
        }
      }
    } else {
      if (memcmp(pCell, buf[i], pColInfo->info.bytes) != 0) {
        return false;
      }
1697 1698 1699
    }
  }

1700
  return 0;
1701 1702
}

1703
static bool saveCurrentTuple(char** rowColData, SArray* pColumnList, SSDataBlock* pBlock, int32_t rowIndex) {
L
Liu Jicong 已提交
1704
  int32_t size = (int32_t)taosArrayGetSize(pColumnList);
1705

L
Liu Jicong 已提交
1706 1707
  for (int32_t i = 0; i < size; ++i) {
    int32_t*         index = taosArrayGet(pColumnList, i);
1708
    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, *index);
H
Haojun Liao 已提交
1709

1710 1711 1712
    char* data = colDataGetData(pColInfo, rowIndex);
    memcpy(rowColData[i], data, colDataGetLength(pColInfo, rowIndex));
  }
1713

1714 1715
  return true;
}
1716

X
Xiaoyu Wang 已提交
1717
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag) {
1718
  // todo add more information about exchange operation
1719
  int32_t type = pOperator->operatorType;
X
Xiaoyu Wang 已提交
1720
  if (type == QUERY_NODE_PHYSICAL_PLAN_EXCHANGE || type == QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN ||
1721
      type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN ||
1722
      type == QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN) {
1723 1724 1725
    *order = TSDB_ORDER_ASC;
    *scanFlag = MAIN_SCAN;
    return TSDB_CODE_SUCCESS;
1726
  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
1727 1728 1729 1730
    STableScanInfo* pTableScanInfo = pOperator->info;
    *order = pTableScanInfo->cond.order;
    *scanFlag = pTableScanInfo->scanFlag;
    return TSDB_CODE_SUCCESS;
1731 1732 1733 1734 1735
  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN) {
    STableMergeScanInfo* pTableScanInfo = pOperator->info;
    *order = pTableScanInfo->cond.order;
    *scanFlag = pTableScanInfo->scanFlag;
    return TSDB_CODE_SUCCESS;
1736
  } else {
H
Haojun Liao 已提交
1737
    if (pOperator->pDownstream == NULL || pOperator->pDownstream[0] == NULL) {
1738
      return TSDB_CODE_INVALID_PARA;
H
Haojun Liao 已提交
1739
    } else {
1740
      return getTableScanInfo(pOperator->pDownstream[0], order, scanFlag);
1741 1742 1743
    }
  }
}
1744

1745
// this is a blocking operator
L
Liu Jicong 已提交
1746
static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
1747 1748
  if (OPTR_IS_OPENED(pOperator)) {
    return TSDB_CODE_SUCCESS;
1749 1750
  }

H
Haojun Liao 已提交
1751
  SExecTaskInfo*    pTaskInfo = pOperator->pTaskInfo;
1752
  SAggOperatorInfo* pAggInfo = pOperator->info;
H
Haojun Liao 已提交
1753

1754 1755
  SExprSupp*     pSup = &pOperator->exprSupp;
  SOperatorInfo* downstream = pOperator->pDownstream[0];
1756

1757 1758
  int64_t st = taosGetTimestampUs();

1759 1760 1761
  int32_t order = TSDB_ORDER_ASC;
  int32_t scanFlag = MAIN_SCAN;

H
Haojun Liao 已提交
1762
  while (1) {
1763
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1764 1765 1766 1767
    if (pBlock == NULL) {
      break;
    }

1768 1769
    int32_t code = getTableScanInfo(pOperator, &order, &scanFlag);
    if (code != TSDB_CODE_SUCCESS) {
1770
      T_LONG_JMP(pTaskInfo->env, code);
1771
    }
1772

1773
    // there is an scalar expression that needs to be calculated before apply the group aggregation.
1774 1775 1776
    if (pAggInfo->scalarExprSup.pExprInfo != NULL) {
      SExprSupp* pSup1 = &pAggInfo->scalarExprSup;
      code = projectApplyFunctions(pSup1->pExprInfo, pBlock, pBlock, pSup1->pCtx, pSup1->numOfExprs, NULL);
1777
      if (code != TSDB_CODE_SUCCESS) {
1778
        T_LONG_JMP(pTaskInfo->env, code);
1779
      }
1780 1781
    }

1782
    // the pDataBlock are always the same one, no need to call this again
1783
    setExecutionContext(pOperator, pOperator->exprSupp.numOfExprs, pBlock->info.groupId);
1784
    setInputDataBlock(pSup, pBlock, order, scanFlag, true);
1785
    code = doAggregateImpl(pOperator, pSup->pCtx);
1786
    if (code != 0) {
1787
      T_LONG_JMP(pTaskInfo->env, code);
1788
    }
1789 1790
  }

1791
  initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, 0);
H
Haojun Liao 已提交
1792
  OPTR_SET_OPENED(pOperator);
1793

1794
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
H
Haojun Liao 已提交
1795 1796 1797
  return TSDB_CODE_SUCCESS;
}

1798
static SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) {
L
Liu Jicong 已提交
1799
  SAggOperatorInfo* pAggInfo = pOperator->info;
H
Haojun Liao 已提交
1800 1801 1802 1803 1804 1805
  SOptrBasicInfo*   pInfo = &pAggInfo->binfo;

  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

L
Liu Jicong 已提交
1806
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1807
  pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
H
Haojun Liao 已提交
1808
  if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1809
    setOperatorCompleted(pOperator);
H
Haojun Liao 已提交
1810 1811 1812
    return NULL;
  }

H
Haojun Liao 已提交
1813
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
S
slzhou 已提交
1814 1815
  while (1) {
    doBuildResultDatablock(pOperator, pInfo, &pAggInfo->groupResInfo, pAggInfo->aggSup.pResultBuf);
H
Haojun Liao 已提交
1816
    doFilter(pInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
S
slzhou 已提交
1817

1818
    if (!hasRemainResults(&pAggInfo->groupResInfo)) {
H
Haojun Liao 已提交
1819
      setOperatorCompleted(pOperator);
S
slzhou 已提交
1820 1821
      break;
    }
1822

S
slzhou 已提交
1823 1824 1825 1826
    if (pInfo->pRes->info.rows > 0) {
      break;
    }
  }
1827

1828
  size_t rows = blockDataGetNumOfRows(pInfo->pRes);
1829 1830
  pOperator->resultInfo.totalRows += rows;

1831
  return (rows == 0) ? NULL : pInfo->pRes;
1832 1833
}

wmmhello's avatar
wmmhello 已提交
1834
int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length) {
1835
  if (result == NULL || length == NULL) {
wmmhello's avatar
wmmhello 已提交
1836 1837 1838
    return TSDB_CODE_TSC_INVALID_INPUT;
  }
  SOptrBasicInfo* pInfo = (SOptrBasicInfo*)(pOperator->info);
1839
  SAggSupporter*  pSup = (SAggSupporter*)POINTER_SHIFT(pOperator->info, sizeof(SOptrBasicInfo));
1840
  int32_t         size = tSimpleHashGetSize(pSup->pResultRowHashTable);
1841 1842 1843
  size_t          keyLen = sizeof(uint64_t) * 2;  // estimate the key length
  int32_t         totalSize =
      sizeof(int32_t) + sizeof(int32_t) + size * (sizeof(int32_t) + keyLen + sizeof(int32_t) + pSup->resultRowSize);
wmmhello's avatar
wmmhello 已提交
1844

C
Cary Xu 已提交
1845 1846 1847 1848 1849 1850
  // no result
  if (getTotalBufSize(pSup->pResultBuf) == 0) {
    *result = NULL;
    *length = 0;
    return TSDB_CODE_SUCCESS;
  }
1851

wmmhello's avatar
wmmhello 已提交
1852
  *result = (char*)taosMemoryCalloc(1, totalSize);
L
Liu Jicong 已提交
1853
  if (*result == NULL) {
wmmhello's avatar
wmmhello 已提交
1854
    return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
1855
  }
wmmhello's avatar
wmmhello 已提交
1856

wmmhello's avatar
wmmhello 已提交
1857
  int32_t offset = sizeof(int32_t);
wmmhello's avatar
wmmhello 已提交
1858 1859
  *(int32_t*)(*result + offset) = size;
  offset += sizeof(int32_t);
1860 1861

  // prepare memory
1862
  SResultRowPosition* pos = &pInfo->resultRowInfo.cur;
dengyihao's avatar
dengyihao 已提交
1863 1864
  void*               pPage = getBufPage(pSup->pResultBuf, pos->pageId);
  SResultRow*         pRow = (SResultRow*)((char*)pPage + pos->offset);
1865 1866
  setBufPageDirty(pPage, true);
  releaseBufPage(pSup->pResultBuf, pPage);
L
Liu Jicong 已提交
1867

1868 1869 1870 1871
  int32_t iter = 0;
  void*   pIter = NULL;
  while ((pIter = tSimpleHashIterate(pSup->pResultRowHashTable, pIter, &iter))) {
    void*               key = tSimpleHashGetKey(pIter, &keyLen);
1872
    SResultRowPosition* p1 = (SResultRowPosition*)pIter;
1873

dengyihao's avatar
dengyihao 已提交
1874
    pPage = (SFilePage*)getBufPage(pSup->pResultBuf, p1->pageId);
1875
    pRow = (SResultRow*)((char*)pPage + p1->offset);
1876 1877
    setBufPageDirty(pPage, true);
    releaseBufPage(pSup->pResultBuf, pPage);
wmmhello's avatar
wmmhello 已提交
1878 1879 1880

    // recalculate the result size
    int32_t realTotalSize = offset + sizeof(int32_t) + keyLen + sizeof(int32_t) + pSup->resultRowSize;
L
Liu Jicong 已提交
1881
    if (realTotalSize > totalSize) {
wmmhello's avatar
wmmhello 已提交
1882
      char* tmp = (char*)taosMemoryRealloc(*result, realTotalSize);
L
Liu Jicong 已提交
1883
      if (tmp == NULL) {
wafwerar's avatar
wafwerar 已提交
1884
        taosMemoryFree(*result);
wmmhello's avatar
wmmhello 已提交
1885
        *result = NULL;
wmmhello's avatar
wmmhello 已提交
1886
        return TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
1887
      } else {
wmmhello's avatar
wmmhello 已提交
1888 1889 1890 1891 1892 1893 1894 1895 1896 1897 1898 1899
        *result = tmp;
      }
    }
    // save key
    *(int32_t*)(*result + offset) = keyLen;
    offset += sizeof(int32_t);
    memcpy(*result + offset, key, keyLen);
    offset += keyLen;

    // save value
    *(int32_t*)(*result + offset) = pSup->resultRowSize;
    offset += sizeof(int32_t);
1900
    memcpy(*result + offset, pRow, pSup->resultRowSize);
wmmhello's avatar
wmmhello 已提交
1901 1902 1903
    offset += pSup->resultRowSize;
  }

wmmhello's avatar
wmmhello 已提交
1904 1905 1906 1907
  *(int32_t*)(*result) = offset;
  *length = offset;

  return TDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
1908 1909
}

1910 1911 1912 1913 1914
int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf) {
  if (pLimitInfo->remainGroupOffset > 0) {
    if (pLimitInfo->currentGroupId == 0) {  // it is the first group
      pLimitInfo->currentGroupId = pBlock->info.groupId;
      blockDataCleanup(pBlock);
1915
      return PROJECT_RETRIEVE_CONTINUE;
1916 1917 1918
    } else if (pLimitInfo->currentGroupId != pBlock->info.groupId) {
      // now it is the data from a new group
      pLimitInfo->remainGroupOffset -= 1;
1919 1920

      // ignore data block in current group
1921 1922
      if (pLimitInfo->remainGroupOffset > 0) {
        blockDataCleanup(pBlock);
1923 1924 1925 1926 1927
        return PROJECT_RETRIEVE_CONTINUE;
      }
    }

    // set current group id of the project operator
1928
    pLimitInfo->currentGroupId = pBlock->info.groupId;
1929 1930
  }

1931
  // here check for a new group data, we need to handle the data of the previous group.
1932 1933 1934
  if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.groupId) {
    pLimitInfo->numOfOutputGroups += 1;
    if ((pLimitInfo->slimit.limit > 0) && (pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
1935
      pOperator->status = OP_EXEC_DONE;
1936
      blockDataCleanup(pBlock);
1937 1938 1939 1940 1941

      return PROJECT_RETRIEVE_DONE;
    }

    // reset the value for a new group data
1942 1943
    pLimitInfo->numOfOutputRows = 0;
    pLimitInfo->remainOffset = pLimitInfo->limit.offset;
1944 1945 1946 1947 1948

    // existing rows that belongs to previous group.
    if (pBlock->info.rows > 0) {
      return PROJECT_RETRIEVE_DONE;
    }
1949 1950 1951 1952 1953
  }

  // here we reach the start position, according to the limit/offset requirements.

  // set current group id
1954
  pLimitInfo->currentGroupId = pBlock->info.groupId;
1955

1956 1957 1958
  if (pLimitInfo->remainOffset >= pBlock->info.rows) {
    pLimitInfo->remainOffset -= pBlock->info.rows;
    blockDataCleanup(pBlock);
1959
    return PROJECT_RETRIEVE_CONTINUE;
1960 1961 1962
  } else if (pLimitInfo->remainOffset < pBlock->info.rows && pLimitInfo->remainOffset > 0) {
    blockDataTrimFirstNRows(pBlock, pLimitInfo->remainOffset);
    pLimitInfo->remainOffset = 0;
1963 1964
  }

1965
  // check for the limitation in each group
1966 1967 1968 1969
  if (pLimitInfo->limit.limit >= 0 && pLimitInfo->numOfOutputRows + pBlock->info.rows >= pLimitInfo->limit.limit) {
    int32_t keepRows = (int32_t)(pLimitInfo->limit.limit - pLimitInfo->numOfOutputRows);
    blockDataKeepFirstNRows(pBlock, keepRows);
    if (pLimitInfo->slimit.limit > 0 && pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups) {
1970 1971 1972
      pOperator->status = OP_EXEC_DONE;
    }

1973
    return PROJECT_RETRIEVE_DONE;
1974
  }
1975

1976
  // todo optimize performance
1977 1978
  // If there are slimit/soffset value exists, multi-round result can not be packed into one group, since the
  // they may not belong to the same group the limit/offset value is not valid in this case.
1979 1980
  if ((!holdDataInBuf) || (pBlock->info.rows >= pOperator->resultInfo.threshold) || pLimitInfo->slimit.offset != -1 ||
      pLimitInfo->slimit.limit != -1) {
1981
    return PROJECT_RETRIEVE_DONE;
L
Liu Jicong 已提交
1982
  } else {  // not full enough, continue to accumulate the output data in the buffer.
1983 1984 1985 1986
    return PROJECT_RETRIEVE_CONTINUE;
  }
}

1987
static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag);
L
Liu Jicong 已提交
1988 1989
static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
                                               SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
1990
  pInfo->totalInputRows = pInfo->existNewGroupBlock->info.rows;
1991 1992 1993 1994 1995
  SSDataBlock* pResBlock = pInfo->pFinalRes;

  int32_t order = TSDB_ORDER_ASC;
  int32_t scanFlag = MAIN_SCAN;
  getTableScanInfo(pOperator, &order, &scanFlag);
H
Haojun Liao 已提交
1996

L
Liu Jicong 已提交
1997 1998
  int64_t ekey =
      Q_STATUS_EQUAL(pTaskInfo->status, TASK_COMPLETED) ? pInfo->win.ekey : pInfo->existNewGroupBlock->info.window.ekey;
1999 2000
  taosResetFillInfo(pInfo->pFillInfo, getFillInfoStart(pInfo->pFillInfo));

2001
  blockDataCleanup(pInfo->pRes);
2002 2003 2004 2005
  doApplyScalarCalculation(pOperator, pInfo->existNewGroupBlock, order, scanFlag);

  taosFillSetStartInfo(pInfo->pFillInfo, pInfo->pRes->info.rows, ekey);
  taosFillSetInputDataBlock(pInfo->pFillInfo, pInfo->pRes);
2006

2007 2008
  int32_t numOfResultRows = pResultInfo->capacity - pResBlock->info.rows;
  taosFillResultDataBlock(pInfo->pFillInfo, pResBlock, numOfResultRows);
H
Haojun Liao 已提交
2009

2010
  pInfo->curGroupId = pInfo->existNewGroupBlock->info.groupId;
2011 2012 2013
  pInfo->existNewGroupBlock = NULL;
}

L
Liu Jicong 已提交
2014 2015
static void doHandleRemainBlockFromNewGroup(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
                                            SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
2016
  if (taosFillHasMoreResults(pInfo->pFillInfo)) {
H
Haojun Liao 已提交
2017 2018
    int32_t numOfResultRows = pResultInfo->capacity - pInfo->pFinalRes->info.rows;
    taosFillResultDataBlock(pInfo->pFillInfo, pInfo->pFinalRes, numOfResultRows);
2019 2020
    pInfo->pRes->info.groupId = pInfo->curGroupId;
    return;
2021 2022 2023 2024
  }

  // handle the cached new group data block
  if (pInfo->existNewGroupBlock) {
2025 2026 2027 2028 2029 2030
    doHandleRemainBlockForNewGroupImpl(pOperator, pInfo, pResultInfo, pTaskInfo);
  }
}

static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag) {
  SFillOperatorInfo* pInfo = pOperator->info;
L
Liu Jicong 已提交
2031
  SExprSupp*         pSup = &pOperator->exprSupp;
2032
  setInputDataBlock(pSup, pBlock, order, scanFlag, false);
2033 2034
  projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs, NULL);

2035 2036 2037 2038
  // reset the row value before applying the no-fill functions to the input data block, which is "pBlock" in this case.
  pInfo->pRes->info.rows = 0;
  SExprSupp* pNoFillSupp = &pInfo->noFillExprSupp;
  setInputDataBlock(pNoFillSupp, pBlock, order, scanFlag, false);
2039

2040 2041
  projectApplyFunctions(pNoFillSupp->pExprInfo, pInfo->pRes, pBlock, pNoFillSupp->pCtx, pNoFillSupp->numOfExprs, NULL);
  pInfo->pRes->info.groupId = pBlock->info.groupId;
2042 2043
}

S
slzhou 已提交
2044
static SSDataBlock* doFillImpl(SOperatorInfo* pOperator) {
L
Liu Jicong 已提交
2045 2046
  SFillOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
2047

H
Haojun Liao 已提交
2048
  SResultInfo* pResultInfo = &pOperator->resultInfo;
H
Haojun Liao 已提交
2049
  SSDataBlock* pResBlock = pInfo->pFinalRes;
2050 2051

  blockDataCleanup(pResBlock);
2052

H
Haojun Liao 已提交
2053 2054
  int32_t order = TSDB_ORDER_ASC;
  int32_t scanFlag = MAIN_SCAN;
2055
  getTableScanInfo(pOperator, &order, &scanFlag);
2056

2057
  doHandleRemainBlockFromNewGroup(pOperator, pInfo, pResultInfo, pTaskInfo);
2058
  if (pResBlock->info.rows > 0) {
2059
    pResBlock->info.groupId = pInfo->curGroupId;
2060
    return pResBlock;
H
Haojun Liao 已提交
2061
  }
2062

H
Haojun Liao 已提交
2063
  SOperatorInfo* pDownstream = pOperator->pDownstream[0];
L
Liu Jicong 已提交
2064
  while (1) {
2065
    SSDataBlock* pBlock = pDownstream->fpSet.getNextFn(pDownstream);
2066 2067
    if (pBlock == NULL) {
      if (pInfo->totalInputRows == 0) {
H
Haojun Liao 已提交
2068
        setOperatorCompleted(pOperator);
2069 2070
        return NULL;
      }
2071

2072
      taosFillSetStartInfo(pInfo->pFillInfo, 0, pInfo->win.ekey);
2073
    } else {
2074
      blockDataUpdateTsWindow(pBlock, pInfo->primarySrcSlotId);
2075 2076

      blockDataCleanup(pInfo->pRes);
2077 2078
      blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);
      blockDataEnsureCapacity(pInfo->pFinalRes, pBlock->info.rows);
2079
      doApplyScalarCalculation(pOperator, pBlock, order, scanFlag);
2080

H
Haojun Liao 已提交
2081 2082 2083
      if (pInfo->curGroupId == 0 || pInfo->curGroupId == pInfo->pRes->info.groupId) {
        pInfo->curGroupId = pInfo->pRes->info.groupId;  // the first data block
        pInfo->totalInputRows += pInfo->pRes->info.rows;
2084

H
Haojun Liao 已提交
2085 2086 2087 2088 2089
        if (order == pInfo->pFillInfo->order) {
          taosFillSetStartInfo(pInfo->pFillInfo, pInfo->pRes->info.rows, pBlock->info.window.ekey);
        } else {
          taosFillSetStartInfo(pInfo->pFillInfo, pInfo->pRes->info.rows, pBlock->info.window.skey);
        }
H
Haojun Liao 已提交
2090
        taosFillSetInputDataBlock(pInfo->pFillInfo, pInfo->pRes);
L
Liu Jicong 已提交
2091
      } else if (pInfo->curGroupId != pBlock->info.groupId) {  // the new group data block
2092 2093 2094 2095 2096
        pInfo->existNewGroupBlock = pBlock;

        // Fill the previous group data block, before handle the data block of new group.
        // Close the fill operation for previous group data block
        taosFillSetStartInfo(pInfo->pFillInfo, 0, pInfo->win.ekey);
2097 2098 2099
      }
    }

2100 2101
    int32_t numOfResultRows = pOperator->resultInfo.capacity - pResBlock->info.rows;
    taosFillResultDataBlock(pInfo->pFillInfo, pResBlock, numOfResultRows);
2102 2103

    // current group has no more result to return
2104
    if (pResBlock->info.rows > 0) {
2105 2106
      // 1. The result in current group not reach the threshold of output result, continue
      // 2. If multiple group results existing in one SSDataBlock is not allowed, return immediately
2107
      if (pResBlock->info.rows > pResultInfo->threshold || pBlock == NULL || pInfo->existNewGroupBlock != NULL) {
2108
        pResBlock->info.groupId = pInfo->curGroupId;
2109
        return pResBlock;
2110 2111
      }

2112
      doHandleRemainBlockFromNewGroup(pOperator, pInfo, pResultInfo, pTaskInfo);
2113
      if (pResBlock->info.rows >= pOperator->resultInfo.threshold || pBlock == NULL) {
2114
        pResBlock->info.groupId = pInfo->curGroupId;
2115
        return pResBlock;
2116 2117 2118
      }
    } else if (pInfo->existNewGroupBlock) {  // try next group
      assert(pBlock != NULL);
2119 2120 2121 2122

      blockDataCleanup(pResBlock);

      doHandleRemainBlockForNewGroupImpl(pOperator, pInfo, pResultInfo, pTaskInfo);
2123
      if (pResBlock->info.rows > pResultInfo->threshold) {
2124
        pResBlock->info.groupId = pInfo->curGroupId;
2125
        return pResBlock;
2126 2127 2128 2129 2130 2131 2132
      }
    } else {
      return NULL;
    }
  }
}

S
slzhou 已提交
2133 2134 2135 2136 2137 2138 2139 2140
static SSDataBlock* doFill(SOperatorInfo* pOperator) {
  SFillOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;

  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

S
slzhou 已提交
2141
  SSDataBlock* fillResult = NULL;
S
slzhou 已提交
2142
  while (true) {
S
slzhou 已提交
2143
    fillResult = doFillImpl(pOperator);
S
slzhou 已提交
2144
    if (fillResult == NULL) {
H
Haojun Liao 已提交
2145
      setOperatorCompleted(pOperator);
S
slzhou 已提交
2146 2147 2148
      break;
    }

H
Haojun Liao 已提交
2149
    doFilter(fillResult, pOperator->exprSupp.pFilterInfo, &pInfo->matchInfo );
S
slzhou 已提交
2150 2151 2152 2153 2154
    if (fillResult->info.rows > 0) {
      break;
    }
  }

S
slzhou 已提交
2155
  if (fillResult != NULL) {
2156
    pOperator->resultInfo.totalRows += fillResult->info.rows;
S
slzhou 已提交
2157
  }
S
slzhou 已提交
2158

S
slzhou 已提交
2159
  return fillResult;
S
slzhou 已提交
2160 2161
}

2162
void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs) {
C
Cary Xu 已提交
2163 2164 2165 2166 2167
  for (int32_t i = 0; i < numOfExprs; ++i) {
    SExprInfo* pExprInfo = &pExpr[i];
    for (int32_t j = 0; j < pExprInfo->base.numOfParams; ++j) {
      if (pExprInfo->base.pParam[j].type == FUNC_PARAM_TYPE_COLUMN) {
        taosMemoryFreeClear(pExprInfo->base.pParam[j].pCol);
H
Haojun Liao 已提交
2168
      }
2169
    }
C
Cary Xu 已提交
2170 2171 2172

    taosMemoryFree(pExprInfo->base.pParam);
    taosMemoryFree(pExprInfo->pExpr);
H
Haojun Liao 已提交
2173 2174 2175
  }
}

5
54liuyao 已提交
2176
void destroyOperatorInfo(SOperatorInfo* pOperator) {
2177 2178 2179 2180
  if (pOperator == NULL) {
    return;
  }

2181
  if (pOperator->fpSet.closeFn != NULL) {
2182
    pOperator->fpSet.closeFn(pOperator->info);
2183 2184
  }

H
Haojun Liao 已提交
2185
  if (pOperator->pDownstream != NULL) {
L
Liu Jicong 已提交
2186
    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
H
Haojun Liao 已提交
2187
      destroyOperatorInfo(pOperator->pDownstream[i]);
2188 2189
    }

wafwerar's avatar
wafwerar 已提交
2190
    taosMemoryFreeClear(pOperator->pDownstream);
H
Haojun Liao 已提交
2191
    pOperator->numOfDownstream = 0;
2192 2193
  }

2194
  cleanupExprSupp(&pOperator->exprSupp);
wafwerar's avatar
wafwerar 已提交
2195
  taosMemoryFreeClear(pOperator);
2196 2197
}

2198 2199 2200 2201 2202 2203
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz) {
  *defaultPgsz = 4096;
  while (*defaultPgsz < rowSize * 4) {
    *defaultPgsz <<= 1u;
  }

2204
  // The default buffer for each operator in query is 10MB.
2205
  // at least four pages need to be in buffer
2206 2207
  // TODO: make this variable to be configurable.
  *defaultBufsz = 4096 * 2560;
2208 2209 2210 2211 2212 2213 2214
  if ((*defaultBufsz) <= (*defaultPgsz)) {
    (*defaultBufsz) = (*defaultPgsz) * 4;
  }

  return 0;
}

dengyihao's avatar
dengyihao 已提交
2215 2216
int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t keyBufSize,
                         const char* pKey) {
L
Liu Jicong 已提交
2217
  int32_t    code = 0;
2218 2219
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

2220
  pAggSup->currentPageId = -1;
dengyihao's avatar
dengyihao 已提交
2221 2222
  pAggSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
  pAggSup->keyBuf = taosMemoryCalloc(1, keyBufSize + POINTER_BYTES + sizeof(int64_t));
2223
  pAggSup->pResultRowHashTable = tSimpleHashInit(10, hashFn);
2224

H
Haojun Liao 已提交
2225
  if (pAggSup->keyBuf == NULL || pAggSup->pResultRowHashTable == NULL) {
2226 2227 2228
    return TSDB_CODE_OUT_OF_MEMORY;
  }

dengyihao's avatar
dengyihao 已提交
2229
  uint32_t defaultPgsz = 0;
2230 2231
  uint32_t defaultBufsz = 0;
  getBufferPgSize(pAggSup->resultRowSize, &defaultPgsz, &defaultBufsz);
H
Haojun Liao 已提交
2232

wafwerar's avatar
wafwerar 已提交
2233
  if (!osTempSpaceAvailable()) {
H
Haojun Liao 已提交
2234 2235 2236
    code = TSDB_CODE_NO_AVAIL_DISK;
    qError("Init stream agg supporter failed since %s, %s", terrstr(code), pKey);
    return code;
wafwerar's avatar
wafwerar 已提交
2237
  }
2238

H
Haojun Liao 已提交
2239
  code = createDiskbasedBuf(&pAggSup->pResultBuf, defaultPgsz, defaultBufsz, pKey, tsTempDir);
H
Haojun Liao 已提交
2240
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2241
    qError("Create agg result buf failed since %s, %s", tstrerror(code), pKey);
H
Haojun Liao 已提交
2242 2243 2244
    return code;
  }

H
Haojun Liao 已提交
2245
  return code;
2246 2247
}

2248
void cleanupAggSup(SAggSupporter* pAggSup) {
wafwerar's avatar
wafwerar 已提交
2249
  taosMemoryFreeClear(pAggSup->keyBuf);
2250
  tSimpleHashCleanup(pAggSup->pResultRowHashTable);
H
Haojun Liao 已提交
2251
  destroyDiskbasedBuf(pAggSup->pResultBuf);
2252 2253
}

L
Liu Jicong 已提交
2254 2255
int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                    const char* pkey) {
2256 2257 2258 2259 2260
  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

2261 2262 2263 2264 2265
  code = doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

L
Liu Jicong 已提交
2266
  for (int32_t i = 0; i < numOfCols; ++i) {
2267
    pSup->pCtx[i].saveHandle.pBuf = pAggSup->pResultBuf;
2268 2269
  }

2270
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2271 2272
}

L
Liu Jicong 已提交
2273
void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows) {
wmmhello's avatar
wmmhello 已提交
2274
  ASSERT(numOfRows != 0);
2275 2276
  pResultInfo->capacity = numOfRows;
  pResultInfo->threshold = numOfRows * 0.75;
2277

2278 2279
  if (pResultInfo->threshold == 0) {
    pResultInfo->threshold = numOfRows;
2280 2281 2282
  }
}

2283 2284 2285 2286 2287
void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock) {
  pInfo->pRes = pBlock;
  initResultRowInfo(&pInfo->resultRowInfo);
}

5
54liuyao 已提交
2288
void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
2289 2290 2291 2292 2293 2294 2295 2296 2297 2298
  if (pCtx == NULL) {
    return NULL;
  }

  for (int32_t i = 0; i < numOfOutput; ++i) {
    for (int32_t j = 0; j < pCtx[i].numOfParams; ++j) {
      taosVariantDestroy(&pCtx[i].param[j].param);
    }

    taosMemoryFreeClear(pCtx[i].subsidiaries.pCtx);
2299
    taosMemoryFreeClear(pCtx[i].subsidiaries.buf);
2300 2301 2302 2303 2304 2305 2306 2307
    taosMemoryFree(pCtx[i].input.pData);
    taosMemoryFree(pCtx[i].input.pColumnDataAgg);
  }

  taosMemoryFreeClear(pCtx);
  return NULL;
}

2308
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr) {
2309 2310 2311 2312
  pSup->pExprInfo = pExprInfo;
  pSup->numOfExprs = numOfExpr;
  if (pSup->pExprInfo != NULL) {
    pSup->pCtx = createSqlFunctionCtx(pExprInfo, numOfExpr, &pSup->rowEntryInfoOffset);
2313 2314 2315
    if (pSup->pCtx == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
2316
  }
2317 2318

  return TSDB_CODE_SUCCESS;
2319 2320
}

2321 2322 2323 2324
void cleanupExprSupp(SExprSupp* pSupp) {
  destroySqlFunctionCtx(pSupp->pCtx, pSupp->numOfExprs);
  if (pSupp->pExprInfo != NULL) {
    destroyExprInfo(pSupp->pExprInfo, pSupp->numOfExprs);
C
Cary Xu 已提交
2325
    taosMemoryFreeClear(pSupp->pExprInfo);
2326
  }
H
Haojun Liao 已提交
2327 2328 2329 2330 2331 2332

  if (pSupp->pFilterInfo != NULL) {
    filterFreeInfo(pSupp->pFilterInfo);
    pSupp->pFilterInfo = NULL;
  }

2333 2334 2335
  taosMemoryFree(pSupp->rowEntryInfoOffset);
}

2336 2337
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode,
                                           SExecTaskInfo* pTaskInfo) {
wafwerar's avatar
wafwerar 已提交
2338
  SAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SAggOperatorInfo));
L
Liu Jicong 已提交
2339
  SOperatorInfo*    pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
2340 2341 2342
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
H
Haojun Liao 已提交
2343

H
Haojun Liao 已提交
2344 2345 2346 2347
  SSDataBlock* pResBlock = createResDataBlock(pAggNode->node.pOutputDataBlockDesc);
  initBasicInfo(&pInfo->binfo, pResBlock);

  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
2348
  initResultSizeInfo(&pOperator->resultInfo, 4096);
H
Haojun Liao 已提交
2349

2350 2351 2352
  int32_t    num = 0;
  SExprInfo* pExprInfo = createExprInfo(pAggNode->pAggFuncs, pAggNode->pGroupKeys, &num);
  int32_t    code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str);
L
Liu Jicong 已提交
2353
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2354 2355
    goto _error;
  }
H
Haojun Liao 已提交
2356

H
Haojun Liao 已提交
2357 2358 2359 2360 2361 2362
  int32_t    numOfScalarExpr = 0;
  SExprInfo* pScalarExprInfo = NULL;
  if (pAggNode->pExprs != NULL) {
    pScalarExprInfo = createExprInfo(pAggNode->pExprs, NULL, &numOfScalarExpr);
  }

2363 2364 2365 2366
  code = initExprSupp(&pInfo->scalarExprSup, pScalarExprInfo, numOfScalarExpr);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2367

H
Haojun Liao 已提交
2368 2369 2370 2371 2372
  code = filterInitFromNode((SNode*)pAggNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
2373
  pInfo->binfo.mergeResultBlock = pAggNode->mergeDataBlock;
2374
  pInfo->groupId = UINT64_MAX;
H
Haojun Liao 已提交
2375

H
Haojun Liao 已提交
2376
  setOperatorInfo(pOperator, "TableAggregate", QUERY_NODE_PHYSICAL_PLAN_HASH_AGG, true, OP_NOT_OPENED, pInfo, pTaskInfo);
5
54liuyao 已提交
2377
  pOperator->fpSet =
H
Haojun Liao 已提交
2378
      createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, destroyAggOperatorInfo, NULL);
H
Haojun Liao 已提交
2379

2380 2381 2382 2383 2384 2385
  if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
    STableScanInfo* pTableScanInfo = downstream->info;
    pTableScanInfo->pdInfo.pExprSup = &pOperator->exprSupp;
    pTableScanInfo->pdInfo.pAggSup = &pInfo->aggSup;
  }

H
Haojun Liao 已提交
2386 2387 2388 2389
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2390 2391

  return pOperator;
H
Haojun Liao 已提交
2392

2393
_error:
H
Haojun Liao 已提交
2394 2395 2396 2397
  if (pInfo != NULL) {
    destroyAggOperatorInfo(pInfo);
  }

2398 2399 2400
  if (pOperator != NULL) {
    cleanupExprSupp(&pOperator->exprSupp);
  }
H
Haojun Liao 已提交
2401

2402
  taosMemoryFreeClear(pOperator);
H
Haojun Liao 已提交
2403
  pTaskInfo->code = code;
H
Haojun Liao 已提交
2404
  return NULL;
2405 2406
}

2407
void cleanupBasicInfo(SOptrBasicInfo* pInfo) {
2408
  assert(pInfo != NULL);
H
Haojun Liao 已提交
2409
  pInfo->pRes = blockDataDestroy(pInfo->pRes);
2410 2411
}

H
Haojun Liao 已提交
2412 2413 2414 2415 2416 2417 2418
static void freeItem(void* pItem) {
  void** p = pItem;
  if (*p != NULL) {
    taosMemoryFreeClear(*p);
  }
}

2419
void destroyAggOperatorInfo(void* param) {
L
Liu Jicong 已提交
2420
  SAggOperatorInfo* pInfo = (SAggOperatorInfo*)param;
L
Liu Jicong 已提交
2421 2422
  cleanupBasicInfo(&pInfo->binfo);

H
Haojun Liao 已提交
2423
  cleanupAggSup(&pInfo->aggSup);
S
shenglian zhou 已提交
2424
  cleanupExprSupp(&pInfo->scalarExprSup);
H
Haojun Liao 已提交
2425
  cleanupGroupResInfo(&pInfo->groupResInfo);
D
dapan1121 已提交
2426
  taosMemoryFreeClear(param);
2427
}
2428

2429
void destroyFillOperatorInfo(void* param) {
L
Liu Jicong 已提交
2430
  SFillOperatorInfo* pInfo = (SFillOperatorInfo*)param;
2431
  pInfo->pFillInfo = taosDestroyFillInfo(pInfo->pFillInfo);
H
Haojun Liao 已提交
2432
  pInfo->pRes = blockDataDestroy(pInfo->pRes);
H
Haojun Liao 已提交
2433 2434
  pInfo->pFinalRes = blockDataDestroy(pInfo->pFinalRes);

2435
  cleanupExprSupp(&pInfo->noFillExprSupp);
H
Haojun Liao 已提交
2436

wafwerar's avatar
wafwerar 已提交
2437
  taosMemoryFreeClear(pInfo->p);
H
Haojun Liao 已提交
2438
  taosArrayDestroy(pInfo->matchInfo.pList);
D
dapan1121 已提交
2439
  taosMemoryFreeClear(param);
2440 2441
}

H
Haojun Liao 已提交
2442 2443 2444 2445
static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t numOfCols, SExprInfo* pNotFillExpr,
                            int32_t numOfNotFillCols, SNodeListNode* pValNode, STimeWindow win, int32_t capacity,
                            const char* id, SInterval* pInterval, int32_t fillType, int32_t order) {
  SFillColInfo* pColInfo = createFillColInfo(pExpr, numOfCols, pNotFillExpr, numOfNotFillCols, pValNode);
H
Haojun Liao 已提交
2446

H
Haojun Liao 已提交
2447 2448 2449
  int64_t     startKey = (order == TSDB_ORDER_ASC) ? win.skey : win.ekey;
  STimeWindow w = getAlignQueryTimeWindow(pInterval, pInterval->precision, startKey);
  w = getFirstQualifiedTimeWindow(startKey, &w, pInterval, order);
H
Haojun Liao 已提交
2450

L
Liu Jicong 已提交
2451 2452
  pInfo->pFillInfo = taosCreateFillInfo(w.skey, numOfCols, numOfNotFillCols, capacity, pInterval, fillType, pColInfo,
                                        pInfo->primaryTsCol, order, id);
H
Haojun Liao 已提交
2453

H
Haojun Liao 已提交
2454 2455 2456 2457 2458 2459 2460
  if (order == TSDB_ORDER_ASC) {
    pInfo->win.skey = win.skey;
    pInfo->win.ekey = win.ekey;
  } else {
    pInfo->win.skey = win.ekey;
    pInfo->win.ekey = win.skey;
  }
L
Liu Jicong 已提交
2461
  pInfo->p = taosMemoryCalloc(numOfCols, POINTER_BYTES);
2462

H
Haojun Liao 已提交
2463
  if (pInfo->pFillInfo == NULL || pInfo->p == NULL) {
H
Haojun Liao 已提交
2464 2465
    taosMemoryFree(pInfo->pFillInfo);
    taosMemoryFree(pInfo->p);
H
Haojun Liao 已提交
2466 2467 2468 2469 2470 2471
    return TSDB_CODE_OUT_OF_MEMORY;
  } else {
    return TSDB_CODE_SUCCESS;
  }
}

2472
static bool isWstartColumnExist(SFillOperatorInfo* pInfo) {
2473
  if (pInfo->noFillExprSupp.numOfExprs == 0) {
2474 2475
    return false;
  }
2476 2477 2478

  for (int32_t i = 0; i < pInfo->noFillExprSupp.numOfExprs; ++i) {
    SExprInfo* exprInfo = pInfo->noFillExprSupp.pExprInfo + i;
2479
    if (exprInfo->pExpr->nodeType == QUERY_NODE_COLUMN && exprInfo->base.numOfParams == 1 &&
2480 2481 2482 2483 2484 2485 2486
        exprInfo->base.pParam[0].pCol->colType == COLUMN_TYPE_WINDOW_START) {
      return true;
    }
  }
  return false;
}

2487 2488
static int32_t createPrimaryTsExprIfNeeded(SFillOperatorInfo* pInfo, SFillPhysiNode* pPhyFillNode, SExprSupp* pExprSupp,
                                           const char* idStr) {
2489
  bool wstartExist = isWstartColumnExist(pInfo);
2490

2491 2492
  if (wstartExist == false) {
    if (pPhyFillNode->pWStartTs->type != QUERY_NODE_TARGET) {
2493
      qError("pWStartTs of fill physical node is not a target node, %s", idStr);
2494 2495 2496
      return TSDB_CODE_QRY_SYS_ERROR;
    }

2497 2498
    SExprInfo* pExpr = taosMemoryRealloc(pExprSupp->pExprInfo, (pExprSupp->numOfExprs + 1) * sizeof(SExprInfo));
    if (pExpr == NULL) {
2499 2500 2501
      return TSDB_CODE_OUT_OF_MEMORY;
    }

2502 2503 2504
    createExprFromTargetNode(&pExpr[pExprSupp->numOfExprs], (STargetNode*)pPhyFillNode->pWStartTs);
    pExprSupp->numOfExprs += 1;
    pExprSupp->pExprInfo = pExpr;
2505
  }
2506

2507 2508 2509
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
2510 2511
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode,
                                      SExecTaskInfo* pTaskInfo) {
2512 2513 2514 2515 2516 2517
  SFillOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SFillOperatorInfo));
  SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

H
Haojun Liao 已提交
2518 2519 2520 2521
  pInfo->pRes = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc);
  SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &pInfo->numOfExpr);
  pOperator->exprSupp.pExprInfo = pExprInfo;

2522 2523 2524 2525 2526 2527 2528 2529
  SExprSupp* pNoFillSupp = &pInfo->noFillExprSupp;
  pNoFillSupp->pExprInfo = createExprInfo(pPhyFillNode->pNotFillExprs, NULL, &pNoFillSupp->numOfExprs);
  int32_t code = createPrimaryTsExprIfNeeded(pInfo, pPhyFillNode, pNoFillSupp, pTaskInfo->id.str);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  code = initExprSupp(pNoFillSupp, pNoFillSupp->pExprInfo, pNoFillSupp->numOfExprs);
2530 2531 2532
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
2533

L
Liu Jicong 已提交
2534
  SInterval* pInterval =
2535
      QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == downstream->operatorType
2536 2537
          ? &((SMergeAlignedIntervalAggOperatorInfo*)downstream->info)->intervalAggOperatorInfo->interval
          : &((SIntervalAggOperatorInfo*)downstream->info)->interval;
2538

2539
  int32_t order = (pPhyFillNode->inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
2540
  int32_t type = convertFillType(pPhyFillNode->mode);
2541

H
Haojun Liao 已提交
2542
  SResultInfo* pResultInfo = &pOperator->resultInfo;
H
Haojun Liao 已提交
2543

2544
  initResultSizeInfo(&pOperator->resultInfo, 4096);
H
Haojun Liao 已提交
2545 2546 2547 2548 2549
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
  code = initExprSupp(&pOperator->exprSupp, pExprInfo, pInfo->numOfExpr);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
2550

H
Haojun Liao 已提交
2551 2552
  pInfo->primaryTsCol = ((STargetNode*)pPhyFillNode->pWStartTs)->slotId;
  pInfo->primarySrcSlotId = ((SColumnNode*)((STargetNode*)pPhyFillNode->pWStartTs)->pExpr)->slotId;
2553

2554
  int32_t numOfOutputCols = 0;
H
Haojun Liao 已提交
2555 2556
  code = extractColMatchInfo(pPhyFillNode->pFillExprs, pPhyFillNode->node.pOutputDataBlockDesc, &numOfOutputCols,
                             COL_MATCH_FROM_SLOT_ID, &pInfo->matchInfo);
2557

2558
  code = initFillInfo(pInfo, pExprInfo, pInfo->numOfExpr, pNoFillSupp->pExprInfo, pNoFillSupp->numOfExprs,
2559 2560
                      (SNodeListNode*)pPhyFillNode->pValues, pPhyFillNode->timeRange, pResultInfo->capacity,
                      pTaskInfo->id.str, pInterval, type, order);
2561 2562 2563
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2564

H
Haojun Liao 已提交
2565
  pInfo->pFinalRes = createOneDataBlock(pInfo->pRes, false);
H
Haojun Liao 已提交
2566 2567
  blockDataEnsureCapacity(pInfo->pFinalRes, pOperator->resultInfo.capacity);

H
Haojun Liao 已提交
2568 2569 2570 2571 2572
  code = filterInitFromNode((SNode*)pPhyFillNode->node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
2573
  setOperatorInfo(pOperator, "FillOperator", QUERY_NODE_PHYSICAL_PLAN_FILL, false, OP_NOT_OPENED, pInfo, pTaskInfo);
2574
  pOperator->exprSupp.numOfExprs = pInfo->numOfExpr;
H
Haojun Liao 已提交
2575
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, destroyFillOperatorInfo, NULL);
2576

2577
  code = appendDownstream(pOperator, &downstream, 1);
2578
  return pOperator;
H
Haojun Liao 已提交
2579

2580
_error:
H
Haojun Liao 已提交
2581 2582 2583 2584
  if (pInfo != NULL) {
    destroyFillOperatorInfo(pInfo);
  }

2585
  pTaskInfo->code = code;
wafwerar's avatar
wafwerar 已提交
2586
  taosMemoryFreeClear(pOperator);
H
Haojun Liao 已提交
2587
  return NULL;
2588 2589
}

D
dapan1121 已提交
2590
static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPTR_EXEC_MODEL model, char* dbFName) {
wafwerar's avatar
wafwerar 已提交
2591
  SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo));
H
Haojun Liao 已提交
2592 2593 2594 2595 2596
  if (pTaskInfo == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

2597
  setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
H
Haojun Liao 已提交
2598

2599
  pTaskInfo->schemaInfo.dbname = strdup(dbFName);
2600
  pTaskInfo->cost.created = taosGetTimestampMs();
H
Haojun Liao 已提交
2601
  pTaskInfo->id.queryId = queryId;
dengyihao's avatar
dengyihao 已提交
2602
  pTaskInfo->execModel = model;
H
Haojun Liao 已提交
2603
  pTaskInfo->pTableInfoList = tableListCreate();
D
dapan1121 已提交
2604
  pTaskInfo->stopInfo.pStopInfo = taosArrayInit(4, sizeof(SExchangeOpStopInfo));
H
Haojun Liao 已提交
2605

wafwerar's avatar
wafwerar 已提交
2606
  char* p = taosMemoryCalloc(1, 128);
L
Liu Jicong 已提交
2607
  snprintf(p, 128, "TID:0x%" PRIx64 " QID:0x%" PRIx64, taskId, queryId);
H
Haojun Liao 已提交
2608
  pTaskInfo->id.str = p;
H
Haojun Liao 已提交
2609

2610 2611
  return pTaskInfo;
}
H
Haojun Liao 已提交
2612

H
Haojun Liao 已提交
2613 2614
SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode);

S
shenglian zhou 已提交
2615 2616
SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* handle, STableCountScanPhysiNode* pNode,
                                                SExecTaskInfo* pTaskInfo);
2617
int32_t extractTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, SExecTaskInfo* pTaskInfo) {
2618 2619
  SMetaReader mr = {0};
  metaReaderInit(&mr, pHandle->meta, 0);
2620
  int32_t code = metaGetTableEntryByUid(&mr, pScanNode->uid);
2621
  if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
2622 2623
    qError("failed to get the table meta, uid:0x%" PRIx64 ", suid:0x%" PRIx64 ", %s", pScanNode->uid, pScanNode->suid,
           GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
2624

D
dapan1121 已提交
2625
    metaReaderClear(&mr);
2626
    return terrno;
D
dapan1121 已提交
2627
  }
2628

2629 2630
  SSchemaInfo* pSchemaInfo = &pTaskInfo->schemaInfo;
  pSchemaInfo->tablename = strdup(mr.me.name);
2631 2632

  if (mr.me.type == TSDB_SUPER_TABLE) {
2633 2634
    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow);
    pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version;
2635
  } else if (mr.me.type == TSDB_CHILD_TABLE) {
2636 2637
    tDecoderClear(&mr.coder);

2638 2639
    tb_uid_t suid = mr.me.ctbEntry.suid;
    metaGetTableEntryByUid(&mr, suid);
2640 2641
    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow);
    pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version;
2642
  } else {
2643
    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.ntbEntry.schemaRow);
2644
  }
2645 2646

  metaReaderClear(&mr);
2647

H
Haojun Liao 已提交
2648 2649 2650 2651 2652
  pSchemaInfo->qsw = extractQueriedColumnSchema(pScanNode);
  return TSDB_CODE_SUCCESS;
}

SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode) {
2653 2654 2655
  int32_t numOfCols = LIST_LENGTH(pScanNode->pScanCols);
  int32_t numOfTags = LIST_LENGTH(pScanNode->pScanPseudoCols);

2656
  SSchemaWrapper* pqSw = taosMemoryCalloc(1, sizeof(SSchemaWrapper));
2657
  pqSw->pSchema = taosMemoryCalloc(numOfCols + numOfTags, sizeof(SSchema));
2658

L
Liu Jicong 已提交
2659
  for (int32_t i = 0; i < numOfCols; ++i) {
H
Haojun Liao 已提交
2660
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pScanNode->pScanCols, i);
2661 2662
    SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;

H
Haojun Liao 已提交
2663 2664 2665
    SSchema* pSchema = &pqSw->pSchema[pqSw->nCols++];
    pSchema->colId = pColNode->colId;
    pSchema->type = pColNode->node.resType.type;
H
Haojun Liao 已提交
2666 2667
    pSchema->bytes = pColNode->node.resType.bytes;
    tstrncpy(pSchema->name, pColNode->colName, tListLen(pSchema->name));
2668 2669
  }

2670
  // this the tags and pseudo function columns, we only keep the tag columns
2671
  for (int32_t i = 0; i < numOfTags; ++i) {
2672 2673 2674 2675 2676 2677 2678 2679 2680
    STargetNode* pNode = (STargetNode*)nodesListGetNode(pScanNode->pScanPseudoCols, i);

    int32_t type = nodeType(pNode->pExpr);
    if (type == QUERY_NODE_COLUMN) {
      SColumnNode* pColNode = (SColumnNode*)pNode->pExpr;

      SSchema* pSchema = &pqSw->pSchema[pqSw->nCols++];
      pSchema->colId = pColNode->colId;
      pSchema->type = pColNode->node.resType.type;
H
Haojun Liao 已提交
2681
      pSchema->bytes = pColNode->node.resType.bytes;
H
Haojun Liao 已提交
2682
      tstrncpy(pSchema->name, pColNode->colName, tListLen(pSchema->name));
2683 2684 2685
    }
  }

H
Haojun Liao 已提交
2686
  return pqSw;
2687 2688
}

2689 2690
static void cleanupTableSchemaInfo(SSchemaInfo* pSchemaInfo) {
  taosMemoryFreeClear(pSchemaInfo->dbname);
2691
  taosMemoryFreeClear(pSchemaInfo->tablename);
2692 2693
  tDeleteSSchemaWrapper(pSchemaInfo->sw);
  tDeleteSSchemaWrapper(pSchemaInfo->qsw);
2694 2695
}

2696
static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) { tDeleteSSchemaWrapper(pStreamInfo->schema); }
2697

2698
bool groupbyTbname(SNodeList* pGroupList) {
2699
  bool bytbname = false;
2700
  if (LIST_LENGTH(pGroupList) == 1) {
2701 2702 2703 2704 2705 2706 2707 2708 2709 2710
    SNode* p = nodesListGetNode(pGroupList, 0);
    if (p->type == QUERY_NODE_FUNCTION) {
      // partition by tbname/group by tbname
      bytbname = (strcmp(((struct SFunctionNode*)p)->functionName, "tbname") == 0);
    }
  }

  return bytbname;
}

2711 2712
SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SNode* pTagCond,
                                  SNode* pTagIndexCond, const char* pUser) {
2713
  int32_t         type = nodeType(pPhyNode);
2714
  STableListInfo* pTableListInfo = pTaskInfo->pTableInfoList;
2715
  const char*     idstr = GET_TASKID(pTaskInfo);
2716

X
Xiaoyu Wang 已提交
2717
  if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
2718
    SOperatorInfo* pOperator = NULL;
H
Haojun Liao 已提交
2719
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
dengyihao's avatar
dengyihao 已提交
2720
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
H
Haojun Liao 已提交
2721

2722 2723 2724 2725 2726 2727
      // NOTE: this is an patch to fix the physical plan
      // TODO remove it later
      if (pTableScanNode->scan.node.pLimit != NULL) {
        pTableScanNode->groupSort = true;
      }

L
Liu Jicong 已提交
2728 2729
      int32_t code =
          createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort, pHandle,
H
Haojun Liao 已提交
2730
                                  pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
2731
      if (code) {
wmmhello's avatar
wmmhello 已提交
2732
        pTaskInfo->code = code;
2733
        qError("failed to createScanTableListInfo, code:%s, %s", tstrerror(code), idstr);
D
dapan1121 已提交
2734 2735
        return NULL;
      }
wmmhello's avatar
wmmhello 已提交
2736

2737
      code = extractTableSchemaInfo(pHandle, &pTableScanNode->scan, pTaskInfo);
S
slzhou 已提交
2738
      if (code) {
2739
        pTaskInfo->code = terrno;
wmmhello's avatar
wmmhello 已提交
2740 2741 2742
        return NULL;
      }

2743
      pOperator = createTableScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo);
D
dapan1121 已提交
2744 2745 2746 2747 2748
      if (NULL == pOperator) {
        pTaskInfo->code = terrno;
        return NULL;
      }

2749 2750
      STableScanInfo* pScanInfo = pOperator->info;
      pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
S
slzhou 已提交
2751 2752
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
      STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
H
Haojun Liao 已提交
2753 2754 2755

      int32_t code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, true, pHandle,
                                             pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
L
Liu Jicong 已提交
2756
      if (code) {
wmmhello's avatar
wmmhello 已提交
2757
        pTaskInfo->code = code;
H
Haojun Liao 已提交
2758
        qError("failed to createScanTableListInfo, code: %s", tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2759 2760
        return NULL;
      }
2761

2762
      code = extractTableSchemaInfo(pHandle, &pTableScanNode->scan, pTaskInfo);
wmmhello's avatar
wmmhello 已提交
2763 2764 2765 2766
      if (code) {
        pTaskInfo->code = terrno;
        return NULL;
      }
wmmhello's avatar
wmmhello 已提交
2767

2768
      pOperator = createTableMergeScanOperatorInfo(pTableScanNode, pTableListInfo, pHandle, pTaskInfo);
D
dapan1121 已提交
2769 2770 2771 2772
      if (NULL == pOperator) {
        pTaskInfo->code = terrno;
        return NULL;
      }
wmmhello's avatar
wmmhello 已提交
2773

2774 2775
      STableScanInfo* pScanInfo = pOperator->info;
      pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
H
Haojun Liao 已提交
2776
    } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
2777 2778
      pOperator = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode,
                                             pTaskInfo);
H
Haojun Liao 已提交
2779
    } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
5
54liuyao 已提交
2780
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
5
54liuyao 已提交
2781
      if (pHandle->vnode) {
L
Liu Jicong 已提交
2782 2783
        int32_t code =
            createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
H
Haojun Liao 已提交
2784
                                    pHandle, pTableListInfo, pTagCond, pTagIndexCond, pTaskInfo);
L
Liu Jicong 已提交
2785
        if (code) {
wmmhello's avatar
wmmhello 已提交
2786
          pTaskInfo->code = code;
H
Haojun Liao 已提交
2787
          qError("failed to createScanTableListInfo, code: %s", tstrerror(code));
wmmhello's avatar
wmmhello 已提交
2788 2789
          return NULL;
        }
L
Liu Jicong 已提交
2790 2791

#ifndef NDEBUG
H
Haojun Liao 已提交
2792
        int32_t sz = tableListGetSize(pTableListInfo);
H
Haojun Liao 已提交
2793 2794
        qDebug("create stream task, total:%d", sz);

L
Liu Jicong 已提交
2795
        for (int32_t i = 0; i < sz; i++) {
H
Haojun Liao 已提交
2796
          STableKeyInfo* pKeyInfo = tableListGetInfo(pTableListInfo, i);
2797
          qDebug("add table uid:%" PRIu64 ", gid:%" PRIu64, pKeyInfo->uid, pKeyInfo->groupId);
L
Liu Jicong 已提交
2798 2799
        }
#endif
2800
      }
2801

H
Haojun Liao 已提交
2802
      pTaskInfo->schemaInfo.qsw = extractQueriedColumnSchema(&pTableScanNode->scan);
2803
      pOperator = createStreamScanOperatorInfo(pHandle, pTableScanNode, pTagCond, pTaskInfo);
H
Haojun Liao 已提交
2804
    } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
L
Liu Jicong 已提交
2805
      SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode;
2806
      pOperator = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pUser, pTaskInfo);
S
shenglian zhou 已提交
2807 2808 2809
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN == type) {
      STableCountScanPhysiNode* pTblCountScanNode = (STableCountScanPhysiNode*)pPhyNode;
      pOperator = createTableCountScanOperatorInfo(pHandle, pTblCountScanNode, pTaskInfo);
2810
    } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) {
X
Xiaoyu Wang 已提交
2811
      STagScanPhysiNode* pScanPhyNode = (STagScanPhysiNode*)pPhyNode;
2812 2813

      int32_t code = createScanTableListInfo(pScanPhyNode, NULL, false, pHandle, pTableListInfo, pTagCond,
H
Haojun Liao 已提交
2814
                                             pTagIndexCond, pTaskInfo);
2815
      if (code != TSDB_CODE_SUCCESS) {
2816
        pTaskInfo->code = code;
H
Haojun Liao 已提交
2817
        qError("failed to getTableList, code: %s", tstrerror(code));
2818 2819 2820
        return NULL;
      }

2821
      pOperator = createTagScanOperatorInfo(pHandle, pScanPhyNode, pTableListInfo, pTaskInfo);
2822
    } else if (QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN == type) {
2823
      SBlockDistScanPhysiNode* pBlockNode = (SBlockDistScanPhysiNode*)pPhyNode;
2824 2825

      if (pBlockNode->tableType == TSDB_SUPER_TABLE) {
H
Haojun Liao 已提交
2826 2827
        SArray* pList = taosArrayInit(4, sizeof(STableKeyInfo));
        int32_t code = vnodeGetAllTableList(pHandle->vnode, pBlockNode->uid, pList);
2828 2829 2830 2831
        if (code != TSDB_CODE_SUCCESS) {
          pTaskInfo->code = terrno;
          return NULL;
        }
H
Haojun Liao 已提交
2832

2833
        for (int32_t i = 0; i < tableListGetSize(pTableListInfo); ++i) {
H
Haojun Liao 已提交
2834
          STableKeyInfo* p = taosArrayGet(pList, i);
H
Haojun Liao 已提交
2835
          tableListAddTableInfo(pTableListInfo, p->uid, 0);
H
Haojun Liao 已提交
2836 2837
        }
        taosArrayDestroy(pList);
2838
      } else {  // Create group with only one table
H
Haojun Liao 已提交
2839
        tableListAddTableInfo(pTableListInfo, pBlockNode->uid, 0);
2840 2841
      }

2842
      pOperator = createDataBlockInfoScanOperator(pHandle, pBlockNode, pTaskInfo);
H
Haojun Liao 已提交
2843 2844 2845
    } else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) {
      SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode;

L
Liu Jicong 已提交
2846
      int32_t code = createScanTableListInfo(&pScanNode->scan, pScanNode->pGroupTags, true, pHandle, pTableListInfo,
H
Haojun Liao 已提交
2847
                                             pTagCond, pTagIndexCond, pTaskInfo);
2848 2849 2850 2851
      if (code != TSDB_CODE_SUCCESS) {
        pTaskInfo->code = code;
        return NULL;
      }
2852

2853
      code = extractTableSchemaInfo(pHandle, &pScanNode->scan, pTaskInfo);
2854 2855 2856
      if (code != TSDB_CODE_SUCCESS) {
        pTaskInfo->code = code;
        return NULL;
H
Haojun Liao 已提交
2857 2858
      }

2859
      pOperator = createCacherowsScanOperator(pScanNode, pHandle, pTaskInfo);
2860
    } else if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
2861
      pOperator = createProjectOperatorInfo(NULL, (SProjectPhysiNode*)pPhyNode, pTaskInfo);
H
Haojun Liao 已提交
2862 2863
    } else {
      ASSERT(0);
H
Haojun Liao 已提交
2864
    }
2865 2866 2867 2868 2869

    if (pOperator != NULL) {
      pOperator->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
    }

2870
    return pOperator;
H
Haojun Liao 已提交
2871 2872
  }

2873
  size_t          size = LIST_LENGTH(pPhyNode->pChildren);
2874
  SOperatorInfo** ops = taosMemoryCalloc(size, POINTER_BYTES);
2875 2876 2877 2878
  if (ops == NULL) {
    return NULL;
  }

dengyihao's avatar
dengyihao 已提交
2879
  for (int32_t i = 0; i < size; ++i) {
2880
    SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, i);
2881
    ops[i] = createOperatorTree(pChildNode, pTaskInfo, pHandle, pTagCond, pTagIndexCond, pUser);
2882
    if (ops[i] == NULL) {
H
Haojun Liao 已提交
2883
      taosMemoryFree(ops);
2884 2885
      return NULL;
    }
2886
  }
H
Haojun Liao 已提交
2887

2888
  SOperatorInfo* pOptr = NULL;
H
Haojun Liao 已提交
2889
  if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
2890
    pOptr = createProjectOperatorInfo(ops[0], (SProjectPhysiNode*)pPhyNode, pTaskInfo);
2891
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_AGG == type) {
H
Haojun Liao 已提交
2892 2893
    SAggPhysiNode* pAggNode = (SAggPhysiNode*)pPhyNode;
    if (pAggNode->pGroupKeys != NULL) {
H
Haojun Liao 已提交
2894
      pOptr = createGroupOperatorInfo(ops[0], pAggNode, pTaskInfo);
H
Haojun Liao 已提交
2895
    } else {
H
Haojun Liao 已提交
2896
      pOptr = createAggregateOperatorInfo(ops[0], pAggNode, pTaskInfo);
H
Haojun Liao 已提交
2897
    }
2898
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type) {
H
Haojun Liao 已提交
2899
    SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
H
Haojun Liao 已提交
2900

H
Haojun Liao 已提交
2901 2902
    bool isStream = (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type);
    pOptr = createIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, isStream);
2903
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type) {
2904
    pOptr = createStreamIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo);
2905 2906
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) {
    SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode;
2907
    pOptr = createMergeAlignedIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo);
S
shenglian zhou 已提交
2908
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL == type) {
X
Xiaoyu Wang 已提交
2909
    SMergeIntervalPhysiNode* pIntervalPhyNode = (SMergeIntervalPhysiNode*)pPhyNode;
2910
    pOptr = createMergeIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo);
5
54liuyao 已提交
2911
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL == type) {
2912
    int32_t children = 0;
5
54liuyao 已提交
2913 2914
    pOptr = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL == type) {
5
54liuyao 已提交
2915
    int32_t children = pHandle->numOfVgroups;
5
54liuyao 已提交
2916
    pOptr = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
H
Haojun Liao 已提交
2917
  } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
2918
    pOptr = createSortOperatorInfo(ops[0], (SSortPhysiNode*)pPhyNode, pTaskInfo);
S
shenglian zhou 已提交
2919 2920
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT == type) {
    pOptr = createGroupSortOperatorInfo(ops[0], (SGroupSortPhysiNode*)pPhyNode, pTaskInfo);
X
Xiaoyu Wang 已提交
2921
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) {
2922
    SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode;
2923
    pOptr = createMultiwayMergeOperatorInfo(ops, size, pMergePhyNode, pTaskInfo);
2924
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION == type) {
H
Haojun Liao 已提交
2925
    SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
H
Haojun Liao 已提交
2926
    pOptr = createSessionAggOperatorInfo(ops[0], pSessionNode, pTaskInfo);
2927
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION == type) {
2928 2929 2930 2931 2932
    pOptr = createStreamSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo);
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION == type) {
    int32_t children = 0;
    pOptr = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION == type) {
2933
    int32_t children = pHandle->numOfVgroups;
2934
    pOptr = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
H
Haojun Liao 已提交
2935
  } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
2936
    pOptr = createPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo);
2937
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION == type) {
2938
    pOptr = createStreamPartitionOperatorInfo(ops[0], (SStreamPartitionPhysiNode*)pPhyNode, pTaskInfo);
2939
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE == type) {
dengyihao's avatar
dengyihao 已提交
2940
    SStateWinodwPhysiNode* pStateNode = (SStateWinodwPhysiNode*)pPhyNode;
2941
    pOptr = createStatewindowOperatorInfo(ops[0], pStateNode, pTaskInfo);
2942
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE == type) {
5
54liuyao 已提交
2943
    pOptr = createStreamStateAggOperatorInfo(ops[0], pPhyNode, pTaskInfo);
2944
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN == type) {
2945
    pOptr = createMergeJoinOperatorInfo(ops, size, (SSortMergeJoinPhysiNode*)pPhyNode, pTaskInfo);
2946
  } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) {
H
Haojun Liao 已提交
2947
    pOptr = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, pTaskInfo);
5
54liuyao 已提交
2948 2949
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL == type) {
    pOptr = createStreamFillOperatorInfo(ops[0], (SStreamFillPhysiNode*)pPhyNode, pTaskInfo);
H
Haojun Liao 已提交
2950 2951
  } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) {
    pOptr = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo);
2952 2953
  } else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) {
    pOptr = createTimeSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo);
H
Haojun Liao 已提交
2954 2955
  } else {
    ASSERT(0);
H
Haojun Liao 已提交
2956
  }
2957

2958
  taosMemoryFree(ops);
2959 2960 2961 2962
  if (pOptr) {
    pOptr->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
  }

2963
  return pOptr;
2964
}
H
Haojun Liao 已提交
2965

L
Liu Jicong 已提交
2966 2967 2968 2969 2970 2971 2972 2973 2974 2975 2976 2977 2978
static int32_t extractTbscanInStreamOpTree(SOperatorInfo* pOperator, STableScanInfo** ppInfo) {
  if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
    if (pOperator->numOfDownstream == 0) {
      qError("failed to find stream scan operator");
      return TSDB_CODE_QRY_APP_ERROR;
    }

    if (pOperator->numOfDownstream > 1) {
      qError("join not supported for stream block scan");
      return TSDB_CODE_QRY_APP_ERROR;
    }
    return extractTbscanInStreamOpTree(pOperator->pDownstream[0], ppInfo);
  } else {
2979 2980 2981
    SStreamScanInfo* pInfo = pOperator->info;
    ASSERT(pInfo->pTableScanOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN);
    *ppInfo = pInfo->pTableScanOp->info;
L
Liu Jicong 已提交
2982 2983 2984 2985
    return 0;
  }
}

2986 2987 2988 2989 2990 2991 2992 2993 2994 2995 2996 2997 2998 2999 3000 3001 3002 3003 3004 3005 3006 3007
int32_t extractTableScanNode(SPhysiNode* pNode, STableScanPhysiNode** ppNode) {
  if (pNode->pChildren == NULL || LIST_LENGTH(pNode->pChildren) == 0) {
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == pNode->type) {
      *ppNode = (STableScanPhysiNode*)pNode;
      return 0;
    } else {
      ASSERT(0);
      terrno = TSDB_CODE_QRY_APP_ERROR;
      return -1;
    }
  } else {
    if (LIST_LENGTH(pNode->pChildren) != 1) {
      ASSERT(0);
      terrno = TSDB_CODE_QRY_APP_ERROR;
      return -1;
    }
    SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pNode->pChildren, 0);
    return extractTableScanNode(pChildNode, ppNode);
  }
  return -1;
}

3008
#if 0
L
Liu Jicong 已提交
3009 3010 3011 3012 3013
int32_t rebuildReader(SOperatorInfo* pOperator, SSubplan* plan, SReadHandle* pHandle, int64_t uid, int64_t ts) {
  STableScanInfo* pTableScanInfo = NULL;
  if (extractTbscanInStreamOpTree(pOperator, &pTableScanInfo) < 0) {
    return -1;
  }
3014

L
Liu Jicong 已提交
3015 3016 3017 3018
  STableScanPhysiNode* pNode = NULL;
  if (extractTableScanNode(plan->pNode, &pNode) < 0) {
    ASSERT(0);
  }
3019

H
Haojun Liao 已提交
3020
  tsdbReaderClose(pTableScanInfo->dataReader);
3021

L
Liu Jicong 已提交
3022
  STableListInfo info = {0};
H
Haojun Liao 已提交
3023
  pTableScanInfo->dataReader = doCreateDataReader(pNode, pHandle, &info, NULL);
L
Liu Jicong 已提交
3024 3025 3026 3027
  if (pTableScanInfo->dataReader == NULL) {
    ASSERT(0);
    qError("failed to create data reader");
    return TSDB_CODE_QRY_APP_ERROR;
3028
  }
L
Liu Jicong 已提交
3029
  // TODO: set uid and ts to data reader
3030 3031
  return 0;
}
3032
#endif
3033

C
Cary Xu 已提交
3034
int32_t encodeOperator(SOperatorInfo* ops, char** result, int32_t* length, int32_t* nOptrWithVal) {
wmmhello's avatar
wmmhello 已提交
3035
  int32_t code = TDB_CODE_SUCCESS;
3036
  char*   pCurrent = NULL;
wmmhello's avatar
wmmhello 已提交
3037
  int32_t currLength = 0;
3038
  if (ops->fpSet.encodeResultRow) {
C
Cary Xu 已提交
3039
    if (result == NULL || length == NULL || nOptrWithVal == NULL) {
wmmhello's avatar
wmmhello 已提交
3040 3041 3042
      return TSDB_CODE_TSC_INVALID_INPUT;
    }
    code = ops->fpSet.encodeResultRow(ops, &pCurrent, &currLength);
wmmhello's avatar
wmmhello 已提交
3043

3044 3045
    if (code != TDB_CODE_SUCCESS) {
      if (*result != NULL) {
wmmhello's avatar
wmmhello 已提交
3046 3047 3048 3049
        taosMemoryFree(*result);
        *result = NULL;
      }
      return code;
C
Cary Xu 已提交
3050 3051 3052
    } else if (currLength == 0) {
      ASSERT(!pCurrent);
      goto _downstream;
wmmhello's avatar
wmmhello 已提交
3053
    }
wmmhello's avatar
wmmhello 已提交
3054

C
Cary Xu 已提交
3055 3056
    ++(*nOptrWithVal);

C
Cary Xu 已提交
3057
    ASSERT(currLength >= 0);
wmmhello's avatar
wmmhello 已提交
3058

3059
    if (*result == NULL) {
wmmhello's avatar
wmmhello 已提交
3060
      *result = (char*)taosMemoryCalloc(1, currLength + sizeof(int32_t));
wmmhello's avatar
wmmhello 已提交
3061 3062 3063 3064 3065 3066
      if (*result == NULL) {
        taosMemoryFree(pCurrent);
        return TSDB_CODE_OUT_OF_MEMORY;
      }
      memcpy(*result + sizeof(int32_t), pCurrent, currLength);
      *(int32_t*)(*result) = currLength + sizeof(int32_t);
3067
    } else {
wmmhello's avatar
wmmhello 已提交
3068
      int32_t sizePre = *(int32_t*)(*result);
3069
      char*   tmp = (char*)taosMemoryRealloc(*result, sizePre + currLength);
wmmhello's avatar
wmmhello 已提交
3070 3071 3072 3073 3074 3075 3076 3077 3078 3079 3080 3081
      if (tmp == NULL) {
        taosMemoryFree(pCurrent);
        taosMemoryFree(*result);
        *result = NULL;
        return TSDB_CODE_OUT_OF_MEMORY;
      }
      *result = tmp;
      memcpy(*result + sizePre, pCurrent, currLength);
      *(int32_t*)(*result) += currLength;
    }
    taosMemoryFree(pCurrent);
    *length = *(int32_t*)(*result);
wmmhello's avatar
wmmhello 已提交
3082 3083
  }

3084
_downstream:
wmmhello's avatar
wmmhello 已提交
3085
  for (int32_t i = 0; i < ops->numOfDownstream; ++i) {
C
Cary Xu 已提交
3086
    code = encodeOperator(ops->pDownstream[i], result, length, nOptrWithVal);
3087
    if (code != TDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
3088
      return code;
wmmhello's avatar
wmmhello 已提交
3089 3090
    }
  }
wmmhello's avatar
wmmhello 已提交
3091
  return TDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
3092 3093
}

H
Haojun Liao 已提交
3094
int32_t decodeOperator(SOperatorInfo* ops, const char* result, int32_t length) {
wmmhello's avatar
wmmhello 已提交
3095
  int32_t code = TDB_CODE_SUCCESS;
3096 3097
  if (ops->fpSet.decodeResultRow) {
    if (result == NULL) {
wmmhello's avatar
wmmhello 已提交
3098 3099
      return TSDB_CODE_TSC_INVALID_INPUT;
    }
H
Haojun Liao 已提交
3100

3101
    ASSERT(length == *(int32_t*)result);
H
Haojun Liao 已提交
3102 3103

    const char* data = result + sizeof(int32_t);
L
Liu Jicong 已提交
3104
    code = ops->fpSet.decodeResultRow(ops, (char*)data);
3105
    if (code != TDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
3106 3107
      return code;
    }
wmmhello's avatar
wmmhello 已提交
3108

wmmhello's avatar
wmmhello 已提交
3109
    int32_t totalLength = *(int32_t*)result;
3110 3111
    int32_t dataLength = *(int32_t*)data;

3112
    if (totalLength == dataLength + sizeof(int32_t)) {  // the last data
wmmhello's avatar
wmmhello 已提交
3113 3114
      result = NULL;
      length = 0;
3115
    } else {
wmmhello's avatar
wmmhello 已提交
3116 3117 3118 3119
      result += dataLength;
      *(int32_t*)(result) = totalLength - dataLength;
      length = totalLength - dataLength;
    }
wmmhello's avatar
wmmhello 已提交
3120 3121
  }

wmmhello's avatar
wmmhello 已提交
3122 3123
  for (int32_t i = 0; i < ops->numOfDownstream; ++i) {
    code = decodeOperator(ops->pDownstream[i], result, length);
3124
    if (code != TDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
3125
      return code;
wmmhello's avatar
wmmhello 已提交
3126 3127
    }
  }
wmmhello's avatar
wmmhello 已提交
3128
  return TDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
3129 3130
}

D
dapan1121 已提交
3131
int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle) {
D
dapan1121 已提交
3132
  SExecTaskInfo* pTask = *(SExecTaskInfo**)pTaskInfo;
3133

D
dapan1121 已提交
3134
  switch (pNode->type) {
D
dapan1121 已提交
3135 3136 3137 3138 3139 3140
    case QUERY_NODE_PHYSICAL_PLAN_QUERY_INSERT: {
      SInserterParam* pInserterParam = taosMemoryCalloc(1, sizeof(SInserterParam));
      if (NULL == pInserterParam) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }
      pInserterParam->readHandle = readHandle;
L
Liu Jicong 已提交
3141

D
dapan1121 已提交
3142 3143 3144
      *pParam = pInserterParam;
      break;
    }
D
dapan1121 已提交
3145
    case QUERY_NODE_PHYSICAL_PLAN_DELETE: {
3146
      SDeleterParam* pDeleterParam = taosMemoryCalloc(1, sizeof(SDeleterParam));
D
dapan1121 已提交
3147 3148 3149
      if (NULL == pDeleterParam) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }
H
Haojun Liao 已提交
3150 3151 3152 3153
      int32_t tbNum = tableListGetSize(pTask->pTableInfoList);
      pDeleterParam->suid = tableListGetSuid(pTask->pTableInfoList);

      // TODO extract uid list
D
dapan1121 已提交
3154 3155 3156 3157 3158
      pDeleterParam->pUidList = taosArrayInit(tbNum, sizeof(uint64_t));
      if (NULL == pDeleterParam->pUidList) {
        taosMemoryFree(pDeleterParam);
        return TSDB_CODE_OUT_OF_MEMORY;
      }
H
Haojun Liao 已提交
3159

D
dapan1121 已提交
3160
      for (int32_t i = 0; i < tbNum; ++i) {
H
Haojun Liao 已提交
3161
        STableKeyInfo* pTable = tableListGetInfo(pTask->pTableInfoList, i);
D
dapan1121 已提交
3162 3163 3164 3165 3166 3167 3168 3169 3170 3171 3172 3173 3174
        taosArrayPush(pDeleterParam->pUidList, &pTable->uid);
      }

      *pParam = pDeleterParam;
      break;
    }
    default:
      break;
  }

  return TSDB_CODE_SUCCESS;
}

dengyihao's avatar
dengyihao 已提交
3175
int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId,
D
dapan1121 已提交
3176
                               char* sql, EOPTR_EXEC_MODEL model) {
H
Haojun Liao 已提交
3177 3178
  uint64_t queryId = pPlan->id.queryId;

D
dapan1121 已提交
3179
  *pTaskInfo = createExecTaskInfo(queryId, taskId, model, pPlan->dbFName);
H
Haojun Liao 已提交
3180 3181 3182
  if (*pTaskInfo == NULL) {
    goto _complete;
  }
H
Haojun Liao 已提交
3183

3184
  if (pHandle) {
L
Liu Jicong 已提交
3185
    /*(*pTaskInfo)->streamInfo.fillHistoryVer1 = pHandle->fillHistoryVer1;*/
3186 3187 3188
    if (pHandle->pStateBackend) {
      (*pTaskInfo)->streamInfo.pState = pHandle->pStateBackend;
    }
H
Haojun Liao 已提交
3189 3190
  }

3191
  (*pTaskInfo)->sql = sql;
D
dapan1121 已提交
3192
  sql = NULL;
H
Haojun Liao 已提交
3193

3194
  (*pTaskInfo)->pSubplan = pPlan;
3195 3196
  (*pTaskInfo)->pRoot =
      createOperatorTree(pPlan->pNode, *pTaskInfo, pHandle, pPlan->pTagCond, pPlan->pTagIndexCond, pPlan->user);
L
Liu Jicong 已提交
3197

D
dapan1121 已提交
3198
  if (NULL == (*pTaskInfo)->pRoot) {
H
Haojun Liao 已提交
3199
    terrno = (*pTaskInfo)->code;
D
dapan1121 已提交
3200
    goto _complete;
3201 3202
  }

H
Haojun Liao 已提交
3203
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
3204

H
Haojun Liao 已提交
3205
_complete:
D
dapan1121 已提交
3206
  taosMemoryFree(sql);
H
Haojun Liao 已提交
3207
  doDestroyTask(*pTaskInfo);
H
Haojun Liao 已提交
3208
  return terrno;
H
Haojun Liao 已提交
3209 3210
}

L
Liu Jicong 已提交
3211
void doDestroyTask(SExecTaskInfo* pTaskInfo) {
H
Haojun Liao 已提交
3212 3213
  qDebug("%s execTask is freed", GET_TASKID(pTaskInfo));

H
Haojun Liao 已提交
3214
  pTaskInfo->pTableInfoList = tableListDestroy(pTaskInfo->pTableInfoList);
H
Haojun Liao 已提交
3215
  destroyOperatorInfo(pTaskInfo->pRoot);
3216
  cleanupTableSchemaInfo(&pTaskInfo->schemaInfo);
3217
  cleanupStreamInfo(&pTaskInfo->streamInfo);
3218

D
dapan1121 已提交
3219
  if (!pTaskInfo->localFetch.localExec) {
D
dapan1121 已提交
3220 3221
    nodesDestroyNode((SNode*)pTaskInfo->pSubplan);
  }
3222

D
dapan1121 已提交
3223
  taosArrayDestroy(pTaskInfo->stopInfo.pStopInfo);
wafwerar's avatar
wafwerar 已提交
3224 3225 3226
  taosMemoryFreeClear(pTaskInfo->sql);
  taosMemoryFreeClear(pTaskInfo->id.str);
  taosMemoryFreeClear(pTaskInfo);
3227 3228 3229 3230
}

static int64_t getQuerySupportBufSize(size_t numOfTables) {
  size_t s1 = sizeof(STableQueryInfo);
L
Liu Jicong 已提交
3231 3232
  //  size_t s3 = sizeof(STableCheckInfo);  buffer consumption in tsdb
  return (int64_t)(s1 * 1.5 * numOfTables);
3233 3234 3235 3236 3237 3238 3239
}

int32_t checkForQueryBuf(size_t numOfTables) {
  int64_t t = getQuerySupportBufSize(numOfTables);
  if (tsQueryBufferSizeBytes < 0) {
    return TSDB_CODE_SUCCESS;
  } else if (tsQueryBufferSizeBytes > 0) {
L
Liu Jicong 已提交
3240
    while (1) {
3241 3242 3243 3244 3245 3246 3247 3248 3249 3250 3251 3252 3253 3254 3255 3256 3257 3258 3259 3260 3261 3262 3263 3264 3265 3266
      int64_t s = tsQueryBufferSizeBytes;
      int64_t remain = s - t;
      if (remain >= 0) {
        if (atomic_val_compare_exchange_64(&tsQueryBufferSizeBytes, s, remain) == s) {
          return TSDB_CODE_SUCCESS;
        }
      } else {
        return TSDB_CODE_QRY_NOT_ENOUGH_BUFFER;
      }
    }
  }

  // disable query processing if the value of tsQueryBufferSize is zero.
  return TSDB_CODE_QRY_NOT_ENOUGH_BUFFER;
}

void releaseQueryBuf(size_t numOfTables) {
  if (tsQueryBufferSizeBytes < 0) {
    return;
  }

  int64_t t = getQuerySupportBufSize(numOfTables);

  // restore value is not enough buffer available
  atomic_add_fetch_64(&tsQueryBufferSizeBytes, t);
}
D
dapan1121 已提交
3267

H
Haojun Liao 已提交
3268
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
3269
  SExplainExecInfo  execInfo = {0};
H
Haojun Liao 已提交
3270
  SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
3271

H
Haojun Liao 已提交
3272 3273 3274 3275 3276
  pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
  pExplainInfo->startupCost = operatorInfo->cost.openCost;
  pExplainInfo->totalCost = operatorInfo->cost.totalCost;
  pExplainInfo->verboseLen = 0;
  pExplainInfo->verboseInfo = NULL;
D
dapan1121 已提交
3277

3278
  if (operatorInfo->fpSet.getExplainFn) {
3279 3280
    int32_t code =
        operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
D
dapan1121 已提交
3281
    if (code) {
3282
      qError("%s operator getExplainFn failed, code:%s", GET_TASKID(operatorInfo->pTaskInfo), tstrerror(code));
D
dapan1121 已提交
3283 3284 3285
      return code;
    }
  }
dengyihao's avatar
dengyihao 已提交
3286

D
dapan1121 已提交
3287
  int32_t code = 0;
D
dapan1121 已提交
3288
  for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
H
Haojun Liao 已提交
3289 3290
    code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
    if (code != TSDB_CODE_SUCCESS) {
3291
      //      taosMemoryFreeClear(*pRes);
D
dapan1121 已提交
3292 3293 3294 3295 3296
      return TSDB_CODE_QRY_OUT_OF_MEMORY;
    }
  }

  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
3297
}
5
54liuyao 已提交
3298

3299 3300
int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
                     SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup) {
3301 3302 3303 3304 3305 3306
  SWinKey key = {
      .ts = win->skey,
      .groupId = tableGroupId,
  };
  char*   value = NULL;
  int32_t size = pAggSup->resultRowSize;
5
54liuyao 已提交
3307

3308
  if (streamStateAddIfNotExist(pState, &key, (void**)&value, &size) < 0) {
3309 3310 3311 3312 3313 3314 3315 3316 3317 3318
    return TSDB_CODE_QRY_OUT_OF_MEMORY;
  }
  *pResult = (SResultRow*)value;
  ASSERT(*pResult);
  // set time window for current result
  (*pResult)->win = (*win);
  setResultRowInitCtx(*pResult, pCtx, numOfOutput, rowEntryInfoOffset);
  return TSDB_CODE_SUCCESS;
}

3319 3320
int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult) {
  streamStateReleaseBuf(pState, pKey, pResult);
3321 3322 3323
  return TSDB_CODE_SUCCESS;
}

3324 3325
int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize) {
  streamStatePut(pState, pKey, pResult, resSize);
3326 3327 3328
  return TSDB_CODE_SUCCESS;
}

3329
int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
3330
                                   SGroupResInfo* pGroupResInfo) {
3331
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
3332 3333 3334 3335 3336 3337 3338 3339 3340 3341 3342 3343
  SExprInfo*      pExprInfo = pSup->pExprInfo;
  int32_t         numOfExprs = pSup->numOfExprs;
  int32_t*        rowEntryOffset = pSup->rowEntryInfoOffset;
  SqlFunctionCtx* pCtx = pSup->pCtx;

  int32_t numOfRows = getNumOfTotalRes(pGroupResInfo);

  for (int32_t i = pGroupResInfo->index; i < numOfRows; i += 1) {
    SResKeyPos* pPos = taosArrayGetP(pGroupResInfo->pRows, i);
    int32_t     size = 0;
    void*       pVal = NULL;
    SWinKey     key = {
3344 3345
            .ts = *(TSKEY*)pPos->key,
            .groupId = pPos->groupId,
3346
    };
3347
    int32_t code = streamStateGet(pState, &key, &pVal, &size);
3348 3349 3350 3351 3352 3353
    ASSERT(code == 0);
    SResultRow* pRow = (SResultRow*)pVal;
    doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset);
    // no results, continue to check the next one
    if (pRow->numOfRows == 0) {
      pGroupResInfo->index += 1;
3354
      releaseOutputBuf(pState, &key, pRow);
3355 3356 3357 3358 3359
      continue;
    }

    if (pBlock->info.groupId == 0) {
      pBlock->info.groupId = pPos->groupId;
3360 3361 3362 3363
      SStreamIntervalOperatorInfo* pInfo = pOperator->info;
      char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
      if (tbname != NULL) {
        memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
L
Liu Jicong 已提交
3364 3365
      } else {
        pBlock->info.parTbName[0] = 0;
3366
      }
3367 3368 3369
    } else {
      // current value belongs to different group, it can't be packed into one datablock
      if (pBlock->info.groupId != pPos->groupId) {
3370
        releaseOutputBuf(pState, &key, pRow);
3371 3372 3373 3374 3375 3376
        break;
      }
    }

    if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
      ASSERT(pBlock->info.rows > 0);
3377
      releaseOutputBuf(pState, &key, pRow);
3378 3379 3380 3381 3382 3383 3384 3385 3386 3387
      break;
    }

    pGroupResInfo->index += 1;

    for (int32_t j = 0; j < numOfExprs; ++j) {
      int32_t slotId = pExprInfo[j].base.resSchema.slotId;

      pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowEntryOffset);
      if (pCtx[j].fpSet.finalize) {
3388 3389 3390 3391
        int32_t code1 = pCtx[j].fpSet.finalize(&pCtx[j], pBlock);
        if (TAOS_FAILED(code1)) {
          qError("%s build result data block error, code %s", GET_TASKID(pTaskInfo), tstrerror(code1));
          T_LONG_JMP(pTaskInfo->env, code1);
3392 3393 3394 3395 3396 3397 3398 3399 3400 3401 3402 3403 3404
        }
      } else if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_select_value") == 0) {
        // do nothing, todo refactor
      } else {
        // expand the result into multiple rows. E.g., _wstart, top(k, 20)
        // the _wstart needs to copy to 20 following rows, since the results of top-k expands to 20 different rows.
        SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId);
        char*            in = GET_ROWCELL_INTERBUF(pCtx[j].resultInfo);
        for (int32_t k = 0; k < pRow->numOfRows; ++k) {
          colDataAppend(pColInfoData, pBlock->info.rows + k, in, pCtx[j].resultInfo->isNullRes);
        }
      }
    }
5
54liuyao 已提交
3405

3406
    pBlock->info.rows += pRow->numOfRows;
3407
    releaseOutputBuf(pState, &key, pRow);
3408 3409 3410 3411
  }
  blockDataUpdateTsWindow(pBlock, 0);
  return TSDB_CODE_SUCCESS;
}
5
54liuyao 已提交
3412 3413 3414 3415 3416 3417 3418

int32_t saveSessionDiscBuf(SStreamState* pState, SSessionKey* key, void* buf, int32_t size) {
  streamStateSessionPut(pState, key, (const void*)buf, size);
  releaseOutputBuf(pState, NULL, (SResultRow*)buf);
  return TSDB_CODE_SUCCESS;
}

3419
int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
5
54liuyao 已提交
3420
                                    SExprSupp* pSup, SGroupResInfo* pGroupResInfo) {
3421
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
3422 3423 3424 3425 3426 3427 3428 3429 3430 3431 3432 3433 3434
  SExprInfo*      pExprInfo = pSup->pExprInfo;
  int32_t         numOfExprs = pSup->numOfExprs;
  int32_t*        rowEntryOffset = pSup->rowEntryInfoOffset;
  SqlFunctionCtx* pCtx = pSup->pCtx;

  int32_t numOfRows = getNumOfTotalRes(pGroupResInfo);

  for (int32_t i = pGroupResInfo->index; i < numOfRows; i += 1) {
    SSessionKey* pKey = taosArrayGet(pGroupResInfo->pRows, i);
    int32_t      size = 0;
    void*        pVal = NULL;
    int32_t      code = streamStateSessionGet(pState, pKey, &pVal, &size);
    ASSERT(code == 0);
3435 3436
    if (code == -1) {
      // coverity scan
5
54liuyao 已提交
3437
      pGroupResInfo->index += 1;
3438 3439
      continue;
    }
5
54liuyao 已提交
3440 3441 3442 3443 3444 3445 3446 3447 3448 3449 3450
    SResultRow* pRow = (SResultRow*)pVal;
    doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset);
    // no results, continue to check the next one
    if (pRow->numOfRows == 0) {
      pGroupResInfo->index += 1;
      releaseOutputBuf(pState, NULL, pRow);
      continue;
    }

    if (pBlock->info.groupId == 0) {
      pBlock->info.groupId = pKey->groupId;
3451 3452 3453 3454 3455 3456 3457 3458 3459 3460 3461 3462 3463 3464 3465 3466 3467 3468 3469 3470 3471 3472 3473 3474 3475

      if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE) {
        SStreamStateAggOperatorInfo* pInfo = pOperator->info;

        char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
        if (tbname != NULL) {
          memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
        } else {
          pBlock->info.parTbName[0] = 0;
        }
      } else if (pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION ||
                 pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION ||
                 pOperator->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
        SStreamSessionAggOperatorInfo* pInfo = pOperator->info;

        char* tbname = taosHashGet(pInfo->pGroupIdTbNameMap, &pBlock->info.groupId, sizeof(int64_t));
        if (tbname != NULL) {
          memcpy(pBlock->info.parTbName, tbname, TSDB_TABLE_NAME_LEN);
        } else {
          pBlock->info.parTbName[0] = 0;
        }
      } else {
        ASSERT(0);
      }

5
54liuyao 已提交
3476 3477 3478 3479 3480 3481 3482 3483 3484 3485 3486 3487 3488 3489 3490 3491 3492 3493 3494 3495 3496 3497 3498 3499 3500 3501 3502 3503 3504 3505 3506 3507 3508 3509 3510 3511 3512 3513 3514 3515 3516 3517 3518 3519 3520
    } else {
      // current value belongs to different group, it can't be packed into one datablock
      if (pBlock->info.groupId != pKey->groupId) {
        releaseOutputBuf(pState, NULL, pRow);
        break;
      }
    }

    if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
      ASSERT(pBlock->info.rows > 0);
      releaseOutputBuf(pState, NULL, pRow);
      break;
    }

    pGroupResInfo->index += 1;

    for (int32_t j = 0; j < numOfExprs; ++j) {
      int32_t slotId = pExprInfo[j].base.resSchema.slotId;

      pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowEntryOffset);
      if (pCtx[j].fpSet.finalize) {
        int32_t code1 = pCtx[j].fpSet.finalize(&pCtx[j], pBlock);
        if (TAOS_FAILED(code1)) {
          qError("%s build result data block error, code %s", GET_TASKID(pTaskInfo), tstrerror(code1));
          T_LONG_JMP(pTaskInfo->env, code1);
        }
      } else if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_select_value") == 0) {
        // do nothing, todo refactor
      } else {
        // expand the result into multiple rows. E.g., _wstart, top(k, 20)
        // the _wstart needs to copy to 20 following rows, since the results of top-k expands to 20 different rows.
        SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, slotId);
        char*            in = GET_ROWCELL_INTERBUF(pCtx[j].resultInfo);
        for (int32_t k = 0; k < pRow->numOfRows; ++k) {
          colDataAppend(pColInfoData, pBlock->info.rows + k, in, pCtx[j].resultInfo->isNullRes);
        }
      }
    }

    pBlock->info.rows += pRow->numOfRows;
    // saveSessionDiscBuf(pState, pKey, pVal, size);
    releaseOutputBuf(pState, NULL, pRow);
  }
  blockDataUpdateTsWindow(pBlock, 0);
  return TSDB_CODE_SUCCESS;
3521
}