executorimpl.c 154.0 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 35
#include "tcompare.h"
#include "tcompression.h"
H
Haojun Liao 已提交
36
#include "thash.h"
37
#include "ttypes.h"
dengyihao's avatar
dengyihao 已提交
38
#include "vnode.h"
39

H
Haojun Liao 已提交
40
#define IS_MAIN_SCAN(runtime)          ((runtime)->scanFlag == MAIN_SCAN)
41 42 43 44 45 46
#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 已提交
47
  uint32_t v = taosRand();
48 49 50 51

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

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

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

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

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

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

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

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

88 89 90 91
static void destroyFillOperatorInfo(void* param);
static void destroyProjectOperatorInfo(void* param);
static void destroyOrderOperatorInfo(void* param);
static void destroyAggOperatorInfo(void* param);
X
Xiaoyu Wang 已提交
92

93 94
static void destroyIntervalOperatorInfo(void* param);
static void destroyExchangeOperatorInfo(void* param);
H
Haojun Liao 已提交
95

96 97
static void destroyOperatorInfo(SOperatorInfo* pOperator);

98
void doSetOperatorCompleted(SOperatorInfo* pOperator) {
99
  pOperator->status = OP_EXEC_DONE;
H
Haojun Liao 已提交
100
  ASSERT(pOperator->pTaskInfo != NULL);
101

102
  pOperator->cost.totalCost = (taosGetTimestampUs() - pOperator->pTaskInfo->cost.start * 1000) / 1000.0;
H
Haojun Liao 已提交
103
  setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
104
}
105

H
Haojun Liao 已提交
106
int32_t operatorDummyOpenFn(SOperatorInfo* pOperator) {
107
  OPTR_SET_OPENED(pOperator);
108
  pOperator->cost.openCost = 0;
H
Haojun Liao 已提交
109
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
110 111
}

112
SOperatorFpSet createOperatorFpSet(__optr_open_fn_t openFn, __optr_fn_t nextFn, __optr_fn_t streamFn,
113
                                   __optr_fn_t cleanup, __optr_close_fn_t closeFn, __optr_explain_fn_t explain) {
114 115 116 117 118 119 120 121 122 123 124 125
  SOperatorFpSet fpSet = {
      ._openFn = openFn,
      .getNextFn = nextFn,
      .getStreamResFn = streamFn,
      .cleanupFn = cleanup,
      .closeFn = closeFn,
      .getExplainFn = explain,
  };

  return fpSet;
}

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

129
static void initCtxOutputBuffer(SqlFunctionCtx* pCtx, int32_t size);
130
static void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId);
131

132
#if 0
L
Liu Jicong 已提交
133 134
static bool chkResultRowFromKey(STaskRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo, char* pData,
                                int16_t bytes, bool masterscan, uint64_t uid) {
135 136 137
  bool existed = false;
  SET_RES_WINDOW_KEY(pRuntimeEnv->keyBuf, pData, bytes, uid);

L
Liu Jicong 已提交
138 139
  SResultRow** p1 =
      (SResultRow**)taosHashGet(pRuntimeEnv->pResultRowHashTable, pRuntimeEnv->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
140 141 142 143 144 145 146 147 148 149 150

  // 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 已提交
151
        //        existed = (pResultRowInfo->pResult[0] == (*p1));
152 153
      } else {  // check if current pResultRowInfo contains the existed pResultRow
        SET_RES_EXT_WINDOW_KEY(pRuntimeEnv->keyBuf, pData, bytes, uid, pResultRowInfo);
L
Liu Jicong 已提交
154 155
        int64_t* index =
            taosHashGet(pRuntimeEnv->pResultRowListSet, pRuntimeEnv->keyBuf, GET_RES_EXT_WINDOW_KEY_LEN(bytes));
156 157 158 159 160 161 162 163 164 165 166 167 168
        if (index != NULL) {
          existed = true;
        } else {
          existed = false;
        }
      }
    }

    return existed;
  }

  return p1 != NULL;
}
169
#endif
170

171
SResultRow* getNewResultRow(SDiskbasedBuf* pResultBuf, int32_t* currentPageId, int32_t interBufSize) {
L
Liu Jicong 已提交
172
  SFilePage* pData = NULL;
173 174 175

  // in the first scan, new space needed for results
  int32_t pageId = -1;
176
  if (*currentPageId == -1) {
177
    pData = getNewBufPage(pResultBuf, &pageId);
178 179
    pData->num = sizeof(SFilePage);
  } else {
180 181
    pData = getBufPage(pResultBuf, *currentPageId);
    pageId = *currentPageId;
182

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

187
      pData = getNewBufPage(pResultBuf, &pageId);
188 189 190 191 192 193 194 195 196 197
      if (pData != NULL) {
        pData->num = sizeof(SFilePage);
      }
    }
  }

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

198 199
  setBufPageDirty(pData, true);

200 201 202 203
  // 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;
204
  *currentPageId = pageId;
205

wmmhello's avatar
wmmhello 已提交
206
  pData->num += interBufSize;
207 208 209
  return pResultRow;
}

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

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

225 226
  SResultRow* pResult = NULL;

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

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

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

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

261 262 263
  // 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 已提交
264
  // too many time window in query
265
  if (pTaskInfo->execModel == OPTR_EXEC_MODEL_BATCH &&
266
      tSimpleHashGetSize(pSup->pResultRowHashTable) > MAX_INTERVAL_TIME_WINDOW) {
267
    T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW);
H
Haojun Liao 已提交
268 269
  }

H
Haojun Liao 已提交
270
  return pResult;
H
Haojun Liao 已提交
271 272
}

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

L
Liu Jicong 已提交
279
  SFilePage* pData = NULL;
280 281 282

  // in the first scan, new space needed for results
  int32_t pageId = -1;
283
  SIDList list = getDataBufPagesIdList(pResultBuf);
284 285

  if (taosArrayGetSize(list) == 0) {
286
    pData = getNewBufPage(pResultBuf, &pageId);
287
    pData->num = sizeof(SFilePage);
288 289
  } else {
    SPageInfo* pi = getLastPageInfo(list);
290
    pData = getBufPage(pResultBuf, getPageId(pi));
291
    pageId = getPageId(pi);
292

293
    if (pData->num + size > getBufPageSize(pResultBuf)) {
294
      // release current page first, and prepare the next one
295
      releaseBufPageInfo(pResultBuf, pi);
296

297
      pData = getNewBufPage(pResultBuf, &pageId);
298
      if (pData != NULL) {
299
        pData->num = sizeof(SFilePage);
300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319
      }
    }
  }

  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;
}

320
//  query_range_start, query_range_end, window_duration, window_start, window_end
321
void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow) {
322 323 324
  pColData->info.type = TSDB_DATA_TYPE_TIMESTAMP;
  pColData->info.bytes = sizeof(int64_t);

325
  colInfoDataEnsureCapacity(pColData, 5);
326 327 328 329 330 331 332 333 334
  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 已提交
335
void cleanupExecTimeWindowInfo(SColumnInfoData* pColData) { colDataDestroy(pColData); }
H
Haojun Liao 已提交
336

337 338 339 340 341 342 343 344 345 346 347 348 349 350
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 已提交
351
  pCtx->input.numOfRows = pStatus->numOfRows;
352 353 354 355 356
  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) {
357
  for (int32_t k = 0; k < numOfOutput; ++k) {
H
Haojun Liao 已提交
358
    // keep it temporarily
359 360
    SFunctionCtxStatus status = {0};
    functionCtxSave(&pCtx[k], &status);
361

362
    pCtx[k].input.startRowIndex = offset;
363
    pCtx[k].input.numOfRows = forwardStep;
364 365 366

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

371 372
    if (fmIsWindowPseudoColumnFunc(pCtx[k].functionId)) {
      SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pCtx[k]);
373 374

      char* p = GET_ROWCELL_INTERBUF(pEntryInfo);
375

376
      SColumnInfoData idata = {0};
dengyihao's avatar
dengyihao 已提交
377
      idata.info.type = TSDB_DATA_TYPE_BIGINT;
378
      idata.info.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes;
dengyihao's avatar
dengyihao 已提交
379
      idata.pData = p;
380 381 382 383

      SScalarParam out = {.columnData = &idata};
      SScalarParam tw = {.numOfRows = 5, .columnData = pTimeWindowData};
      pCtx[k].sfp.process(&tw, 1, &out);
384
      pEntryInfo->numOfRes = 1;
385 386 387 388 389 390 391 392
    } 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;
393
          T_LONG_JMP(taskInfo->env, code);
394
        }
395
      }
396

397
      // restore it
398
      functionCtxRestore(&pCtx[k], &status);
399
    }
400 401 402
  }
}

403 404
static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
                                   bool createDummyCol);
405

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

416
void setInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag, bool createDummyCol) {
417
  if (pBlock->pBlockAgg != NULL) {
418
    doSetInputDataBlockInfo(pExprSup, pBlock, order);
419
  } else {
420
    doSetInputDataBlock(pExprSup, pBlock, order, scanFlag, createDummyCol);
H
Haojun Liao 已提交
421
  }
422 423
}

L
Liu Jicong 已提交
424 425
static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunctParam* pFuncParam, int32_t paramIndex,
                                             int32_t numOfRows) {
426 427 428 429 430 431 432 433
  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)
434 435
    pColInfo->info.type = pFuncParam->param.nType;
    pColInfo->info.bytes = pFuncParam->param.nLen;
436 437

    pInput->pData[paramIndex] = pColInfo;
438 439
  } else {
    pColInfo = pInput->pData[paramIndex];
440 441
  }

442
  colInfoDataEnsureCapacity(pColInfo, numOfRows);
443

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

  return TSDB_CODE_SUCCESS;
}

467 468
static int32_t doSetInputDataBlock(SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t order, int32_t scanFlag,
                                   bool createDummyCol) {
469
  int32_t         code = TSDB_CODE_SUCCESS;
470
  SqlFunctionCtx* pCtx = pExprSup->pCtx;
471

472
  for (int32_t i = 0; i < pExprSup->numOfExprs; ++i) {
L
Liu Jicong 已提交
473
    pCtx[i].order = order;
474 475
    pCtx[i].input.numOfRows = pBlock->info.rows;

L
Liu Jicong 已提交
476
    pCtx[i].pSrcBlock = pBlock;
X
Xiaoyu Wang 已提交
477
    pCtx[i].scanFlag = scanFlag;
H
Haojun Liao 已提交
478

479
    SInputColumnInfoData* pInput = &pCtx[i].input;
480
    pInput->uid = pBlock->info.uid;
C
Cary Xu 已提交
481
    pInput->colDataAggIsSet = false;
482

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

493
        // NOTE: the last parameter is the primary timestamp column
H
Haojun Liao 已提交
494
        // todo: refactor this
495
        if (fmIsImplicitTsFunc(pCtx[i].functionId) && (j == pOneExpr->base.numOfParams - 1)) {
L
Liu Jicong 已提交
496 497
          pInput->pPTS = pInput->pData[j];  // in case of merge function, this is not always the ts column data.
                                            //          ASSERT(pInput->pPTS->info.type == TSDB_DATA_TYPE_TIMESTAMP);
498
        }
499 500
        ASSERT(pInput->pData[j] != NULL);
      } else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) {
501 502 503
        // 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) {
504 505 506 507
          pInput->totalRows = pBlock->info.rows;
          pInput->numOfRows = pBlock->info.rows;
          pInput->startRowIndex = 0;

508
          code = doCreateConstantValColumnInfo(pInput, pFuncParam, j, pBlock->info.rows);
509 510 511
          if (code != TSDB_CODE_SUCCESS) {
            return code;
          }
512
        }
G
Ganlin Zhao 已提交
513 514
      }
    }
H
Haojun Liao 已提交
515
  }
516 517

  return code;
H
Haojun Liao 已提交
518 519
}

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

528 529 530 531
      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;
532
      }
533 534
    }
  }
535 536

  return TSDB_CODE_SUCCESS;
537 538
}

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

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

  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 已提交
569 570 571 572
  if (pResult != pSrcBlock) {
    pResult->info.groupId = pSrcBlock->info.groupId;
    memcpy(pResult->info.parTbName, pSrcBlock->info.parTbName, TSDB_TABLE_NAME_LEN);
  }
H
Haojun Liao 已提交
573

574 575
  // if the source equals to the destination, it is to create a new column as the result of scalar
  // function or some operators.
576 577
  bool createNewColModel = (pResult == pSrcBlock);

578 579
  int32_t numOfRows = 0;

580
  for (int32_t k = 0; k < numOfOutput; ++k) {
581 582
    int32_t               outputSlotId = pExpr[k].base.resSchema.slotId;
    SqlFunctionCtx*       pfCtx = &pCtx[k];
583
    SInputColumnInfoData* pInputData = &pfCtx->input;
584

L
Liu Jicong 已提交
585
    if (pExpr[k].pExpr->nodeType == QUERY_NODE_COLUMN) {  // it is a project query
586
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
587
      if (pResult->info.rows > 0 && !createNewColModel) {
588
        colDataMergeCol(pColInfoData, pResult->info.rows, (int32_t*)&pResult->info.capacity, pInputData->pData[0],
589
                        pInputData->numOfRows);
590
      } else {
591
        colDataAssign(pColInfoData, pInputData->pData[0], pInputData->numOfRows, &pResult->info);
592
      }
593

594
      numOfRows = pInputData->numOfRows;
595
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_VALUE) {
596
      SColumnInfoData* pColInfoData = taosArrayGet(pResult->pDataBlock, outputSlotId);
597

dengyihao's avatar
dengyihao 已提交
598
      int32_t offset = createNewColModel ? 0 : pResult->info.rows;
599 600 601 602 603 604 605 606

      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);
        }
607
      }
608 609

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

614
      SColumnInfoData* pResColData = taosArrayGet(pResult->pDataBlock, outputSlotId);
615
      SColumnInfoData  idata = {.info = pResColData->info, .hasNull = true};
616

617
      SScalarParam dest = {.columnData = &idata};
X
Xiaoyu Wang 已提交
618
      int32_t      code = scalarCalculate(pExpr[k].pExpr->_optrRoot.pRootNode, pBlockList, &dest);
619 620 621 622
      if (code != TSDB_CODE_SUCCESS) {
        taosArrayDestroy(pBlockList);
        return code;
      }
623

dengyihao's avatar
dengyihao 已提交
624
      int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
625
      ASSERT(pResult->info.capacity > 0);
626
      colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
D
dapan1121 已提交
627
      colDataDestroy(&idata);
L
Liu Jicong 已提交
628

629
      numOfRows = dest.numOfRows;
630 631
      taosArrayDestroy(pBlockList);
    } else if (pExpr[k].pExpr->nodeType == QUERY_NODE_FUNCTION) {
632 633
      // _rowts/_c0, not tbname column
      if (fmIsPseudoColumnFunc(pfCtx->functionId) && (!fmIsScanPseudoColumnFunc(pfCtx->functionId))) {
H
Haojun Liao 已提交
634
        // do nothing
635
      } else if (fmIsIndefiniteRowsFunc(pfCtx->functionId)) {
636 637
        SResultRowEntryInfo* pResInfo = GET_RES_INFO(pfCtx);
        pfCtx->fpSet.init(pfCtx, pResInfo);
638 639 640 641 642 643 644 645 646 647

        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;
        }

648 649 650 651 652
        // link pDstBlock to set selectivity value
        if (pfCtx->subsidiaries.num > 0) {
          pfCtx->pDstBlock = pResult;
        }

653
        numOfRows = pfCtx->fpSet.process(pfCtx);
H
Haojun Liao 已提交
654
      } else if (fmIsAggFunc(pfCtx->functionId)) {
G
Ganlin Zhao 已提交
655
        // selective value output should be set during corresponding function execution
656 657 658
        if (fmIsSelectValueFunc(pfCtx->functionId)) {
          continue;
        }
659 660
        // _group_key function for "partition by tbname" + csum(col_name) query
        SColumnInfoData* pOutput = taosArrayGet(pResult->pDataBlock, outputSlotId);
661
        int32_t          slotId = pfCtx->param[0].pCol->slotId;
662 663 664

        // todo handle the json tag
        SColumnInfoData* pInput = taosArrayGet(pSrcBlock->pDataBlock, slotId);
665
        for (int32_t f = 0; f < pSrcBlock->info.rows; ++f) {
666 667 668 669 670 671 672 673 674
          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 已提交
675 676 677
      } else {
        SArray* pBlockList = taosArrayInit(4, POINTER_BYTES);
        taosArrayPush(pBlockList, &pSrcBlock);
G
Ganlin Zhao 已提交
678

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

682
        SScalarParam dest = {.columnData = &idata};
X
Xiaoyu Wang 已提交
683
        int32_t      code = scalarCalculate((SNode*)pExpr[k].pExpr->_function.pFunctNode, pBlockList, &dest);
684 685 686 687
        if (code != TSDB_CODE_SUCCESS) {
          taosArrayDestroy(pBlockList);
          return code;
        }
688

dengyihao's avatar
dengyihao 已提交
689
        int32_t startOffset = createNewColModel ? 0 : pResult->info.rows;
690
        ASSERT(pResult->info.capacity > 0);
691
        colDataMergeCol(pResColData, startOffset, (int32_t*)&pResult->info.capacity, &idata, dest.numOfRows);
D
dapan1121 已提交
692
        colDataDestroy(&idata);
693 694

        numOfRows = dest.numOfRows;
H
Haojun Liao 已提交
695 696
        taosArrayDestroy(pBlockList);
      }
697
    } else {
698
      return TSDB_CODE_OPS_NOT_SUPPORT;
699 700
    }
  }
701

702 703 704
  if (!createNewColModel) {
    pResult->info.rows += numOfRows;
  }
705 706

  return TSDB_CODE_SUCCESS;
707 708
}

5
54liuyao 已提交
709
bool functionNeedToExecute(SqlFunctionCtx* pCtx) {
710
  struct SResultRowEntryInfo* pResInfo = GET_RES_INFO(pCtx);
711

712 713 714 715 716
  // in case of timestamp column, always generated results.
  int32_t functionId = pCtx->functionId;
  if (functionId == -1) {
    return false;
  }
717

718 719
  if (pCtx->scanFlag == REPEAT_SCAN) {
    return fmIsRepeatScanFunc(pCtx->functionId);
720 721
  }

722 723
  if (isRowEntryCompleted(pResInfo)) {
    return false;
724 725
  }

726 727 728
  return true;
}

729 730 731 732 733 734 735
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;
    }
736

737 738 739
    // Set the correct column info (data type and bytes)
    pInput->pData[paramIndex]->info.type = type;
    pInput->pData[paramIndex]->info.bytes = tDataTypes[type].bytes;
740
  }
H
Haojun Liao 已提交
741

742 743 744 745 746 747
  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;
748 749
    }
  } else {
750
    da = pInput->pColumnDataAgg[paramIndex];
751 752
  }

753
  ASSERT(!IS_VAR_DATA_TYPE(type));
754

755 756
  if (type == TSDB_DATA_TYPE_BIGINT) {
    int64_t v = pFuncParam->param.i;
757
    *da = (SColumnDataAgg){.numOfNull = 0, .min = v, .max = v, .sum = v * numOfRows};
758 759
  } else if (type == TSDB_DATA_TYPE_DOUBLE) {
    double v = pFuncParam->param.d;
760
    *da = (SColumnDataAgg){.numOfNull = 0};
761

762 763 764 765 766 767
    *(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;

768
    *da = (SColumnDataAgg){.numOfNull = 0};
769 770 771 772 773
    *(bool*)&da->min = 0;
    *(bool*)&da->max = v;
    *(bool*)&da->sum = v * numOfRows;
  } else if (type == TSDB_DATA_TYPE_TIMESTAMP) {
    // do nothing
774
  } else {
775
    ASSERT(0);
776 777
  }

778 779
  return TSDB_CODE_SUCCESS;
}
780

781
void setBlockSMAInfo(SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, SSDataBlock* pBlock) {
782 783 784 785 786 787 788 789 790
  int32_t numOfRows = pBlock->info.rows;

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

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

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

794 795
      if (pFuncParam->type == FUNC_PARAM_TYPE_COLUMN) {
        int32_t slotId = pFuncParam->pCol->slotId;
796 797 798 799
        pInput->pColumnDataAgg[j] = pBlock->pBlockAgg[slotId];
        if (pInput->pColumnDataAgg[j] == NULL) {
          pInput->colDataAggIsSet = false;
        }
800 801 802 803

        // 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);
804 805
      } else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) {
        doCreateConstantValColumnAggInfo(pInput, pFuncParam, pFuncParam->param.nType, j, pBlock->info.rows);
806 807
      }
    }
808
  } else {
809
    pInput->colDataAggIsSet = false;
810 811 812
  }
}

L
Liu Jicong 已提交
813
bool isTaskKilled(SExecTaskInfo* pTaskInfo) {
814 815
  // query has been executed more than tsShellActivityTimer, and the retrieve has not arrived
  // abort current query execution.
L
Liu Jicong 已提交
816 817
  if (pTaskInfo->owner != 0 &&
      ((taosGetTimestampSec() - pTaskInfo->cost.start / 1000) > 10 * getMaximumIdleDurationSec())
818 819
      /*(!needBuildResAfterQueryComplete(pTaskInfo))*/) {
    assert(pTaskInfo->cost.start != 0);
L
Liu Jicong 已提交
820 821 822
    //    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;
823 824 825 826 827
  }

  return false;
}

L
Liu Jicong 已提交
828
void setTaskKilled(SExecTaskInfo* pTaskInfo) { pTaskInfo->code = TSDB_CODE_TSC_QUERY_CANCELLED; }
829 830

/////////////////////////////////////////////////////////////////////////////////////////////
831
STimeWindow getAlignQueryTimeWindow(SInterval* pInterval, int32_t precision, int64_t key) {
L
Liu Jicong 已提交
832
  STimeWindow win = {0};
833
  win.skey = taosTimeTruncate(key, pInterval, precision);
834 835

  /*
H
Haojun Liao 已提交
836
   * if the realSkey > INT64_MAX - pInterval->interval, the query duration between
837 838
   * realSkey and realEkey must be less than one interval.Therefore, no need to adjust the query ranges.
   */
839 840 841
  win.ekey = taosTimeAdd(win.skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
  if (win.ekey < win.skey) {
    win.ekey = INT64_MAX;
842
  }
843 844

  return win;
845 846
}

847
#if 0
H
Haojun Liao 已提交
848
static bool overlapWithTimeWindow(STaskAttr* pQueryAttr, SDataBlockInfo* pBlockInfo) {
849 850
  STimeWindow w = {0};

dengyihao's avatar
dengyihao 已提交
851 852
  TSKEY sk = TMIN(pQueryAttr->window.skey, pQueryAttr->window.ekey);
  TSKEY ek = TMAX(pQueryAttr->window.skey, pQueryAttr->window.ekey);
853

854
  if (true) {
L
Liu Jicong 已提交
855
    //    getAlignQueryTimeWindow(pQueryAttr, pBlockInfo->window.skey, sk, ek, &w);
856 857 858 859 860 861
    assert(w.ekey >= pBlockInfo->window.skey);

    if (w.ekey < pBlockInfo->window.ekey) {
      return true;
    }

L
Liu Jicong 已提交
862 863
    while (1) {
      //      getNextTimeWindow(pQueryAttr, &w);
864 865 866 867 868 869 870 871 872 873
      if (w.skey > pBlockInfo->window.ekey) {
        break;
      }

      assert(w.ekey > pBlockInfo->window.ekey);
      if (w.skey <= pBlockInfo->window.ekey && w.skey > pBlockInfo->window.skey) {
        return true;
      }
    }
  } else {
L
Liu Jicong 已提交
874
    //    getAlignQueryTimeWindow(pQueryAttr, pBlockInfo->window.ekey, sk, ek, &w);
875 876 877 878 879 880
    assert(w.skey <= pBlockInfo->window.ekey);

    if (w.skey > pBlockInfo->window.skey) {
      return true;
    }

L
Liu Jicong 已提交
881 882
    while (1) {
      //      getNextTimeWindow(pQueryAttr, &w);
883 884 885 886 887 888 889 890 891 892 893 894 895
      if (w.ekey < pBlockInfo->window.skey) {
        break;
      }

      assert(w.skey < pBlockInfo->window.skey);
      if (w.ekey < pBlockInfo->window.ekey && w.ekey >= pBlockInfo->window.skey) {
        return true;
      }
    }
  }

  return false;
}
896
#endif
897

L
Liu Jicong 已提交
898 899
int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock,
                              uint32_t* status) {
900
  *status = BLK_DATA_NOT_LOAD;
901

H
Haojun Liao 已提交
902
  pBlock->pDataBlock = NULL;
L
Liu Jicong 已提交
903
  pBlock->pBlockAgg = NULL;
H
Haojun Liao 已提交
904

L
Liu Jicong 已提交
905 906
  //  int64_t groupId = pRuntimeEnv->current->groupIndex;
  //  bool    ascQuery = QUERY_IS_ASC_QUERY(pQueryAttr);
907

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

910 911
//  pCost->totalBlocks += 1;
//  pCost->totalRows += pBlock->info.rows;
H
Haojun Liao 已提交
912
#if 0
913 914 915
  // 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 已提交
916
      (QUERY_IS_INTERVAL_QUERY(pQueryAttr) && overlapWithTimeWindow(pTaskInfo, &pBlock->info))) {
917
    (*status) = BLK_DATA_DATA_LOAD;
918 919 920
  }

  // check if this data block is required to load
921
  if ((*status) != BLK_DATA_DATA_LOAD) {
922 923 924 925 926 927 928
    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 已提交
929
      bool  masterScan = IS_MAIN_SCAN(pRuntimeEnv);
930 931 932 933 934 935
      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,
936
                                    pTableScanInfo->rowEntryInfoOffset);
937 938 939
      } else {
        if (setResultOutputBufByKey(pRuntimeEnv, pTableScanInfo->pResultRowInfo, pBlock->info.uid, &win, masterScan, &pResult, groupId,
                                    pTableScanInfo->pCtx, pTableScanInfo->numOfOutput,
940
                                    pTableScanInfo->rowEntryInfoOffset) != TSDB_CODE_SUCCESS) {
941
          T_LONG_JMP(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
942 943 944 945
        }
      }
    } else if (pQueryAttr->stableQuery && (!pQueryAttr->tsCompQuery) && (!pQueryAttr->diffQuery)) { // stable aggregate, not interval aggregate or normal column aggregate
      doSetTableGroupOutputBuf(pRuntimeEnv, pTableScanInfo->pResultRowInfo, pTableScanInfo->pCtx,
946
                               pTableScanInfo->rowEntryInfoOffset, pTableScanInfo->numOfOutput,
947 948 949 950 951 952
                               pRuntimeEnv->current->groupIndex);
    }

    if (needFilter) {
      (*status) = doFilterByBlockTimeWindow(pTableScanInfo, pBlock);
    } else {
953
      (*status) = BLK_DATA_DATA_LOAD;
954 955 956 957
    }
  }

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

960
  if ((*status) == BLK_DATA_NOT_LOAD || (*status) == BLK_DATA_FILTEROUT) {
961 962
    //qDebug("QInfo:0x%"PRIx64" data block discard, brange:%" PRId64 "-%" PRId64 ", rows:%d", pQInfo->qId, pBlockInfo->window.skey,
//           pBlockInfo->window.ekey, pBlockInfo->rows);
963
    pCost->skipBlocks += 1;
964
  } else if ((*status) == BLK_DATA_SMA_LOAD) {
965 966
    // this function never returns error?
    pCost->loadBlockStatis += 1;
967
//    tsdbRetrieveDatablockSMA(pTableScanInfo->pTsdbReadHandle, &pBlock->pBlockAgg);
968 969

    if (pBlock->pBlockAgg == NULL) {  // data block statistics does not exist, load data block
970
//      pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pTsdbReadHandle, NULL);
971 972 973
      pCost->totalCheckedRows += pBlock->info.rows;
    }
  } else {
974
    assert((*status) == BLK_DATA_DATA_LOAD);
975 976 977

    // load the data block statistics to perform further filter
    pCost->loadBlockStatis += 1;
978
//    tsdbRetrieveDatablockSMA(pTableScanInfo->pTsdbReadHandle, &pBlock->pBlockAgg);
979 980 981 982 983 984

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

H
Haojun Liao 已提交
985
          bool  masterScan = IS_MAIN_SCAN(pRuntimeEnv);
986 987 988 989 990
          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,
991
                                      pTableScanInfo->rowEntryInfoOffset) != TSDB_CODE_SUCCESS) {
992
            T_LONG_JMP(pRuntimeEnv->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
993 994 995 996 997 998 999 1000 1001 1002
          }
        }
      }
      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
1003
            pCost->skipBlocks += 1;
1004 1005
            //qDebug("QInfo:0x%"PRIx64" data block discard, brange:%" PRId64 "-%" PRId64 ", rows:%d", pQInfo->qId,
//                   pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows);
1006
            (*status) = BLK_DATA_FILTEROUT;
1007 1008 1009 1010 1011 1012 1013
            return TSDB_CODE_SUCCESS;
          }
        }
      }
    }

    // current block has been discard due to filter applied
H
Haojun Liao 已提交
1014
//    if (!doFilterByBlockSMA(pRuntimeEnv, pBlock->pBlockAgg, pTableScanInfo->pCtx, pBlockInfo->rows)) {
1015
//      pCost->skipBlocks += 1;
1016 1017
//      qDebug("QInfo:0x%"PRIx64" data block discard, brange:%" PRId64 "-%" PRId64 ", rows:%d", pQInfo->qId, pBlockInfo->window.skey,
//             pBlockInfo->window.ekey, pBlockInfo->rows);
1018
//      (*status) = BLK_DATA_FILTEROUT;
1019 1020 1021 1022 1023
//      return TSDB_CODE_SUCCESS;
//    }

    pCost->totalCheckedRows += pBlockInfo->rows;
    pCost->loadBlocks += 1;
1024
//    pBlock->pDataBlock = tsdbRetrieveDataBlock(pTableScanInfo->pTsdbReadHandle, NULL);
1025 1026 1027 1028 1029
//    if (pBlock->pDataBlock == NULL) {
//      return terrno;
//    }

//    if (pQueryAttr->pFilters != NULL) {
1030
//      filterSetColFieldData(pQueryAttr->pFilters, taosArrayGetSize(pBlock->pDataBlock), pBlock->pDataBlock);
1031
//    }
1032

1033 1034 1035 1036
//    if (pQueryAttr->pFilters != NULL || pRuntimeEnv->pTsBuf != NULL) {
//      filterColRowsInDataBlock(pRuntimeEnv, pBlock, ascQuery);
//    }
  }
H
Haojun Liao 已提交
1037
#endif
1038 1039 1040
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
1041
static void updateTableQueryInfoForReverseScan(STableQueryInfo* pTableQueryInfo) {
1042 1043 1044 1045 1046
  if (pTableQueryInfo == NULL) {
    return;
  }
}

L
Liu Jicong 已提交
1047
void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status) {
1048
  if (status == TASK_NOT_COMPLETED) {
H
Haojun Liao 已提交
1049
    pTaskInfo->status = status;
1050 1051
  } else {
    // QUERY_NOT_COMPLETED is not compatible with any other status, so clear its position first
1052
    CLEAR_QUERY_STATUS(pTaskInfo, TASK_NOT_COMPLETED);
H
Haojun Liao 已提交
1053
    pTaskInfo->status |= status;
1054 1055 1056
  }
}

1057
void setResultRowInitCtx(SResultRow* pResult, SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset) {
5
54liuyao 已提交
1058
  bool init = false;
1059
  for (int32_t i = 0; i < numOfOutput; ++i) {
1060
    pCtx[i].resultInfo = getResultEntryInfo(pResult, i, rowEntryInfoOffset);
5
54liuyao 已提交
1061 1062 1063
    if (init) {
      continue;
    }
1064 1065 1066 1067 1068

    struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo;
    if (isRowEntryCompleted(pResInfo) && isRowEntryInitialized(pResInfo)) {
      continue;
    }
1069 1070 1071 1072 1073

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

1074 1075 1076 1077 1078 1079
    if (!pResInfo->initialized) {
      if (pCtx[i].functionId != -1) {
        pCtx[i].fpSet.init(&pCtx[i], pResInfo);
      } else {
        pResInfo->initialized = true;
      }
5
54liuyao 已提交
1080 1081
    } else {
      init = true;
1082 1083 1084 1085
    }
  }
}

1086 1087
static void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep,
                                                int32_t status);
1088

H
Haojun Liao 已提交
1089
void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock, SColMatchInfo* pColMatchInfo, SFilterInfo* pFilterInfo) {
1090
  if (pFilterNode == NULL || pBlock->info.rows == 0) {
S
shenglian zhou 已提交
1091 1092
    return;
  }
1093

H
Haojun Liao 已提交
1094
  SFilterInfo* filter = pFilterInfo;
H
Haojun Liao 已提交
1095
  int64_t      st = taosGetTimestampUs();
H
Haojun Liao 已提交
1096

H
Haojun Liao 已提交
1097
  //  pError("start filter");
H
Haojun Liao 已提交
1098

H
Haojun Liao 已提交
1099
  // todo move to the initialization function
H
Haojun Liao 已提交
1100
  int32_t code = 0;
H
Haojun Liao 已提交
1101
  bool    needFree = false;
H
Haojun Liao 已提交
1102
  if (filter == NULL) {
1103
    needFree = true;
H
Haojun Liao 已提交
1104 1105
    code = filterInitFromNode((SNode*)pFilterNode, &filter, 0);
  }
1106

1107
  SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};
1108 1109
  code = filterSetDataFromSlotId(filter, &param1);

1110
  SColumnInfoData* p = NULL;
1111
  int32_t          status = 0;
H
Haojun Liao 已提交
1112

1113
  // todo the keep seems never to be True??
1114
  bool keep = filterExecute(filter, pBlock, &p, NULL, param1.numOfCols, &status);
1115 1116 1117 1118

  if (needFree) {
    filterFreeInfo(filter);
  }
H
Haojun Liao 已提交
1119

1120
  extractQualifiedTupleByFilterResult(pBlock, p, keep, status);
H
Haojun Liao 已提交
1121

1122
  if (pColMatchInfo != NULL) {
H
Haojun Liao 已提交
1123 1124
    for (int32_t i = 0; i < taosArrayGetSize(pColMatchInfo->pList); ++i) {
      SColMatchItem* pInfo = taosArrayGet(pColMatchInfo->pList, i);
1125
      if (pInfo->colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
H
Haojun Liao 已提交
1126
        SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, pInfo->dstSlotId);
1127
        if (pColData->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
H
Haojun Liao 已提交
1128
          blockDataUpdateTsWindow(pBlock, pInfo->dstSlotId);
1129 1130 1131 1132 1133 1134
          break;
        }
      }
    }
  }

1135 1136
  colDataDestroy(p);
  taosMemoryFree(p);
1137 1138
}

1139
void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const SColumnInfoData* p, bool keep, int32_t status) {
1140 1141 1142 1143
  if (keep) {
    return;
  }

H
Haojun Liao 已提交
1144 1145 1146
  int32_t totalRows = pBlock->info.rows;

  if (status == FILTER_RESULT_ALL_QUALIFIED) {
1147
    // here nothing needs to be done
H
Haojun Liao 已提交
1148
  } else if (status == FILTER_RESULT_NONE_QUALIFIED) {
1149
    pBlock->info.rows = 0;
H
Haojun Liao 已提交
1150
  } else {
1151
    SSDataBlock* px = createOneDataBlock(pBlock, true);
1152

1153 1154
    size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
    for (int32_t i = 0; i < numOfCols; ++i) {
1155 1156
      SColumnInfoData* pSrc = taosArrayGet(px->pDataBlock, i);
      SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
1157
      // it is a reserved column for scalar function, and no data in this column yet.
1158
      if (pDst->pData == NULL || pSrc->pData == NULL) {
1159 1160 1161
        continue;
      }

1162 1163
      colInfoDataCleanup(pDst, pBlock->info.rows);

1164
      int32_t numOfRows = 0;
1165
      for (int32_t j = 0; j < totalRows; ++j) {
1166
        if (((int8_t*)p->pData)[j] == 0) {
D
dapan1121 已提交
1167 1168
          continue;
        }
1169

D
dapan1121 已提交
1170
        if (colDataIsNull_s(pSrc, j)) {
1171
          colDataAppendNULL(pDst, numOfRows);
D
dapan1121 已提交
1172
        } else {
1173
          colDataAppend(pDst, numOfRows, colDataGetData(pSrc, j), false);
D
dapan1121 已提交
1174
        }
1175
        numOfRows += 1;
H
Haojun Liao 已提交
1176
      }
1177

1178
      // todo this value can be assigned directly
1179 1180 1181 1182 1183
      if (pBlock->info.rows == totalRows) {
        pBlock->info.rows = numOfRows;
      } else {
        ASSERT(pBlock->info.rows == numOfRows);
      }
1184
    }
1185

dengyihao's avatar
dengyihao 已提交
1186
    blockDataDestroy(px);  // fix memory leak
1187 1188 1189
  }
}

1190
void doSetTableGroupOutputBuf(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId) {
1191
  // for simple group by query without interval, all the tables belong to one group result.
1192 1193 1194
  SExecTaskInfo*    pTaskInfo = pOperator->pTaskInfo;
  SAggOperatorInfo* pAggInfo = pOperator->info;

1195
  SResultRowInfo* pResultRowInfo = &pAggInfo->binfo.resultRowInfo;
1196 1197
  SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;
  int32_t*        rowEntryInfoOffset = pOperator->exprSupp.rowEntryInfoOffset;
1198

1199
  SResultRow* pResultRow = doSetResultOutBufByKey(pAggInfo->aggSup.pResultBuf, pResultRowInfo, (char*)&groupId,
L
Liu Jicong 已提交
1200
                                                  sizeof(groupId), true, groupId, pTaskInfo, false, &pAggInfo->aggSup);
L
Liu Jicong 已提交
1201
  assert(pResultRow != NULL);
1202 1203 1204 1205 1206 1207

  /*
   * 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 已提交
1208 1209
    int32_t ret =
        addNewWindowResultBuf(pResultRow, pAggInfo->aggSup.pResultBuf, groupId, pAggInfo->binfo.pRes->info.rowSize);
1210 1211 1212 1213 1214
    if (ret != TSDB_CODE_SUCCESS) {
      return;
    }
  }

1215
  setResultRowInitCtx(pResultRow, pCtx, numOfOutput, rowEntryInfoOffset);
1216 1217
}

1218 1219 1220
static void setExecutionContext(SOperatorInfo* pOperator, int32_t numOfOutput, uint64_t groupId) {
  SAggOperatorInfo* pAggInfo = pOperator->info;
  if (pAggInfo->groupId != UINT64_MAX && pAggInfo->groupId == groupId) {
1221 1222
    return;
  }
1223 1224

  doSetTableGroupOutputBuf(pOperator, numOfOutput, groupId);
1225 1226

  // record the current active group id
H
Haojun Liao 已提交
1227
  pAggInfo->groupId = groupId;
1228 1229
}

dengyihao's avatar
dengyihao 已提交
1230 1231
static void doUpdateNumOfRows(SqlFunctionCtx* pCtx, SResultRow* pRow, int32_t numOfExprs,
                              const int32_t* rowCellOffset) {
1232
  bool returnNotNull = false;
1233
  for (int32_t j = 0; j < numOfExprs; ++j) {
1234
    struct SResultRowEntryInfo* pResInfo = getResultEntryInfo(pRow, j, rowCellOffset);
1235 1236 1237 1238 1239 1240 1241
    if (!isRowEntryInitialized(pResInfo)) {
      continue;
    }

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

1243
    if (fmIsNotNullOutputFunc(pCtx[j].functionId)) {
1244 1245
      returnNotNull = true;
    }
1246
  }
S
shenglian zhou 已提交
1247 1248
  // 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
1249
  if (pRow->numOfRows == 0 && !returnNotNull) {
1250
    pRow->numOfRows = 1;
1251 1252 1253
  }
}

1254 1255
static void doCopyResultToDataBlock(SExprInfo* pExprInfo, int32_t numOfExprs, SResultRow* pRow, SqlFunctionCtx* pCtx,
                                    SSDataBlock* pBlock, const int32_t* rowEntryOffset, SExecTaskInfo* pTaskInfo) {
1256 1257 1258
  for (int32_t j = 0; j < numOfExprs; ++j) {
    int32_t slotId = pExprInfo[j].base.resSchema.slotId;

1259
    pCtx[j].resultInfo = getResultEntryInfo(pRow, j, rowEntryOffset);
1260
    if (pCtx[j].fpSet.finalize) {
1261
      if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_group_key") == 0) {
1262 1263
        // for groupkey along with functions that output multiple lines(e.g. Histogram)
        // need to match groupkey result for each output row of that function.
1264 1265 1266 1267 1268
        if (pCtx[j].resultInfo->numOfRes != 0) {
          pCtx[j].resultInfo->numOfRes = pRow->numOfRows;
        }
      }

1269 1270 1271
      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));
1272
        T_LONG_JMP(pTaskInfo->env, code);
1273 1274
      }
    } else if (strcmp(pCtx[j].pExpr->pExpr->_function.functionName, "_select_value") == 0) {
1275
      // do nothing
1276
    } else {
1277 1278
      // 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.
1279 1280 1281 1282 1283 1284 1285
      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);
      }
    }
  }
1286 1287
}

1288 1289 1290
// todo refactor. SResultRow has direct pointer in miainfo
int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup,
                           SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo) {
1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316
  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);
1317 1318

  releaseBufPage(pBuf, page);
1319
  pBlock->info.rows += pRow->numOfRows;
1320 1321 1322
  return 0;
}

1323 1324 1325 1326 1327 1328 1329
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;

1330
  int32_t numOfRows = getNumOfTotalRes(pGroupResInfo);
1331

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

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

H
Haojun Liao 已提交
1338
    doUpdateNumOfRows(pCtx, pRow, numOfExprs, rowEntryOffset);
1339 1340

    // no results, continue to check the next one
1341 1342
    if (pRow->numOfRows == 0) {
      pGroupResInfo->index += 1;
1343
      releaseBufPage(pBuf, page);
1344 1345 1346
      continue;
    }

1347 1348 1349 1350 1351
    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) {
1352
        releaseBufPage(pBuf, page);
1353 1354 1355 1356
        break;
      }
    }

1357
    if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
1358
      ASSERT(pBlock->info.rows > 0);
1359
      releaseBufPage(pBuf, page);
1360 1361 1362 1363
      break;
    }

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

1366
    releaseBufPage(pBuf, page);
1367
    pBlock->info.rows += pRow->numOfRows;
1368 1369
  }

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

1373
  blockDataUpdateTsWindow(pBlock, 0);
1374 1375 1376
  return 0;
}

1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416
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 已提交
1417 1418
void doBuildResultDatablock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo,
                            SDiskbasedBuf* pBuf) {
1419
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1420
  SSDataBlock*   pBlock = pbInfo->pRes;
1421

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

1425
  blockDataCleanup(pBlock);
1426
  if (!hasRemainResults(pGroupResInfo)) {
1427 1428 1429
    return;
  }

1430 1431
  // clear the existed group id
  pBlock->info.groupId = 0;
1432 1433 1434
  if (!pbInfo->mergeResultBlock) {
    doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo);
  } else {
dengyihao's avatar
dengyihao 已提交
1435
    while (hasRemainResults(pGroupResInfo)) {
1436 1437 1438
      doCopyToSDataBlock(pTaskInfo, pBlock, &pOperator->exprSupp, pBuf, pGroupResInfo);
      if (pBlock->info.rows >= pOperator->resultInfo.threshold) {
        break;
1439 1440
      }

1441 1442
      // clearing group id to continue to merge data that belong to different groups
      pBlock->info.groupId = 0;
1443
    }
1444 1445 1446

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

L
Liu Jicong 已提交
1450
static int32_t compressQueryColData(SColumnInfoData* pColRes, int32_t numOfRows, char* data, int8_t compressed) {
1451 1452
  int32_t colSize = pColRes->info.bytes * numOfRows;
  return (*(tDataTypes[pColRes->info.type].compFunc))(pColRes->pData, colSize, numOfRows, data,
L
Liu Jicong 已提交
1453
                                                      colSize + COMP_OVERFLOW_BYTES, compressed, NULL, 0);
1454 1455
}

L
Liu Jicong 已提交
1456 1457
void queryCostStatis(SExecTaskInfo* pTaskInfo) {
  STaskCostInfo* pSummary = &pTaskInfo->cost;
1458

L
Liu Jicong 已提交
1459 1460 1461
  //  uint64_t hashSize = taosHashGetMemSize(pQInfo->runtimeEnv.pResultRowHashTable);
  //  hashSize += taosHashGetMemSize(pRuntimeEnv->tableqinfoGroupInfo.map);
  //  pSummary->hashSize = hashSize;
1462

L
Liu Jicong 已提交
1463 1464 1465 1466 1467 1468 1469 1470 1471
  //  SResultRowPool* p = pTaskInfo->pool;
  //  if (p != NULL) {
  //    pSummary->winInfoSize = getResultRowPoolMemSize(p);
  //    pSummary->numOfTimeWindows = getNumOfAllocatedResultRows(p);
  //  } else {
  //    pSummary->winInfoSize = 0;
  //    pSummary->numOfTimeWindows = 0;
  //  }

1472 1473
  SFileBlockLoadRecorder* pRecorder = pSummary->pRecoder;
  if (pSummary->pRecoder != NULL) {
1474
    qDebug(
X
Xiaoyu Wang 已提交
1475 1476 1477 1478
        "%s :cost summary: elapsed time:%.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, pRecorder->totalBlocks, pRecorder->loadBlockStatis,
        pRecorder->loadBlocks, pRecorder->totalRows, pRecorder->totalCheckedRows);
1479
  }
1480

L
Liu Jicong 已提交
1481 1482 1483
  // qDebug("QInfo:0x%"PRIx64" :cost summary: winResPool size:%.2f Kb, numOfWin:%"PRId64", tableInfoSize:%.2f Kb,
  // hashTable:%.2f Kb", pQInfo->qId, pSummary->winInfoSize/1024.0,
  //      pSummary->numOfTimeWindows, pSummary->tableInfoSize/1024.0, pSummary->hashSize/1024.0);
1484 1485
}

L
Liu Jicong 已提交
1486 1487 1488
// static void updateOffsetVal(STaskRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBlockInfo) {
//   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
//   STableQueryInfo* pTableQueryInfo = pRuntimeEnv->current;
1489
//
L
Liu Jicong 已提交
1490
//   int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQueryAttr->order.order);
1491
//
L
Liu Jicong 已提交
1492 1493 1494 1495
//   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;
//   }
1496
//
L
Liu Jicong 已提交
1497 1498 1499 1500 1501
//   if (QUERY_IS_ASC_QUERY(pQueryAttr)) {
//     pQueryAttr->pos = (int32_t)pQueryAttr->limit.offset;
//   } else {
//     pQueryAttr->pos = pBlockInfo->rows - (int32_t)pQueryAttr->limit.offset - 1;
//   }
1502
//
L
Liu Jicong 已提交
1503
//   assert(pQueryAttr->pos >= 0 && pQueryAttr->pos <= pBlockInfo->rows - 1);
1504
//
L
Liu Jicong 已提交
1505 1506
//   SArray *         pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
//   SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
1507
//
L
Liu Jicong 已提交
1508 1509
//   // update the pQueryAttr->limit.offset value, and pQueryAttr->pos value
//   TSKEY *keys = (TSKEY *) pColInfoData->pData;
1510
//
L
Liu Jicong 已提交
1511 1512 1513
//   // update the offset value
//   pTableQueryInfo->lastKey = keys[pQueryAttr->pos];
//   pQueryAttr->limit.offset = 0;
1514
//
L
Liu Jicong 已提交
1515
//   int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
1516
//
L
Liu Jicong 已提交
1517 1518 1519 1520
//   //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);
// }
1521

L
Liu Jicong 已提交
1522 1523
// void skipBlocks(STaskRuntimeEnv *pRuntimeEnv) {
//   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
1524
//
L
Liu Jicong 已提交
1525 1526 1527
//   if (pQueryAttr->limit.offset <= 0 || pQueryAttr->numOfFilterCols > 0) {
//     return;
//   }
1528
//
L
Liu Jicong 已提交
1529 1530
//   pQueryAttr->pos = 0;
//   int32_t step = GET_FORWARD_DIRECTION_FACTOR(pQueryAttr->order.order);
1531
//
L
Liu Jicong 已提交
1532 1533
//   STableQueryInfo* pTableQueryInfo = pRuntimeEnv->current;
//   TsdbQueryHandleT pTsdbReadHandle = pRuntimeEnv->pTsdbReadHandle;
1534
//
L
Liu Jicong 已提交
1535 1536 1537
//   SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER;
//   while (tsdbNextDataBlock(pTsdbReadHandle)) {
//     if (isTaskKilled(pRuntimeEnv->qinfo)) {
1538
//       T_LONG_JMP(pRuntimeEnv->env, TSDB_CODE_TSC_QUERY_CANCELLED);
L
Liu Jicong 已提交
1539
//     }
1540
//
L
Liu Jicong 已提交
1541
//     tsdbRetrieveDataBlockInfo(pTsdbReadHandle, &blockInfo);
1542
//
L
Liu Jicong 已提交
1543 1544 1545 1546
//     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;
1547
//
L
Liu Jicong 已提交
1548 1549 1550 1551 1552 1553 1554
//       //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;
//     }
//   }
1555
//
L
Liu Jicong 已提交
1556
//   if (terrno != TSDB_CODE_SUCCESS) {
1557
//     T_LONG_JMP(pRuntimeEnv->env, terrno);
L
Liu Jicong 已提交
1558 1559 1560 1561 1562 1563 1564
//   }
// }

// static TSKEY doSkipIntervalProcess(STaskRuntimeEnv* pRuntimeEnv, STimeWindow* win, SDataBlockInfo* pBlockInfo,
// STableQueryInfo* pTableQueryInfo) {
//   STaskAttr *pQueryAttr = pRuntimeEnv->pQueryAttr;
//   SResultRowInfo *pWindowResInfo = &pRuntimeEnv->resultRowInfo;
1565
//
L
Liu Jicong 已提交
1566 1567 1568
//   assert(pQueryAttr->limit.offset == 0);
//   STimeWindow tw = *win;
//   getNextTimeWindow(pQueryAttr, &tw);
1569
//
L
Liu Jicong 已提交
1570 1571
//   if ((tw.skey <= pBlockInfo->window.ekey && QUERY_IS_ASC_QUERY(pQueryAttr)) ||
//       (tw.ekey >= pBlockInfo->window.skey && !QUERY_IS_ASC_QUERY(pQueryAttr))) {
1572
//
L
Liu Jicong 已提交
1573 1574 1575 1576
//     // 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);
1577
//
L
Liu Jicong 已提交
1578 1579 1580 1581
//     tw = *win;
//     int32_t startPos =
//         getNextQualifiedWindow(pQueryAttr, &tw, pBlockInfo, pColInfoData->pData, binarySearchForKey, -1);
//     assert(startPos >= 0);
1582
//
L
Liu Jicong 已提交
1583 1584
//     // set the abort info
//     pQueryAttr->pos = startPos;
1585
//
L
Liu Jicong 已提交
1586 1587 1588 1589
//     // reset the query start timestamp
//     pTableQueryInfo->win.skey = ((TSKEY *)pColInfoData->pData)[startPos];
//     pQueryAttr->window.skey = pTableQueryInfo->win.skey;
//     TSKEY key = pTableQueryInfo->win.skey;
1590
//
L
Liu Jicong 已提交
1591 1592
//     pWindowResInfo->prevSKey = tw.skey;
//     int32_t index = pRuntimeEnv->resultRowInfo.curIndex;
1593
//
L
Liu Jicong 已提交
1594 1595
//     int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
//     pRuntimeEnv->resultRowInfo.curIndex = index;  // restore the window index
1596
//
L
Liu Jicong 已提交
1597 1598 1599 1600
//     //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);
1601
//
L
Liu Jicong 已提交
1602 1603 1604 1605 1606
//     return key;
//   } else {  // do nothing
//     pQueryAttr->window.skey      = tw.skey;
//     pWindowResInfo->prevSKey = tw.skey;
//     pTableQueryInfo->lastKey = tw.skey;
1607
//
L
Liu Jicong 已提交
1608 1609
//     return tw.skey;
//   }
1610
//
L
Liu Jicong 已提交
1611 1612 1613 1614 1615 1616 1617 1618 1619 1620
//   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);
//   }
1621
//
L
Liu Jicong 已提交
1622 1623 1624 1625 1626
//   // 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;
//   }
1627
//
L
Liu Jicong 已提交
1628 1629 1630 1631 1632 1633 1634
//   /*
//    * 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);
1635
//
L
Liu Jicong 已提交
1636 1637
//   STimeWindow w = TSWINDOW_INITIALIZER;
//   bool ascQuery = QUERY_IS_ASC_QUERY(pQueryAttr);
1638
//
L
Liu Jicong 已提交
1639 1640
//   SResultRowInfo *pWindowResInfo = &pRuntimeEnv->resultRowInfo;
//   STableQueryInfo *pTableQueryInfo = pRuntimeEnv->current;
1641
//
L
Liu Jicong 已提交
1642 1643 1644
//   SDataBlockInfo blockInfo = SDATA_BLOCK_INITIALIZER;
//   while (tsdbNextDataBlock(pRuntimeEnv->pTsdbReadHandle)) {
//     tsdbRetrieveDataBlockInfo(pRuntimeEnv->pTsdbReadHandle, &blockInfo);
1645
//
L
Liu Jicong 已提交
1646 1647 1648 1649 1650 1651 1652 1653 1654
//     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;
//     }
1655
//
L
Liu Jicong 已提交
1656 1657
//     // the first time window
//     STimeWindow win = getActiveTimeWindow(pWindowResInfo, pWindowResInfo->prevSKey, pQueryAttr);
1658
//
L
Liu Jicong 已提交
1659 1660
//     while (pQueryAttr->limit.offset > 0) {
//       STimeWindow tw = win;
1661
//
L
Liu Jicong 已提交
1662 1663 1664
//       if ((win.ekey <= blockInfo.window.ekey && ascQuery) || (win.ekey >= blockInfo.window.skey && !ascQuery)) {
//         pQueryAttr->limit.offset -= 1;
//         pWindowResInfo->prevSKey = win.skey;
1665
//
L
Liu Jicong 已提交
1666 1667 1668 1669 1670 1671
//         // 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;
//         }
//       }
1672
//
L
Liu Jicong 已提交
1673 1674 1675 1676
//       if (pQueryAttr->limit.offset == 0) {
//         *start = doSkipIntervalProcess(pRuntimeEnv, &win, &blockInfo, pTableQueryInfo);
//         return true;
//       }
1677
//
L
Liu Jicong 已提交
1678 1679
//       // current window does not ended in current data block, try next data block
//       getNextTimeWindow(pQueryAttr, &tw);
1680
//
L
Liu Jicong 已提交
1681 1682 1683 1684 1685 1686 1687 1688 1689
//       /*
//        * 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)) {
1690
//
L
Liu Jicong 已提交
1691 1692
//         SArray *pDataBlock = tsdbRetrieveDataBlock(pRuntimeEnv->pTsdbReadHandle, NULL);
//         SColumnInfoData *pColInfoData = taosArrayGet(pDataBlock, 0);
1693
//
L
Liu Jicong 已提交
1694 1695 1696
//         if ((win.ekey > blockInfo.window.ekey && ascQuery) || (win.ekey < blockInfo.window.skey && !ascQuery)) {
//           pQueryAttr->limit.offset -= 1;
//         }
1697
//
L
Liu Jicong 已提交
1698 1699 1700 1701 1702 1703 1704 1705
//         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);
1706
//
L
Liu Jicong 已提交
1707 1708 1709 1710 1711 1712 1713 1714 1715 1716 1717
//           // 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.
//       }
//     }
//   }
1718
//
L
Liu Jicong 已提交
1719 1720
//   // check for error
//   if (terrno != TSDB_CODE_SUCCESS) {
1721
//     T_LONG_JMP(pRuntimeEnv->env, terrno);
L
Liu Jicong 已提交
1722
//   }
1723
//
L
Liu Jicong 已提交
1724 1725
//   return true;
// }
1726

1727
int32_t appendDownstream(SOperatorInfo* p, SOperatorInfo** pDownstream, int32_t num) {
H
Haojun Liao 已提交
1728
  if (p->pDownstream == NULL) {
H
Haojun Liao 已提交
1729
    assert(p->numOfDownstream == 0);
1730 1731
  }

wafwerar's avatar
wafwerar 已提交
1732
  p->pDownstream = taosMemoryCalloc(1, num * POINTER_BYTES);
1733 1734 1735 1736 1737 1738 1739
  if (p->pDownstream == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

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

wmmhello's avatar
wmmhello 已提交
1742
static void doDestroyTableList(STableListInfo* pTableqinfoList);
1743

1744 1745 1746 1747
typedef struct SFetchRspHandleWrapper {
  uint32_t exchangeId;
  int32_t  sourceIndex;
} SFetchRspHandleWrapper;
1748

D
dapan1121 已提交
1749
int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
X
Xiaoyu Wang 已提交
1750
  SFetchRspHandleWrapper* pWrapper = (SFetchRspHandleWrapper*)param;
1751 1752 1753 1754

  SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pWrapper->exchangeId);
  if (pExchangeInfo == NULL) {
    qWarn("failed to acquire exchange operator, since it may have been released");
1755
    taosMemoryFree(pMsg->pData);
1756 1757 1758
    return TSDB_CODE_SUCCESS;
  }

X
Xiaoyu Wang 已提交
1759
  int32_t          index = pWrapper->sourceIndex;
1760
  SSourceDataInfo* pSourceDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, index);
1761

H
Haojun Liao 已提交
1762 1763
  if (code == TSDB_CODE_SUCCESS) {
    pSourceDataInfo->pRsp = pMsg->pData;
1764

H
Haojun Liao 已提交
1765 1766
    SRetrieveTableRsp* pRsp = pSourceDataInfo->pRsp;
    pRsp->numOfRows = htonl(pRsp->numOfRows);
dengyihao's avatar
dengyihao 已提交
1767
    pRsp->compLen = htonl(pRsp->compLen);
1768
    pRsp->numOfCols = htonl(pRsp->numOfCols);
dengyihao's avatar
dengyihao 已提交
1769
    pRsp->useconds = htobe64(pRsp->useconds);
1770
    pRsp->numOfBlocks = htonl(pRsp->numOfBlocks);
1771

1772
    ASSERT(pRsp != NULL);
H
Haojun Liao 已提交
1773 1774
    qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d", pSourceDataInfo->taskId, index, pRsp->numOfBlocks,
           pRsp->numOfRows);
H
Haojun Liao 已提交
1775
  } else {
1776
    taosMemoryFree(pMsg->pData);
H
Haojun Liao 已提交
1777
    pSourceDataInfo->code = code;
1778
    qDebug("%s fetch rsp received, index:%d, error:%s", pSourceDataInfo->taskId, index, tstrerror(code));
H
Haojun Liao 已提交
1779
  }
H
Haojun Liao 已提交
1780

H
Haojun Liao 已提交
1781
  pSourceDataInfo->status = EX_SOURCE_DATA_READY;
1782 1783 1784 1785

  tsem_post(&pExchangeInfo->ready);
  taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);

wmmhello's avatar
wmmhello 已提交
1786
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
1787 1788
}

D
dapan1121 已提交
1789
void qProcessRspMsg(void* parent, SRpcMsg* pMsg, SEpSet* pEpSet) {
S
Shengliang Guan 已提交
1790 1791
  SMsgSendInfo* pSendInfo = (SMsgSendInfo*)pMsg->info.ahandle;
  assert(pMsg->info.ahandle != NULL);
H
Haojun Liao 已提交
1792 1793 1794 1795

  SDataBuf buf = {.len = pMsg->contLen, .pData = NULL};

  if (pMsg->contLen > 0) {
wafwerar's avatar
wafwerar 已提交
1796
    buf.pData = taosMemoryCalloc(1, pMsg->contLen);
H
Haojun Liao 已提交
1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807
    if (buf.pData == NULL) {
      terrno = TSDB_CODE_OUT_OF_MEMORY;
      pMsg->code = TSDB_CODE_OUT_OF_MEMORY;
    } else {
      memcpy(buf.pData, pMsg->pCont, pMsg->contLen);
    }
  }

  pSendInfo->fp(pSendInfo->param, &buf, pMsg->code);
  rpcFreeCont(pMsg->pCont);
  destroySendMsgInfo(pSendInfo);
1808 1809
}

L
Liu Jicong 已提交
1810
static int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex) {
1811
  size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
1812

L
Liu Jicong 已提交
1813 1814
  SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, sourceIndex);
  SSourceDataInfo*       pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, sourceIndex);
1815

1816 1817
  ASSERT(pDataInfo->status == EX_SOURCE_DATA_NOT_READY);

1818
  SFetchRspHandleWrapper* pWrapper = taosMemoryCalloc(1, sizeof(SFetchRspHandleWrapper));
X
Xiaoyu Wang 已提交
1819
  pWrapper->exchangeId = pExchangeInfo->self;
1820 1821
  pWrapper->sourceIndex = sourceIndex;

D
dapan1121 已提交
1822 1823
  if (pSource->localExec) {
    SDataBuf pBuf = {0};
1824 1825 1826
    int32_t  code =
        (*pTaskInfo->localFetch.fp)(pTaskInfo->localFetch.handle, pSource->schedId, pTaskInfo->id.queryId,
                                    pSource->taskId, 0, pSource->execId, &pBuf.pData, pTaskInfo->localFetch.explainRes);
D
dapan1121 已提交
1827
    loadRemoteDataCallback(pWrapper, &pBuf, code);
D
dapan1121 已提交
1828
    taosMemoryFree(pWrapper);
D
dapan1121 已提交
1829
  } else {
D
dapan1121 已提交
1830 1831 1832
    SResFetchReq* pMsg = taosMemoryCalloc(1, sizeof(SResFetchReq));
    if (NULL == pMsg) {
      pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
H
Haojun Liao 已提交
1833
      taosMemoryFree(pWrapper);
D
dapan1121 已提交
1834 1835 1836
      return pTaskInfo->code;
    }

D
dapan1121 已提交
1837
    qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %d/%" PRIzu,
1838 1839
           GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId,
           pSource->execId, sourceIndex, totalSources);
D
dapan1121 已提交
1840 1841 1842 1843 1844 1845 1846 1847 1848 1849 1850

    pMsg->header.vgId = htonl(pSource->addr.nodeId);
    pMsg->sId = htobe64(pSource->schedId);
    pMsg->taskId = htobe64(pSource->taskId);
    pMsg->queryId = htobe64(pTaskInfo->id.queryId);
    pMsg->execId = htonl(pSource->execId);

    // send the fetch remote task result reques
    SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
    if (NULL == pMsgSendInfo) {
      taosMemoryFreeClear(pMsg);
H
Haojun Liao 已提交
1851
      taosMemoryFree(pWrapper);
D
dapan1121 已提交
1852 1853 1854 1855 1856 1857 1858 1859 1860 1861 1862
      qError("%s prepare message %d failed", GET_TASKID(pTaskInfo), (int32_t)sizeof(SMsgSendInfo));
      pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
      return pTaskInfo->code;
    }

    pMsgSendInfo->param = pWrapper;
    pMsgSendInfo->paramFreeFp = taosMemoryFree;
    pMsgSendInfo->msgInfo.pData = pMsg;
    pMsgSendInfo->msgInfo.len = sizeof(SResFetchReq);
    pMsgSendInfo->msgType = pSource->fetchMsgType;
    pMsgSendInfo->fp = loadRemoteDataCallback;
1863

D
dapan1121 已提交
1864
    int64_t transporterId = 0;
1865 1866
    int32_t code =
        asyncSendMsgToServer(pExchangeInfo->pTransporter, &pSource->addr.epSet, &transporterId, pMsgSendInfo);
D
dapan1121 已提交
1867
  }
1868

1869 1870 1871
  return TSDB_CODE_SUCCESS;
}

1872 1873 1874 1875 1876 1877 1878 1879
void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int32_t numOfRows, int32_t dataLen, int64_t startTs,
                          SOperatorInfo* pOperator) {
  pInfo->totalRows += numOfRows;
  pInfo->totalSize += dataLen;
  pInfo->totalElapsed += (taosGetTimestampUs() - startTs);
  pOperator->resultInfo.totalRows += numOfRows;
}

H
Haojun Liao 已提交
1880
int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart) {
H
Haojun Liao 已提交
1881
  if (pColList == NULL) {  // data from other sources
1882
    blockDataCleanup(pRes);
dengyihao's avatar
dengyihao 已提交
1883
    *pNextStart = (char*)blockDecode(pRes, pData);
H
Haojun Liao 已提交
1884
  } else {  // extract data according to pColList
1885 1886 1887 1888 1889
    char* pStart = pData;

    int32_t numOfCols = htonl(*(int32_t*)pStart);
    pStart += sizeof(int32_t);

1890
    // todo refactor:extract method
1891
    SSysTableSchema* pSchema = (SSysTableSchema*)pStart;
dengyihao's avatar
dengyihao 已提交
1892
    for (int32_t i = 0; i < numOfCols; ++i) {
1893 1894 1895 1896 1897 1898 1899
      SSysTableSchema* p = (SSysTableSchema*)pStart;

      p->colId = htons(p->colId);
      p->bytes = htonl(p->bytes);
      pStart += sizeof(SSysTableSchema);
    }

1900
    SSDataBlock* pBlock = createDataBlock();
dengyihao's avatar
dengyihao 已提交
1901
    for (int32_t i = 0; i < numOfCols; ++i) {
1902 1903
      SColumnInfoData idata = createColumnInfoData(pSchema[i].type, pSchema[i].bytes, pSchema[i].colId);
      blockDataAppendColInfo(pBlock, &idata);
1904 1905
    }

1906
    blockDecode(pBlock, pStart);
1907
    blockDataEnsureCapacity(pRes, pBlock->info.rows);
1908

H
Haojun Liao 已提交
1909
    // data from mnode
1910
    pRes->info.rows = pBlock->info.rows;
1911 1912
    relocateColumnData(pRes, pColList, pBlock->pDataBlock, false);
    blockDataDestroy(pBlock);
1913
  }
1914

1915 1916
  // todo move this to time window aggregator, since the primary timestamp may not be known by exchange operator.
  blockDataUpdateTsWindow(pRes, 0);
1917 1918
  return TSDB_CODE_SUCCESS;
}
1919

L
Liu Jicong 已提交
1920 1921
static void* setAllSourcesCompleted(SOperatorInfo* pOperator, int64_t startTs) {
  SExchangeInfo* pExchangeInfo = pOperator->info;
1922
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
H
Haojun Liao 已提交
1923

1924
  int64_t              el = taosGetTimestampUs() - startTs;
H
Haojun Liao 已提交
1925
  SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
1926

H
Haojun Liao 已提交
1927
  pLoadInfo->totalElapsed += el;
H
Haojun Liao 已提交
1928

1929
  size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
L
Liu Jicong 已提交
1930 1931 1932
  qDebug("%s all %" PRIzu " sources are exhausted, total rows: %" PRIu64 " bytes:%" PRIu64 ", elapsed:%.2f ms",
         GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize,
         pLoadInfo->totalElapsed / 1000.0);
1933 1934 1935 1936 1937

  doSetOperatorCompleted(pOperator);
  return NULL;
}

1938 1939
static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo,
                                           SExecTaskInfo* pTaskInfo) {
1940 1941 1942 1943 1944 1945 1946 1947
  int32_t code = 0;
  int64_t startTs = taosGetTimestampUs();
  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSources);

  while (1) {
    int32_t completed = 0;
    for (int32_t i = 0; i < totalSources; ++i) {
      SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, i);
1948
      if (pDataInfo->status == EX_SOURCE_DATA_EXHAUSTED) {
1949
        completed += 1;
H
Haojun Liao 已提交
1950 1951
        continue;
      }
1952

1953
      if (pDataInfo->status != EX_SOURCE_DATA_READY) {
1954 1955 1956
        continue;
      }

1957 1958 1959 1960 1961
      if (pDataInfo->code != TSDB_CODE_SUCCESS) {
        code = pDataInfo->code;
        goto _error;
      }

L
Liu Jicong 已提交
1962
      SRetrieveTableRsp*     pRsp = pDataInfo->pRsp;
X
Xiaoyu Wang 已提交
1963
      SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, i);
1964

H
Haojun Liao 已提交
1965
      SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
1966
      if (pRsp->numOfRows == 0) {
1967 1968
        qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64
               ", totalRows:%" PRIu64 ", completed:%d try next %d/%" PRIzu,
D
dapan1121 已提交
1969
               GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows,
1970
               pExchangeInfo->loadInfo.totalRows, completed + 1, i + 1, totalSources);
1971
        pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
1972
        completed += 1;
D
dapan1121 已提交
1973
        taosMemoryFreeClear(pDataInfo->pRsp);
1974 1975
        continue;
      }
H
Haojun Liao 已提交
1976

1977
      SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
dengyihao's avatar
dengyihao 已提交
1978 1979 1980
      int32_t            index = 0;
      char*              pStart = pRetrieveRsp->data;
      while (index++ < pRetrieveRsp->numOfBlocks) {
1981
        SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
H
Haojun Liao 已提交
1982
        code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
1983 1984 1985 1986 1987 1988
        if (code != 0) {
          taosMemoryFreeClear(pDataInfo->pRsp);
          goto _error;
        }

        taosArrayPush(pExchangeInfo->pResultBlockList, &pb);
1989 1990
      }

1991
      updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator);
1992

1993
      if (pRsp->completed == 1) {
dengyihao's avatar
dengyihao 已提交
1994 1995 1996 1997
        qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
               " execId:%d"
               " index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64
               ", total:%.2f Kb,"
1998
               " completed:%d try next %d/%" PRIzu,
H
Haojun Liao 已提交
1999 2000 2001
               GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pRsp->numOfBlocks,
               pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0,
               completed + 1, i + 1, totalSources);
2002
        completed += 1;
2003
        pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
2004
      } else {
dengyihao's avatar
dengyihao 已提交
2005 2006 2007 2008
        qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
               " execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb",
               GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRsp->numOfBlocks,
               pRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0);
2009 2010
      }

2011 2012
      taosMemoryFreeClear(pDataInfo->pRsp);

2013 2014
      if (pDataInfo->status != EX_SOURCE_DATA_EXHAUSTED) {
        pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
2015 2016
        code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i);
        if (code != TSDB_CODE_SUCCESS) {
2017
          taosMemoryFreeClear(pDataInfo->pRsp);
2018 2019 2020 2021
          goto _error;
        }
      }

2022
      return;
2023 2024
    }

2025
    if (completed == totalSources) {
2026 2027
      setAllSourcesCompleted(pOperator, startTs);
      return;
2028
    }
H
Haojun Liao 已提交
2029 2030

    sched_yield();
2031 2032 2033 2034 2035 2036
  }

_error:
  pTaskInfo->code = code;
}

L
Liu Jicong 已提交
2037 2038 2039
static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
  SExchangeInfo* pExchangeInfo = pOperator->info;
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
2040

L
Liu Jicong 已提交
2041
  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSources);
2042 2043 2044
  int64_t startTs = taosGetTimestampUs();

  // Asynchronously send all fetch requests to all sources.
L
Liu Jicong 已提交
2045
  for (int32_t i = 0; i < totalSources; ++i) {
2046 2047
    int32_t code = doSendFetchDataRequest(pExchangeInfo, pTaskInfo, i);
    if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2048 2049
      pTaskInfo->code = code;
      return code;
2050 2051 2052 2053
    }
  }

  int64_t endTs = taosGetTimestampUs();
2054
  qDebug("%s send all fetch requests to %" PRIzu " sources completed, elapsed:%.2fms", GET_TASKID(pTaskInfo),
X
Xiaoyu Wang 已提交
2055
         totalSources, (endTs - startTs) / 1000.0);
2056

2057
  pOperator->status = OP_RES_TO_RETURN;
H
Haojun Liao 已提交
2058
  pOperator->cost.openCost = taosGetTimestampUs() - startTs;
2059

2060
  tsem_wait(&pExchangeInfo->ready);
H
Haojun Liao 已提交
2061
  return TSDB_CODE_SUCCESS;
2062 2063
}

2064
static int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
L
Liu Jicong 已提交
2065 2066
  SExchangeInfo* pExchangeInfo = pOperator->info;
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
2067

L
Liu Jicong 已提交
2068
  size_t  totalSources = taosArrayGetSize(pExchangeInfo->pSources);
2069
  int64_t startTs = taosGetTimestampUs();
2070

L
Liu Jicong 已提交
2071
  while (1) {
2072
    if (pExchangeInfo->current >= totalSources) {
2073 2074
      setAllSourcesCompleted(pOperator, startTs);
      return TSDB_CODE_SUCCESS;
2075
    }
2076

2077 2078 2079
    doSendFetchDataRequest(pExchangeInfo, pTaskInfo, pExchangeInfo->current);
    tsem_wait(&pExchangeInfo->ready);

dengyihao's avatar
dengyihao 已提交
2080
    SSourceDataInfo*       pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current);
X
Xiaoyu Wang 已提交
2081
    SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pExchangeInfo->current);
2082

H
Haojun Liao 已提交
2083
    if (pDataInfo->code != TSDB_CODE_SUCCESS) {
2084 2085
      qError("%s vgId:%d, taskID:0x%" PRIx64 " execId:%d error happens, code:%s", GET_TASKID(pTaskInfo),
             pSource->addr.nodeId, pSource->taskId, pSource->execId, tstrerror(pDataInfo->code));
H
Haojun Liao 已提交
2086
      pOperator->pTaskInfo->code = pDataInfo->code;
2087
      return pOperator->pTaskInfo->code;
H
Haojun Liao 已提交
2088 2089
    }

L
Liu Jicong 已提交
2090
    SRetrieveTableRsp*   pRsp = pDataInfo->pRsp;
H
Haojun Liao 已提交
2091
    SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
2092
    if (pRsp->numOfRows == 0) {
2093 2094
      qDebug("%s vgId:%d, taskID:0x%" PRIx64 " execId:%d %d of total completed, rowsOfSource:%" PRIu64
             ", totalRows:%" PRIu64 " try next",
D
dapan1121 已提交
2095
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pExchangeInfo->current + 1,
H
Haojun Liao 已提交
2096
             pDataInfo->totalRows, pLoadInfo->totalRows);
H
Haojun Liao 已提交
2097

2098
      pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
2099
      pExchangeInfo->current += 1;
D
dapan1121 已提交
2100
      taosMemoryFreeClear(pDataInfo->pRsp);
2101 2102
      continue;
    }
H
Haojun Liao 已提交
2103

2104 2105 2106
    SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;

    char*   pStart = pRetrieveRsp->data;
H
Haojun Liao 已提交
2107
    int32_t code = extractDataBlockFromFetchRsp(NULL, pStart, NULL, &pStart);
2108 2109

    if (pRsp->completed == 1) {
D
dapan1121 已提交
2110
      qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, rowsOfSource:%" PRIu64
L
Liu Jicong 已提交
2111
             ", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%" PRIzu,
2112
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows,
2113 2114
             pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize, pExchangeInfo->current + 1,
             totalSources);
2115

2116
      pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
2117 2118
      pExchangeInfo->current += 1;
    } else {
D
dapan1121 已提交
2119
      qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, totalRows:%" PRIu64
L
Liu Jicong 已提交
2120
             ", totalBytes:%" PRIu64,
2121
             GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows,
2122
             pLoadInfo->totalRows, pLoadInfo->totalSize);
2123 2124
    }

2125 2126 2127
    updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, startTs, pOperator);
    pDataInfo->totalRows += pRetrieveRsp->numOfRows;

2128
    taosMemoryFreeClear(pDataInfo->pRsp);
2129
    return TSDB_CODE_SUCCESS;
2130
  }
2131 2132
}

L
Liu Jicong 已提交
2133
static int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) {
2134
  if (OPTR_IS_OPENED(pOperator)) {
H
Haojun Liao 已提交
2135 2136 2137
    return TSDB_CODE_SUCCESS;
  }

2138 2139
  int64_t st = taosGetTimestampUs();

L
Liu Jicong 已提交
2140
  SExchangeInfo* pExchangeInfo = pOperator->info;
2141
  if (!pExchangeInfo->seqLoadData) {
H
Haojun Liao 已提交
2142 2143 2144 2145 2146 2147
    int32_t code = prepareConcurrentlyLoad(pOperator);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  }

2148
  OPTR_SET_OPENED(pOperator);
2149
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
H
Haojun Liao 已提交
2150 2151 2152
  return TSDB_CODE_SUCCESS;
}

2153 2154 2155 2156 2157
static void freeBlock(void* pParam) {
  SSDataBlock* pBlock = *(SSDataBlock**)pParam;
  blockDataDestroy(pBlock);
}

2158
static SSDataBlock* doLoadRemoteDataImpl(SOperatorInfo* pOperator) {
L
Liu Jicong 已提交
2159 2160
  SExchangeInfo* pExchangeInfo = pOperator->info;
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
2161

2162
  pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
2163
  if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2164 2165
    return NULL;
  }
2166

2167
  size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
H
Haojun Liao 已提交
2168

2169
  SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
2170
  if (pOperator->status == OP_EXEC_DONE) {
L
Liu Jicong 已提交
2171 2172 2173
    qDebug("%s all %" PRIzu " source(s) are exhausted, total rows:%" PRIu64 " bytes:%" PRIu64 ", elapsed:%.2f ms",
           GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize,
           pLoadInfo->totalElapsed / 1000.0);
2174 2175 2176
    return NULL;
  }

2177 2178 2179 2180 2181 2182 2183 2184 2185 2186 2187 2188 2189
  size_t size = taosArrayGetSize(pExchangeInfo->pResultBlockList);
  if (size == 0 || pExchangeInfo->rspBlockIndex >= size) {
    pExchangeInfo->rspBlockIndex = 0;
    taosArrayClearEx(pExchangeInfo->pResultBlockList, freeBlock);
    if (pExchangeInfo->seqLoadData) {
      seqLoadRemoteData(pOperator);
    } else {
      concurrentlyLoadRemoteDataImpl(pOperator, pExchangeInfo, pTaskInfo);
    }

    if (taosArrayGetSize(pExchangeInfo->pResultBlockList) == 0) {
      return NULL;
    }
2190
  }
2191 2192 2193

  // we have buffered retrieved datablock, return it directly
  return taosArrayGetP(pExchangeInfo->pResultBlockList, pExchangeInfo->rspBlockIndex++);
H
Haojun Liao 已提交
2194
}
2195

2196 2197 2198 2199 2200 2201 2202 2203
static SSDataBlock* doLoadRemoteData(SOperatorInfo* pOperator) {
  SExchangeInfo* pExchangeInfo = pOperator->info;
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

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

L
Liu Jicong 已提交
2204
  while (1) {
2205 2206 2207 2208 2209 2210 2211
    SSDataBlock* pBlock = doLoadRemoteDataImpl(pOperator);
    if (pBlock == NULL) {
      return NULL;
    }

    SLimitInfo* pLimitInfo = &pExchangeInfo->limitInfo;
    if (hasLimitOffsetInfo(pLimitInfo)) {
2212
      int32_t status = handleLimitOffset(pOperator, pLimitInfo, pBlock, false);
2213 2214 2215
      if (status == PROJECT_RETRIEVE_CONTINUE) {
        continue;
      } else if (status == PROJECT_RETRIEVE_DONE) {
2216
        size_t rows = pBlock->info.rows;
2217 2218 2219 2220 2221 2222
        pExchangeInfo->limitInfo.numOfOutputRows += rows;

        if (rows == 0) {
          doSetOperatorCompleted(pOperator);
          return NULL;
        } else {
2223
          return pBlock;
2224 2225 2226
        }
      }
    } else {
2227
      return pBlock;
2228 2229 2230 2231
    }
  }
}

2232
static int32_t initDataSource(int32_t numOfSources, SExchangeInfo* pInfo, const char* id) {
2233
  pInfo->pSourceDataInfo = taosArrayInit(numOfSources, sizeof(SSourceDataInfo));
H
Haojun Liao 已提交
2234 2235
  if (pInfo->pSourceDataInfo == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
2236 2237
  }

L
Liu Jicong 已提交
2238
  for (int32_t i = 0; i < numOfSources; ++i) {
2239
    SSourceDataInfo dataInfo = {0};
H
Haojun Liao 已提交
2240
    dataInfo.status = EX_SOURCE_DATA_NOT_READY;
2241
    dataInfo.taskId = id;
L
Liu Jicong 已提交
2242
    dataInfo.index = i;
X
Xiaoyu Wang 已提交
2243
    SSourceDataInfo* pDs = taosArrayPush(pInfo->pSourceDataInfo, &dataInfo);
2244
    if (pDs == NULL) {
H
Haojun Liao 已提交
2245 2246 2247 2248 2249 2250 2251 2252
      taosArrayDestroy(pInfo->pSourceDataInfo);
      return TSDB_CODE_OUT_OF_MEMORY;
    }
  }

  return TSDB_CODE_SUCCESS;
}

2253
static int32_t initExchangeOperator(SExchangePhysiNode* pExNode, SExchangeInfo* pInfo, const char* id) {
2254
  size_t numOfSources = LIST_LENGTH(pExNode->pSrcEndPoints);
H
Haojun Liao 已提交
2255

2256
  if (numOfSources == 0) {
X
Xiaoyu Wang 已提交
2257
    qError("%s invalid number: %d of sources in exchange operator", id, (int32_t)numOfSources);
2258 2259 2260
    return TSDB_CODE_INVALID_PARA;
  }

H
Haojun Liao 已提交
2261
  pInfo->pSources = taosArrayInit(numOfSources, sizeof(SDownstreamSourceNode));
wmmhello's avatar
wmmhello 已提交
2262
  if (pInfo->pSources == NULL) {
2263
    return TSDB_CODE_OUT_OF_MEMORY;
H
Haojun Liao 已提交
2264 2265
  }

L
Liu Jicong 已提交
2266
  for (int32_t i = 0; i < numOfSources; ++i) {
D
dapan1121 已提交
2267
    SDownstreamSourceNode* pNode = (SDownstreamSourceNode*)nodesListGetNode((SNodeList*)pExNode->pSrcEndPoints, i);
H
Haojun Liao 已提交
2268 2269
    taosArrayPush(pInfo->pSources, pNode);
  }
2270

2271
  initLimitInfo(pExNode->node.pLimit, pExNode->node.pSlimit, &pInfo->limitInfo);
2272 2273
  pInfo->self = taosAddRef(exchangeObjRefPool, pInfo);

2274
  return initDataSource(numOfSources, pInfo, id);
2275 2276 2277 2278 2279 2280
}

SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode* pExNode, SExecTaskInfo* pTaskInfo) {
  SExchangeInfo* pInfo = taosMemoryCalloc(1, sizeof(SExchangeInfo));
  SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
H
Haojun Liao 已提交
2281
    goto _error;
2282
  }
H
Haojun Liao 已提交
2283

2284
  int32_t code = initExchangeOperator(pExNode, pInfo, GET_TASKID(pTaskInfo));
2285 2286 2287
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2288 2289

  tsem_init(&pInfo->ready, 0, 0);
2290 2291
  pInfo->pDummyBlock = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
  pInfo->pResultBlockList = taosArrayInit(1, POINTER_BYTES);
2292

2293
  pInfo->seqLoadData = false;
2294
  pInfo->pTransporter = pTransporter;
2295

2296
  pOperator->name = "ExchangeOperator";
X
Xiaoyu Wang 已提交
2297
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;
X
Xiaoyu Wang 已提交
2298
  pOperator->blocking = false;
2299 2300
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
2301
  pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pDummyBlock->pDataBlock);
X
Xiaoyu Wang 已提交
2302
  pOperator->pTaskInfo = pTaskInfo;
2303

L
Liu Jicong 已提交
2304
  pOperator->fpSet = createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, NULL,
2305
                                         destroyExchangeOperatorInfo, NULL);
2306
  return pOperator;
H
Haojun Liao 已提交
2307

L
Liu Jicong 已提交
2308
_error:
H
Haojun Liao 已提交
2309
  if (pInfo != NULL) {
2310
    doDestroyExchangeOperatorInfo(pInfo);
H
Haojun Liao 已提交
2311 2312
  }

wafwerar's avatar
wafwerar 已提交
2313
  taosMemoryFreeClear(pOperator);
2314
  pTaskInfo->code = code;
H
Haojun Liao 已提交
2315
  return NULL;
2316 2317
}

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

L
Liu Jicong 已提交
2321
static bool needToMerge(SSDataBlock* pBlock, SArray* groupInfo, char** buf, int32_t rowIndex) {
2322 2323 2324 2325
  size_t size = taosArrayGetSize(groupInfo);
  if (size == 0) {
    return true;
  }
2326

2327 2328
  for (int32_t i = 0; i < size; ++i) {
    int32_t* index = taosArrayGet(groupInfo, i);
2329

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

2333 2334 2335
    if ((isNull && buf[i] != NULL) || (!isNull && buf[i] == NULL)) {
      return false;
    }
2336

2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349
    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;
      }
2350 2351 2352
    }
  }

2353
  return 0;
2354 2355
}

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

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

2363 2364 2365
    char* data = colDataGetData(pColInfo, rowIndex);
    memcpy(rowColData[i], data, colDataGetLength(pColInfo, rowIndex));
  }
2366

2367 2368
  return true;
}
2369

X
Xiaoyu Wang 已提交
2370
int32_t getTableScanInfo(SOperatorInfo* pOperator, int32_t* order, int32_t* scanFlag) {
2371
  // todo add more information about exchange operation
2372
  int32_t type = pOperator->operatorType;
X
Xiaoyu Wang 已提交
2373
  if (type == QUERY_NODE_PHYSICAL_PLAN_EXCHANGE || type == QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN ||
2374
      type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN ||
2375
      type == QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN || type == QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN) {
2376 2377 2378
    *order = TSDB_ORDER_ASC;
    *scanFlag = MAIN_SCAN;
    return TSDB_CODE_SUCCESS;
2379
  } else if (type == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN) {
2380 2381 2382 2383
    STableScanInfo* pTableScanInfo = pOperator->info;
    *order = pTableScanInfo->cond.order;
    *scanFlag = pTableScanInfo->scanFlag;
    return TSDB_CODE_SUCCESS;
2384 2385 2386 2387 2388
  } 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;
2389
  } else {
H
Haojun Liao 已提交
2390
    if (pOperator->pDownstream == NULL || pOperator->pDownstream[0] == NULL) {
2391
      return TSDB_CODE_INVALID_PARA;
H
Haojun Liao 已提交
2392
    } else {
2393
      return getTableScanInfo(pOperator->pDownstream[0], order, scanFlag);
2394 2395 2396
    }
  }
}
2397

2398
// this is a blocking operator
L
Liu Jicong 已提交
2399
static int32_t doOpenAggregateOptr(SOperatorInfo* pOperator) {
H
Haojun Liao 已提交
2400 2401
  if (OPTR_IS_OPENED(pOperator)) {
    return TSDB_CODE_SUCCESS;
2402 2403
  }

H
Haojun Liao 已提交
2404
  SExecTaskInfo*    pTaskInfo = pOperator->pTaskInfo;
2405
  SAggOperatorInfo* pAggInfo = pOperator->info;
H
Haojun Liao 已提交
2406

2407 2408
  SExprSupp*     pSup = &pOperator->exprSupp;
  SOperatorInfo* downstream = pOperator->pDownstream[0];
2409

2410 2411
  int64_t st = taosGetTimestampUs();

2412 2413 2414
  int32_t order = TSDB_ORDER_ASC;
  int32_t scanFlag = MAIN_SCAN;

H
Haojun Liao 已提交
2415
  while (1) {
2416
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
2417 2418 2419 2420
    if (pBlock == NULL) {
      break;
    }

2421 2422
    int32_t code = getTableScanInfo(pOperator, &order, &scanFlag);
    if (code != TSDB_CODE_SUCCESS) {
2423
      T_LONG_JMP(pTaskInfo->env, code);
2424
    }
2425

2426
    // there is an scalar expression that needs to be calculated before apply the group aggregation.
2427 2428 2429
    if (pAggInfo->scalarExprSup.pExprInfo != NULL) {
      SExprSupp* pSup1 = &pAggInfo->scalarExprSup;
      code = projectApplyFunctions(pSup1->pExprInfo, pBlock, pBlock, pSup1->pCtx, pSup1->numOfExprs, NULL);
2430
      if (code != TSDB_CODE_SUCCESS) {
2431
        T_LONG_JMP(pTaskInfo->env, code);
2432
      }
2433 2434
    }

2435
    // the pDataBlock are always the same one, no need to call this again
2436
    setExecutionContext(pOperator, pOperator->exprSupp.numOfExprs, pBlock->info.groupId);
2437
    setInputDataBlock(pSup, pBlock, order, scanFlag, true);
2438
    code = doAggregateImpl(pOperator, pSup->pCtx);
2439
    if (code != 0) {
2440
      T_LONG_JMP(pTaskInfo->env, code);
2441
    }
2442 2443
  }

2444
  initGroupedResultInfo(&pAggInfo->groupResInfo, pAggInfo->aggSup.pResultRowHashTable, 0);
H
Haojun Liao 已提交
2445
  OPTR_SET_OPENED(pOperator);
2446

2447
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
H
Haojun Liao 已提交
2448 2449 2450
  return TSDB_CODE_SUCCESS;
}

2451
static SSDataBlock* getAggregateResult(SOperatorInfo* pOperator) {
L
Liu Jicong 已提交
2452
  SAggOperatorInfo* pAggInfo = pOperator->info;
H
Haojun Liao 已提交
2453 2454 2455 2456 2457 2458
  SOptrBasicInfo*   pInfo = &pAggInfo->binfo;

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

L
Liu Jicong 已提交
2459
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
2460
  pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
H
Haojun Liao 已提交
2461
  if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
2462
    doSetOperatorCompleted(pOperator);
H
Haojun Liao 已提交
2463 2464 2465
    return NULL;
  }

H
Haojun Liao 已提交
2466
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
S
slzhou 已提交
2467 2468
  while (1) {
    doBuildResultDatablock(pOperator, pInfo, &pAggInfo->groupResInfo, pAggInfo->aggSup.pResultBuf);
H
Haojun Liao 已提交
2469
    doFilter(pAggInfo->pCondition, pInfo->pRes, NULL, NULL);
S
slzhou 已提交
2470

2471
    if (!hasRemainResults(&pAggInfo->groupResInfo)) {
S
slzhou 已提交
2472 2473 2474
      doSetOperatorCompleted(pOperator);
      break;
    }
2475

S
slzhou 已提交
2476 2477 2478 2479
    if (pInfo->pRes->info.rows > 0) {
      break;
    }
  }
2480

2481
  size_t rows = blockDataGetNumOfRows(pInfo->pRes);
2482 2483
  pOperator->resultInfo.totalRows += rows;

2484
  return (rows == 0) ? NULL : pInfo->pRes;
2485 2486
}

wmmhello's avatar
wmmhello 已提交
2487
int32_t aggEncodeResultRow(SOperatorInfo* pOperator, char** result, int32_t* length) {
2488
  if (result == NULL || length == NULL) {
wmmhello's avatar
wmmhello 已提交
2489 2490 2491
    return TSDB_CODE_TSC_INVALID_INPUT;
  }
  SOptrBasicInfo* pInfo = (SOptrBasicInfo*)(pOperator->info);
2492
  SAggSupporter*  pSup = (SAggSupporter*)POINTER_SHIFT(pOperator->info, sizeof(SOptrBasicInfo));
2493
  int32_t         size = tSimpleHashGetSize(pSup->pResultRowHashTable);
2494 2495 2496
  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 已提交
2497

C
Cary Xu 已提交
2498 2499 2500 2501 2502 2503
  // no result
  if (getTotalBufSize(pSup->pResultBuf) == 0) {
    *result = NULL;
    *length = 0;
    return TSDB_CODE_SUCCESS;
  }
2504

wmmhello's avatar
wmmhello 已提交
2505
  *result = (char*)taosMemoryCalloc(1, totalSize);
L
Liu Jicong 已提交
2506
  if (*result == NULL) {
wmmhello's avatar
wmmhello 已提交
2507
    return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
2508
  }
wmmhello's avatar
wmmhello 已提交
2509

wmmhello's avatar
wmmhello 已提交
2510
  int32_t offset = sizeof(int32_t);
wmmhello's avatar
wmmhello 已提交
2511 2512
  *(int32_t*)(*result + offset) = size;
  offset += sizeof(int32_t);
2513 2514

  // prepare memory
2515
  SResultRowPosition* pos = &pInfo->resultRowInfo.cur;
dengyihao's avatar
dengyihao 已提交
2516 2517
  void*               pPage = getBufPage(pSup->pResultBuf, pos->pageId);
  SResultRow*         pRow = (SResultRow*)((char*)pPage + pos->offset);
2518 2519
  setBufPageDirty(pPage, true);
  releaseBufPage(pSup->pResultBuf, pPage);
L
Liu Jicong 已提交
2520

2521 2522 2523 2524
  int32_t iter = 0;
  void*   pIter = NULL;
  while ((pIter = tSimpleHashIterate(pSup->pResultRowHashTable, pIter, &iter))) {
    void*               key = tSimpleHashGetKey(pIter, &keyLen);
2525
    SResultRowPosition* p1 = (SResultRowPosition*)pIter;
2526

dengyihao's avatar
dengyihao 已提交
2527
    pPage = (SFilePage*)getBufPage(pSup->pResultBuf, p1->pageId);
2528
    pRow = (SResultRow*)((char*)pPage + p1->offset);
2529 2530
    setBufPageDirty(pPage, true);
    releaseBufPage(pSup->pResultBuf, pPage);
wmmhello's avatar
wmmhello 已提交
2531 2532 2533

    // recalculate the result size
    int32_t realTotalSize = offset + sizeof(int32_t) + keyLen + sizeof(int32_t) + pSup->resultRowSize;
L
Liu Jicong 已提交
2534
    if (realTotalSize > totalSize) {
wmmhello's avatar
wmmhello 已提交
2535
      char* tmp = (char*)taosMemoryRealloc(*result, realTotalSize);
L
Liu Jicong 已提交
2536
      if (tmp == NULL) {
wafwerar's avatar
wafwerar 已提交
2537
        taosMemoryFree(*result);
wmmhello's avatar
wmmhello 已提交
2538
        *result = NULL;
wmmhello's avatar
wmmhello 已提交
2539
        return TSDB_CODE_OUT_OF_MEMORY;
L
Liu Jicong 已提交
2540
      } else {
wmmhello's avatar
wmmhello 已提交
2541 2542 2543 2544 2545 2546 2547 2548 2549 2550 2551 2552
        *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);
2553
    memcpy(*result + offset, pRow, pSup->resultRowSize);
wmmhello's avatar
wmmhello 已提交
2554 2555 2556
    offset += pSup->resultRowSize;
  }

wmmhello's avatar
wmmhello 已提交
2557 2558 2559 2560
  *(int32_t*)(*result) = offset;
  *length = offset;

  return TDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
2561 2562
}

2563 2564 2565 2566 2567
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);
2568
      return PROJECT_RETRIEVE_CONTINUE;
2569 2570 2571
    } else if (pLimitInfo->currentGroupId != pBlock->info.groupId) {
      // now it is the data from a new group
      pLimitInfo->remainGroupOffset -= 1;
2572 2573

      // ignore data block in current group
2574 2575
      if (pLimitInfo->remainGroupOffset > 0) {
        blockDataCleanup(pBlock);
2576 2577 2578 2579 2580
        return PROJECT_RETRIEVE_CONTINUE;
      }
    }

    // set current group id of the project operator
2581
    pLimitInfo->currentGroupId = pBlock->info.groupId;
2582 2583
  }

2584
  // here check for a new group data, we need to handle the data of the previous group.
2585 2586 2587
  if (pLimitInfo->currentGroupId != 0 && pLimitInfo->currentGroupId != pBlock->info.groupId) {
    pLimitInfo->numOfOutputGroups += 1;
    if ((pLimitInfo->slimit.limit > 0) && (pLimitInfo->slimit.limit <= pLimitInfo->numOfOutputGroups)) {
2588
      pOperator->status = OP_EXEC_DONE;
2589
      blockDataCleanup(pBlock);
2590 2591 2592 2593 2594

      return PROJECT_RETRIEVE_DONE;
    }

    // reset the value for a new group data
2595 2596
    pLimitInfo->numOfOutputRows = 0;
    pLimitInfo->remainOffset = pLimitInfo->limit.offset;
2597 2598 2599 2600 2601

    // existing rows that belongs to previous group.
    if (pBlock->info.rows > 0) {
      return PROJECT_RETRIEVE_DONE;
    }
2602 2603 2604 2605 2606
  }

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

  // set current group id
2607
  pLimitInfo->currentGroupId = pBlock->info.groupId;
2608

2609 2610 2611
  if (pLimitInfo->remainOffset >= pBlock->info.rows) {
    pLimitInfo->remainOffset -= pBlock->info.rows;
    blockDataCleanup(pBlock);
2612
    return PROJECT_RETRIEVE_CONTINUE;
2613 2614 2615
  } else if (pLimitInfo->remainOffset < pBlock->info.rows && pLimitInfo->remainOffset > 0) {
    blockDataTrimFirstNRows(pBlock, pLimitInfo->remainOffset);
    pLimitInfo->remainOffset = 0;
2616 2617
  }

2618
  // check for the limitation in each group
2619 2620 2621 2622
  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) {
2623 2624 2625
      pOperator->status = OP_EXEC_DONE;
    }

2626
    return PROJECT_RETRIEVE_DONE;
2627
  }
2628

2629
  // todo optimize performance
2630 2631
  // 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.
2632 2633
  if ((!holdDataInBuf) || (pBlock->info.rows >= pOperator->resultInfo.threshold) || pLimitInfo->slimit.offset != -1 ||
      pLimitInfo->slimit.limit != -1) {
2634
    return PROJECT_RETRIEVE_DONE;
L
Liu Jicong 已提交
2635
  } else {  // not full enough, continue to accumulate the output data in the buffer.
2636 2637 2638 2639
    return PROJECT_RETRIEVE_CONTINUE;
  }
}

2640
static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag);
L
Liu Jicong 已提交
2641 2642
static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
                                               SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
2643
  pInfo->totalInputRows = pInfo->existNewGroupBlock->info.rows;
2644 2645 2646 2647 2648
  SSDataBlock* pResBlock = pInfo->pFinalRes;

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

L
Liu Jicong 已提交
2650 2651
  int64_t ekey =
      Q_STATUS_EQUAL(pTaskInfo->status, TASK_COMPLETED) ? pInfo->win.ekey : pInfo->existNewGroupBlock->info.window.ekey;
2652 2653
  taosResetFillInfo(pInfo->pFillInfo, getFillInfoStart(pInfo->pFillInfo));

2654
  blockDataCleanup(pInfo->pRes);
2655 2656 2657 2658
  doApplyScalarCalculation(pOperator, pInfo->existNewGroupBlock, order, scanFlag);

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

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

2663
  pInfo->curGroupId = pInfo->existNewGroupBlock->info.groupId;
2664 2665 2666
  pInfo->existNewGroupBlock = NULL;
}

L
Liu Jicong 已提交
2667 2668
static void doHandleRemainBlockFromNewGroup(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo,
                                            SResultInfo* pResultInfo, SExecTaskInfo* pTaskInfo) {
2669
  if (taosFillHasMoreResults(pInfo->pFillInfo)) {
H
Haojun Liao 已提交
2670 2671
    int32_t numOfResultRows = pResultInfo->capacity - pInfo->pFinalRes->info.rows;
    taosFillResultDataBlock(pInfo->pFillInfo, pInfo->pFinalRes, numOfResultRows);
2672 2673
    pInfo->pRes->info.groupId = pInfo->curGroupId;
    return;
2674 2675 2676 2677
  }

  // handle the cached new group data block
  if (pInfo->existNewGroupBlock) {
2678 2679 2680 2681 2682 2683
    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 已提交
2684
  SExprSupp*         pSup = &pOperator->exprSupp;
2685
  setInputDataBlock(pSup, pBlock, order, scanFlag, false);
2686 2687
  projectApplyFunctions(pSup->pExprInfo, pInfo->pRes, pBlock, pSup->pCtx, pSup->numOfExprs, NULL);

2688 2689 2690 2691
  // 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);
2692

2693 2694
  projectApplyFunctions(pNoFillSupp->pExprInfo, pInfo->pRes, pBlock, pNoFillSupp->pCtx, pNoFillSupp->numOfExprs, NULL);
  pInfo->pRes->info.groupId = pBlock->info.groupId;
2695 2696
}

S
slzhou 已提交
2697
static SSDataBlock* doFillImpl(SOperatorInfo* pOperator) {
L
Liu Jicong 已提交
2698 2699
  SFillOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;
2700

H
Haojun Liao 已提交
2701
  SResultInfo* pResultInfo = &pOperator->resultInfo;
H
Haojun Liao 已提交
2702
  SSDataBlock* pResBlock = pInfo->pFinalRes;
2703 2704

  blockDataCleanup(pResBlock);
2705

H
Haojun Liao 已提交
2706 2707
  int32_t order = TSDB_ORDER_ASC;
  int32_t scanFlag = MAIN_SCAN;
2708
  getTableScanInfo(pOperator, &order, &scanFlag);
2709

2710
  doHandleRemainBlockFromNewGroup(pOperator, pInfo, pResultInfo, pTaskInfo);
2711
  if (pResBlock->info.rows > 0) {
2712
    pResBlock->info.groupId = pInfo->curGroupId;
2713
    return pResBlock;
H
Haojun Liao 已提交
2714
  }
2715

H
Haojun Liao 已提交
2716
  SOperatorInfo* pDownstream = pOperator->pDownstream[0];
L
Liu Jicong 已提交
2717
  while (1) {
2718
    SSDataBlock* pBlock = pDownstream->fpSet.getNextFn(pDownstream);
2719 2720
    if (pBlock == NULL) {
      if (pInfo->totalInputRows == 0) {
2721
        doSetOperatorCompleted(pOperator);
2722 2723
        return NULL;
      }
2724

2725
      taosFillSetStartInfo(pInfo->pFillInfo, 0, pInfo->win.ekey);
2726
    } else {
2727
      blockDataUpdateTsWindow(pBlock, pInfo->primarySrcSlotId);
2728 2729

      blockDataCleanup(pInfo->pRes);
2730 2731
      blockDataEnsureCapacity(pInfo->pRes, pBlock->info.rows);
      blockDataEnsureCapacity(pInfo->pFinalRes, pBlock->info.rows);
2732
      doApplyScalarCalculation(pOperator, pBlock, order, scanFlag);
2733

H
Haojun Liao 已提交
2734 2735 2736
      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;
2737

H
Haojun Liao 已提交
2738 2739 2740 2741 2742
        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 已提交
2743
        taosFillSetInputDataBlock(pInfo->pFillInfo, pInfo->pRes);
L
Liu Jicong 已提交
2744
      } else if (pInfo->curGroupId != pBlock->info.groupId) {  // the new group data block
2745 2746 2747 2748 2749
        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);
2750 2751 2752
      }
    }

2753 2754
    int32_t numOfResultRows = pOperator->resultInfo.capacity - pResBlock->info.rows;
    taosFillResultDataBlock(pInfo->pFillInfo, pResBlock, numOfResultRows);
2755 2756

    // current group has no more result to return
2757
    if (pResBlock->info.rows > 0) {
2758 2759
      // 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
2760
      if (pResBlock->info.rows > pResultInfo->threshold || pBlock == NULL || pInfo->existNewGroupBlock != NULL) {
2761
        pResBlock->info.groupId = pInfo->curGroupId;
2762
        return pResBlock;
2763 2764
      }

2765
      doHandleRemainBlockFromNewGroup(pOperator, pInfo, pResultInfo, pTaskInfo);
2766
      if (pResBlock->info.rows >= pOperator->resultInfo.threshold || pBlock == NULL) {
2767
        pResBlock->info.groupId = pInfo->curGroupId;
2768
        return pResBlock;
2769 2770 2771
      }
    } else if (pInfo->existNewGroupBlock) {  // try next group
      assert(pBlock != NULL);
2772 2773 2774 2775

      blockDataCleanup(pResBlock);

      doHandleRemainBlockForNewGroupImpl(pOperator, pInfo, pResultInfo, pTaskInfo);
2776
      if (pResBlock->info.rows > pResultInfo->threshold) {
2777
        pResBlock->info.groupId = pInfo->curGroupId;
2778
        return pResBlock;
2779 2780 2781 2782 2783 2784 2785
      }
    } else {
      return NULL;
    }
  }
}

S
slzhou 已提交
2786 2787 2788 2789 2790 2791 2792 2793
static SSDataBlock* doFill(SOperatorInfo* pOperator) {
  SFillOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*     pTaskInfo = pOperator->pTaskInfo;

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

S
slzhou 已提交
2794
  SSDataBlock* fillResult = NULL;
S
slzhou 已提交
2795
  while (true) {
S
slzhou 已提交
2796
    fillResult = doFillImpl(pOperator);
S
slzhou 已提交
2797 2798 2799 2800 2801
    if (fillResult == NULL) {
      doSetOperatorCompleted(pOperator);
      break;
    }

H
Haojun Liao 已提交
2802
    doFilter(pInfo->pCondition, fillResult, &pInfo->matchInfo, NULL);
S
slzhou 已提交
2803 2804 2805 2806 2807
    if (fillResult->info.rows > 0) {
      break;
    }
  }

S
slzhou 已提交
2808
  if (fillResult != NULL) {
2809
    pOperator->resultInfo.totalRows += fillResult->info.rows;
S
slzhou 已提交
2810
  }
S
slzhou 已提交
2811

S
slzhou 已提交
2812
  return fillResult;
S
slzhou 已提交
2813 2814
}

2815
void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs) {
C
Cary Xu 已提交
2816 2817 2818 2819 2820
  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 已提交
2821
      }
2822
    }
C
Cary Xu 已提交
2823 2824 2825

    taosMemoryFree(pExprInfo->base.pParam);
    taosMemoryFree(pExprInfo->pExpr);
H
Haojun Liao 已提交
2826 2827 2828
  }
}

2829 2830 2831 2832 2833
static void destroyOperatorInfo(SOperatorInfo* pOperator) {
  if (pOperator == NULL) {
    return;
  }

2834
  if (pOperator->fpSet.closeFn != NULL) {
2835
    pOperator->fpSet.closeFn(pOperator->info);
2836 2837
  }

H
Haojun Liao 已提交
2838
  if (pOperator->pDownstream != NULL) {
L
Liu Jicong 已提交
2839
    for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
H
Haojun Liao 已提交
2840
      destroyOperatorInfo(pOperator->pDownstream[i]);
2841 2842
    }

wafwerar's avatar
wafwerar 已提交
2843
    taosMemoryFreeClear(pOperator->pDownstream);
H
Haojun Liao 已提交
2844
    pOperator->numOfDownstream = 0;
2845 2846
  }

2847
  cleanupExprSupp(&pOperator->exprSupp);
wafwerar's avatar
wafwerar 已提交
2848
  taosMemoryFreeClear(pOperator);
2849 2850
}

2851 2852 2853 2854 2855 2856 2857 2858 2859 2860 2861 2862 2863 2864 2865
int32_t getBufferPgSize(int32_t rowSize, uint32_t* defaultPgsz, uint32_t* defaultBufsz) {
  *defaultPgsz = 4096;
  while (*defaultPgsz < rowSize * 4) {
    *defaultPgsz <<= 1u;
  }

  // at least four pages need to be in buffer
  *defaultBufsz = 4096 * 256;
  if ((*defaultBufsz) <= (*defaultPgsz)) {
    (*defaultBufsz) = (*defaultPgsz) * 4;
  }

  return 0;
}

dengyihao's avatar
dengyihao 已提交
2866 2867
int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, size_t keyBufSize,
                         const char* pKey) {
L
Liu Jicong 已提交
2868
  int32_t    code = 0;
2869 2870
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);

2871
  pAggSup->currentPageId = -1;
dengyihao's avatar
dengyihao 已提交
2872 2873
  pAggSup->resultRowSize = getResultRowSize(pCtx, numOfOutput);
  pAggSup->keyBuf = taosMemoryCalloc(1, keyBufSize + POINTER_BYTES + sizeof(int64_t));
2874
  pAggSup->pResultRowHashTable = tSimpleHashInit(10, hashFn);
2875

H
Haojun Liao 已提交
2876
  if (pAggSup->keyBuf == NULL || pAggSup->pResultRowHashTable == NULL) {
2877 2878 2879
    return TSDB_CODE_OUT_OF_MEMORY;
  }

dengyihao's avatar
dengyihao 已提交
2880
  uint32_t defaultPgsz = 0;
2881 2882
  uint32_t defaultBufsz = 0;
  getBufferPgSize(pAggSup->resultRowSize, &defaultPgsz, &defaultBufsz);
H
Haojun Liao 已提交
2883

wafwerar's avatar
wafwerar 已提交
2884
  if (!osTempSpaceAvailable()) {
H
Haojun Liao 已提交
2885 2886 2887
    code = TSDB_CODE_NO_AVAIL_DISK;
    qError("Init stream agg supporter failed since %s, %s", terrstr(code), pKey);
    return code;
wafwerar's avatar
wafwerar 已提交
2888
  }
2889

H
Haojun Liao 已提交
2890
  code = createDiskbasedBuf(&pAggSup->pResultBuf, defaultPgsz, defaultBufsz, pKey, tsTempDir);
H
Haojun Liao 已提交
2891
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
2892
    qError("Create agg result buf failed since %s, %s", tstrerror(code), pKey);
H
Haojun Liao 已提交
2893 2894 2895
    return code;
  }

H
Haojun Liao 已提交
2896
  return code;
2897 2898
}

2899
void cleanupAggSup(SAggSupporter* pAggSup) {
wafwerar's avatar
wafwerar 已提交
2900
  taosMemoryFreeClear(pAggSup->keyBuf);
2901
  tSimpleHashCleanup(pAggSup->pResultRowHashTable);
H
Haojun Liao 已提交
2902
  destroyDiskbasedBuf(pAggSup->pResultBuf);
2903 2904
}

L
Liu Jicong 已提交
2905 2906
int32_t initAggInfo(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize,
                    const char* pkey) {
2907 2908 2909 2910 2911
  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

2912 2913 2914 2915 2916
  code = doInitAggInfoSup(pAggSup, pSup->pCtx, numOfCols, keyBufSize, pkey);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

L
Liu Jicong 已提交
2917
  for (int32_t i = 0; i < numOfCols; ++i) {
2918
    pSup->pCtx[i].saveHandle.pBuf = pAggSup->pResultBuf;
2919 2920
  }

2921
  return TSDB_CODE_SUCCESS;
H
Haojun Liao 已提交
2922 2923
}

L
Liu Jicong 已提交
2924
void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows) {
wmmhello's avatar
wmmhello 已提交
2925
  ASSERT(numOfRows != 0);
2926 2927
  pResultInfo->capacity = numOfRows;
  pResultInfo->threshold = numOfRows * 0.75;
2928

2929 2930
  if (pResultInfo->threshold == 0) {
    pResultInfo->threshold = numOfRows;
2931 2932 2933
  }
}

2934 2935 2936 2937 2938
void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock) {
  pInfo->pRes = pBlock;
  initResultRowInfo(&pInfo->resultRowInfo);
}

5
54liuyao 已提交
2939
void* destroySqlFunctionCtx(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
2940 2941 2942 2943 2944 2945 2946 2947 2948 2949
  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);
2950
    taosMemoryFreeClear(pCtx[i].subsidiaries.buf);
2951 2952 2953 2954 2955 2956 2957 2958
    taosMemoryFree(pCtx[i].input.pData);
    taosMemoryFree(pCtx[i].input.pColumnDataAgg);
  }

  taosMemoryFreeClear(pCtx);
  return NULL;
}

2959
int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr) {
2960 2961 2962 2963
  pSup->pExprInfo = pExprInfo;
  pSup->numOfExprs = numOfExpr;
  if (pSup->pExprInfo != NULL) {
    pSup->pCtx = createSqlFunctionCtx(pExprInfo, numOfExpr, &pSup->rowEntryInfoOffset);
2964 2965 2966
    if (pSup->pCtx == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
2967
  }
2968 2969

  return TSDB_CODE_SUCCESS;
2970 2971
}

2972 2973 2974 2975
void cleanupExprSupp(SExprSupp* pSupp) {
  destroySqlFunctionCtx(pSupp->pCtx, pSupp->numOfExprs);
  if (pSupp->pExprInfo != NULL) {
    destroyExprInfo(pSupp->pExprInfo, pSupp->numOfExprs);
C
Cary Xu 已提交
2976
    taosMemoryFreeClear(pSupp->pExprInfo);
2977
  }
H
Haojun Liao 已提交
2978 2979 2980 2981 2982 2983

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

2984 2985 2986
  taosMemoryFree(pSupp->rowEntryInfoOffset);
}

2987 2988
SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SAggPhysiNode* pAggNode,
                                           SExecTaskInfo* pTaskInfo) {
wafwerar's avatar
wafwerar 已提交
2989
  SAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SAggOperatorInfo));
L
Liu Jicong 已提交
2990
  SOperatorInfo*    pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
H
Haojun Liao 已提交
2991 2992 2993
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
H
Haojun Liao 已提交
2994

H
Haojun Liao 已提交
2995 2996 2997 2998
  SSDataBlock* pResBlock = createResDataBlock(pAggNode->node.pOutputDataBlockDesc);
  initBasicInfo(&pInfo->binfo, pResBlock);

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

3001 3002 3003
  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 已提交
3004
  if (code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
3005 3006
    goto _error;
  }
H
Haojun Liao 已提交
3007

H
Haojun Liao 已提交
3008 3009 3010 3011 3012 3013
  int32_t    numOfScalarExpr = 0;
  SExprInfo* pScalarExprInfo = NULL;
  if (pAggNode->pExprs != NULL) {
    pScalarExprInfo = createExprInfo(pAggNode->pExprs, NULL, &numOfScalarExpr);
  }

3014 3015 3016 3017
  code = initExprSupp(&pInfo->scalarExprSup, pScalarExprInfo, numOfScalarExpr);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
3018

H
Haojun Liao 已提交
3019
  pInfo->binfo.mergeResultBlock = pAggNode->mergeDataBlock;
3020
  pInfo->groupId = UINT64_MAX;
H
Haojun Liao 已提交
3021
  pInfo->pCondition = pAggNode->node.pConditions;
dengyihao's avatar
dengyihao 已提交
3022
  pOperator->name = "TableAggregate";
3023
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_AGG;
3024
  pOperator->blocking = true;
dengyihao's avatar
dengyihao 已提交
3025 3026 3027
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
H
Haojun Liao 已提交
3028

3029
  pOperator->fpSet = createOperatorFpSet(doOpenAggregateOptr, getAggregateResult, NULL, NULL, destroyAggOperatorInfo,
3030
                                         NULL);
H
Haojun Liao 已提交
3031

3032 3033 3034 3035 3036 3037
  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 已提交
3038 3039 3040 3041
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
3042 3043

  return pOperator;
H
Haojun Liao 已提交
3044

L
Liu Jicong 已提交
3045
_error:
H
Haojun Liao 已提交
3046 3047 3048 3049
  if (pInfo != NULL) {
    destroyAggOperatorInfo(pInfo);
  }

3050 3051 3052
  if (pOperator != NULL) {
    cleanupExprSupp(&pOperator->exprSupp);
  }
H
Haojun Liao 已提交
3053

3054
  taosMemoryFreeClear(pOperator);
H
Haojun Liao 已提交
3055 3056
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
  return NULL;
3057 3058
}

3059
void cleanupBasicInfo(SOptrBasicInfo* pInfo) {
3060
  assert(pInfo != NULL);
H
Haojun Liao 已提交
3061
  pInfo->pRes = blockDataDestroy(pInfo->pRes);
3062 3063
}

H
Haojun Liao 已提交
3064 3065 3066 3067 3068 3069 3070
static void freeItem(void* pItem) {
  void** p = pItem;
  if (*p != NULL) {
    taosMemoryFreeClear(*p);
  }
}

3071
void destroyAggOperatorInfo(void* param) {
L
Liu Jicong 已提交
3072
  SAggOperatorInfo* pInfo = (SAggOperatorInfo*)param;
L
Liu Jicong 已提交
3073 3074
  cleanupBasicInfo(&pInfo->binfo);

H
Haojun Liao 已提交
3075
  cleanupAggSup(&pInfo->aggSup);
S
shenglian zhou 已提交
3076
  cleanupExprSupp(&pInfo->scalarExprSup);
H
Haojun Liao 已提交
3077
  cleanupGroupResInfo(&pInfo->groupResInfo);
D
dapan1121 已提交
3078
  taosMemoryFreeClear(param);
3079
}
3080

3081
void destroyFillOperatorInfo(void* param) {
L
Liu Jicong 已提交
3082
  SFillOperatorInfo* pInfo = (SFillOperatorInfo*)param;
3083
  pInfo->pFillInfo = taosDestroyFillInfo(pInfo->pFillInfo);
H
Haojun Liao 已提交
3084
  pInfo->pRes = blockDataDestroy(pInfo->pRes);
H
Haojun Liao 已提交
3085 3086
  pInfo->pFinalRes = blockDataDestroy(pInfo->pFinalRes);

3087
  cleanupExprSupp(&pInfo->noFillExprSupp);
H
Haojun Liao 已提交
3088

wafwerar's avatar
wafwerar 已提交
3089
  taosMemoryFreeClear(pInfo->p);
H
Haojun Liao 已提交
3090
  taosArrayDestroy(pInfo->matchInfo.pList);
D
dapan1121 已提交
3091
  taosMemoryFreeClear(param);
3092 3093
}

3094
void destroyExchangeOperatorInfo(void* param) {
L
Liu Jicong 已提交
3095
  SExchangeInfo* pExInfo = (SExchangeInfo*)param;
3096 3097 3098
  taosRemoveRef(exchangeObjRefPool, pExInfo->self);
}

L
Liu Jicong 已提交
3099
void freeSourceDataInfo(void* p) {
3100 3101 3102 3103
  SSourceDataInfo* pInfo = (SSourceDataInfo*)p;
  taosMemoryFreeClear(pInfo->pRsp);
}

3104
void doDestroyExchangeOperatorInfo(void* param) {
X
Xiaoyu Wang 已提交
3105
  SExchangeInfo* pExInfo = (SExchangeInfo*)param;
3106

H
Haojun Liao 已提交
3107
  taosArrayDestroy(pExInfo->pSources);
3108
  taosArrayDestroyEx(pExInfo->pSourceDataInfo, freeSourceDataInfo);
3109 3110 3111 3112

  if (pExInfo->pResultBlockList != NULL) {
    taosArrayDestroyEx(pExInfo->pResultBlockList, freeBlock);
    pExInfo->pResultBlockList = NULL;
H
Haojun Liao 已提交
3113 3114
  }

3115
  blockDataDestroy(pExInfo->pDummyBlock);
L
Liu Jicong 已提交
3116

3117
  tsem_destroy(&pExInfo->ready);
D
dapan1121 已提交
3118
  taosMemoryFreeClear(param);
H
Haojun Liao 已提交
3119 3120
}

H
Haojun Liao 已提交
3121 3122 3123 3124
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 已提交
3125

H
Haojun Liao 已提交
3126 3127 3128
  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 已提交
3129

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

H
Haojun Liao 已提交
3133 3134 3135 3136 3137 3138 3139
  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 已提交
3140
  pInfo->p = taosMemoryCalloc(numOfCols, POINTER_BYTES);
3141

H
Haojun Liao 已提交
3142
  if (pInfo->pFillInfo == NULL || pInfo->p == NULL) {
H
Haojun Liao 已提交
3143 3144
    taosMemoryFree(pInfo->pFillInfo);
    taosMemoryFree(pInfo->p);
H
Haojun Liao 已提交
3145 3146 3147 3148 3149 3150
    return TSDB_CODE_OUT_OF_MEMORY;
  } else {
    return TSDB_CODE_SUCCESS;
  }
}

3151
static bool isWstartColumnExist(SFillOperatorInfo* pInfo) {
3152
  if (pInfo->noFillExprSupp.numOfExprs == 0) {
3153 3154
    return false;
  }
3155 3156 3157

  for (int32_t i = 0; i < pInfo->noFillExprSupp.numOfExprs; ++i) {
    SExprInfo* exprInfo = pInfo->noFillExprSupp.pExprInfo + i;
3158
    if (exprInfo->pExpr->nodeType == QUERY_NODE_COLUMN && exprInfo->base.numOfParams == 1 &&
3159 3160 3161 3162 3163 3164 3165
        exprInfo->base.pParam[0].pCol->colType == COLUMN_TYPE_WINDOW_START) {
      return true;
    }
  }
  return false;
}

3166 3167
static int32_t createPrimaryTsExprIfNeeded(SFillOperatorInfo* pInfo, SFillPhysiNode* pPhyFillNode, SExprSupp* pExprSupp,
                                           const char* idStr) {
3168
  bool wstartExist = isWstartColumnExist(pInfo);
3169

3170 3171
  if (wstartExist == false) {
    if (pPhyFillNode->pWStartTs->type != QUERY_NODE_TARGET) {
3172
      qError("pWStartTs of fill physical node is not a target node, %s", idStr);
3173 3174 3175
      return TSDB_CODE_QRY_SYS_ERROR;
    }

3176 3177
    SExprInfo* pExpr = taosMemoryRealloc(pExprSupp->pExprInfo, (pExprSupp->numOfExprs + 1) * sizeof(SExprInfo));
    if (pExpr == NULL) {
3178 3179 3180
      return TSDB_CODE_OUT_OF_MEMORY;
    }

3181 3182 3183
    createExprFromTargetNode(&pExpr[pExprSupp->numOfExprs], (STargetNode*)pPhyFillNode->pWStartTs);
    pExprSupp->numOfExprs += 1;
    pExprSupp->pExprInfo = pExpr;
3184
  }
3185

3186 3187 3188
  return TSDB_CODE_SUCCESS;
}

L
Liu Jicong 已提交
3189 3190
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode* pPhyFillNode,
                                      SExecTaskInfo* pTaskInfo) {
3191 3192 3193 3194 3195 3196
  SFillOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SFillOperatorInfo));
  SOperatorInfo*     pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

H
Haojun Liao 已提交
3197 3198 3199 3200
  pInfo->pRes = createResDataBlock(pPhyFillNode->node.pOutputDataBlockDesc);
  SExprInfo* pExprInfo = createExprInfo(pPhyFillNode->pFillExprs, NULL, &pInfo->numOfExpr);
  pOperator->exprSupp.pExprInfo = pExprInfo;

3201 3202 3203 3204 3205 3206 3207 3208
  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);
3209 3210 3211
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
3212

L
Liu Jicong 已提交
3213
  SInterval* pInterval =
3214
      QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == downstream->operatorType
L
Liu Jicong 已提交
3215 3216
          ? &((SMergeAlignedIntervalAggOperatorInfo*)downstream->info)->intervalAggOperatorInfo->interval
          : &((SIntervalAggOperatorInfo*)downstream->info)->interval;
3217

3218
  int32_t order = (pPhyFillNode->inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
3219
  int32_t type = convertFillType(pPhyFillNode->mode);
3220

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

3223
  initResultSizeInfo(&pOperator->resultInfo, 4096);
H
Haojun Liao 已提交
3224 3225 3226 3227 3228
  blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
  code = initExprSupp(&pOperator->exprSupp, pExprInfo, pInfo->numOfExpr);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
3229

H
Haojun Liao 已提交
3230 3231
  pInfo->primaryTsCol = ((STargetNode*)pPhyFillNode->pWStartTs)->slotId;
  pInfo->primarySrcSlotId = ((SColumnNode*)((STargetNode*)pPhyFillNode->pWStartTs)->pExpr)->slotId;
3232

3233
  int32_t numOfOutputCols = 0;
H
Haojun Liao 已提交
3234 3235
  code = extractColMatchInfo(pPhyFillNode->pFillExprs, pPhyFillNode->node.pOutputDataBlockDesc, &numOfOutputCols,
                             COL_MATCH_FROM_SLOT_ID, &pInfo->matchInfo);
3236

3237
  code = initFillInfo(pInfo, pExprInfo, pInfo->numOfExpr, pNoFillSupp->pExprInfo, pNoFillSupp->numOfExprs,
3238 3239
                      (SNodeListNode*)pPhyFillNode->pValues, pPhyFillNode->timeRange, pResultInfo->capacity,
                      pTaskInfo->id.str, pInterval, type, order);
3240 3241 3242
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
3243

H
Haojun Liao 已提交
3244
  pInfo->pFinalRes = createOneDataBlock(pInfo->pRes, false);
H
Haojun Liao 已提交
3245 3246
  blockDataEnsureCapacity(pInfo->pFinalRes, pOperator->resultInfo.capacity);

3247 3248 3249 3250 3251
  pInfo->pCondition = pPhyFillNode->node.pConditions;
  pOperator->name = "FillOperator";
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_FILL;
3252
  pOperator->exprSupp.numOfExprs = pInfo->numOfExpr;
3253 3254
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
H
Haojun Liao 已提交
3255

L
Liu Jicong 已提交
3256
  pOperator->fpSet =
3257
      createOperatorFpSet(operatorDummyOpenFn, doFill, NULL, NULL, destroyFillOperatorInfo, NULL);
3258

3259
  code = appendDownstream(pOperator, &downstream, 1);
3260
  return pOperator;
H
Haojun Liao 已提交
3261

L
Liu Jicong 已提交
3262
_error:
H
Haojun Liao 已提交
3263 3264 3265 3266
  if (pInfo != NULL) {
    destroyFillOperatorInfo(pInfo);
  }

wafwerar's avatar
wafwerar 已提交
3267
  taosMemoryFreeClear(pOperator);
H
Haojun Liao 已提交
3268
  return NULL;
3269 3270
}

D
dapan1121 已提交
3271
static SExecTaskInfo* createExecTaskInfo(uint64_t queryId, uint64_t taskId, EOPTR_EXEC_MODEL model, char* dbFName) {
wafwerar's avatar
wafwerar 已提交
3272
  SExecTaskInfo* pTaskInfo = taosMemoryCalloc(1, sizeof(SExecTaskInfo));
3273
  setTaskStatus(pTaskInfo, TASK_NOT_COMPLETED);
H
Haojun Liao 已提交
3274

3275
  pTaskInfo->schemaInfo.dbname = strdup(dbFName);
3276
  pTaskInfo->cost.created = taosGetTimestampMs();
H
Haojun Liao 已提交
3277
  pTaskInfo->id.queryId = queryId;
dengyihao's avatar
dengyihao 已提交
3278
  pTaskInfo->execModel = model;
H
Haojun Liao 已提交
3279

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

3284 3285
  return pTaskInfo;
}
H
Haojun Liao 已提交
3286

H
Haojun Liao 已提交
3287 3288
SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode);

3289
int32_t extractTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, SExecTaskInfo* pTaskInfo) {
3290 3291
  SMetaReader mr = {0};
  metaReaderInit(&mr, pHandle->meta, 0);
3292
  int32_t code = metaGetTableEntryByUid(&mr, pScanNode->uid);
3293
  if (code != TSDB_CODE_SUCCESS) {
L
Liu Jicong 已提交
3294 3295
    qError("failed to get the table meta, uid:0x%" PRIx64 ", suid:0x%" PRIx64 ", %s", pScanNode->uid, pScanNode->suid,
           GET_TASKID(pTaskInfo));
H
Haojun Liao 已提交
3296

D
dapan1121 已提交
3297
    metaReaderClear(&mr);
3298
    return terrno;
D
dapan1121 已提交
3299
  }
3300

3301 3302
  SSchemaInfo* pSchemaInfo = &pTaskInfo->schemaInfo;
  pSchemaInfo->tablename = strdup(mr.me.name);
3303 3304

  if (mr.me.type == TSDB_SUPER_TABLE) {
3305 3306
    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow);
    pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version;
3307
  } else if (mr.me.type == TSDB_CHILD_TABLE) {
3308 3309
    tDecoderClear(&mr.coder);

3310 3311
    tb_uid_t suid = mr.me.ctbEntry.suid;
    metaGetTableEntryByUid(&mr, suid);
3312 3313
    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.stbEntry.schemaRow);
    pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version;
3314
  } else {
3315
    pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.ntbEntry.schemaRow);
3316
  }
3317 3318

  metaReaderClear(&mr);
3319

H
Haojun Liao 已提交
3320 3321 3322 3323 3324
  pSchemaInfo->qsw = extractQueriedColumnSchema(pScanNode);
  return TSDB_CODE_SUCCESS;
}

SSchemaWrapper* extractQueriedColumnSchema(SScanPhysiNode* pScanNode) {
3325 3326 3327
  int32_t numOfCols = LIST_LENGTH(pScanNode->pScanCols);
  int32_t numOfTags = LIST_LENGTH(pScanNode->pScanPseudoCols);

3328
  SSchemaWrapper* pqSw = taosMemoryCalloc(1, sizeof(SSchemaWrapper));
3329
  pqSw->pSchema = taosMemoryCalloc(numOfCols + numOfTags, sizeof(SSchema));
3330

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

H
Haojun Liao 已提交
3335 3336 3337
    SSchema* pSchema = &pqSw->pSchema[pqSw->nCols++];
    pSchema->colId = pColNode->colId;
    pSchema->type = pColNode->node.resType.type;
H
Haojun Liao 已提交
3338 3339
    pSchema->bytes = pColNode->node.resType.bytes;
    tstrncpy(pSchema->name, pColNode->colName, tListLen(pSchema->name));
3340 3341
  }

3342
  // this the tags and pseudo function columns, we only keep the tag columns
3343
  for (int32_t i = 0; i < numOfTags; ++i) {
3344 3345 3346 3347 3348 3349 3350 3351 3352
    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 已提交
3353
      pSchema->bytes = pColNode->node.resType.bytes;
H
Haojun Liao 已提交
3354
      tstrncpy(pSchema->name, pColNode->colName, tListLen(pSchema->name));
3355 3356 3357
    }
  }

H
Haojun Liao 已提交
3358
  return pqSw;
3359 3360
}

3361 3362
static void cleanupTableSchemaInfo(SSchemaInfo* pSchemaInfo) {
  taosMemoryFreeClear(pSchemaInfo->dbname);
3363
  taosMemoryFreeClear(pSchemaInfo->tablename);
3364 3365
  tDeleteSSchemaWrapper(pSchemaInfo->sw);
  tDeleteSSchemaWrapper(pSchemaInfo->qsw);
3366 3367
}

3368
static void cleanupStreamInfo(SStreamTaskInfo* pStreamInfo) { tDeleteSSchemaWrapper(pStreamInfo->schema); }
3369

H
Haojun Liao 已提交
3370
static int32_t sortTableGroup(STableListInfo* pTableListInfo) {
wmmhello's avatar
wmmhello 已提交
3371
  taosArrayClear(pTableListInfo->pGroupList);
H
Haojun Liao 已提交
3372
  SArray* sortSupport = taosArrayInit(16, sizeof(uint64_t));
3373
  if (sortSupport == NULL) return TSDB_CODE_OUT_OF_MEMORY;
wmmhello's avatar
wmmhello 已提交
3374 3375
  for (int32_t i = 0; i < taosArrayGetSize(pTableListInfo->pTableList); i++) {
    STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
3376
    uint64_t*      groupId = taosHashGet(pTableListInfo->map, &info->uid, sizeof(uint64_t));
wmmhello's avatar
wmmhello 已提交
3377 3378

    int32_t index = taosArraySearchIdx(sortSupport, groupId, compareUint64Val, TD_EQ);
3379 3380 3381 3382
    if (index == -1) {
      void*   p = taosArraySearch(sortSupport, groupId, compareUint64Val, TD_GT);
      SArray* tGroup = taosArrayInit(8, sizeof(STableKeyInfo));
      if (tGroup == NULL) {
wmmhello's avatar
wmmhello 已提交
3383 3384 3385
        taosArrayDestroy(sortSupport);
        return TSDB_CODE_OUT_OF_MEMORY;
      }
3386
      if (taosArrayPush(tGroup, info) == NULL) {
wmmhello's avatar
wmmhello 已提交
3387 3388 3389 3390
        qError("taos push info array error");
        taosArrayDestroy(sortSupport);
        return TSDB_CODE_QRY_APP_ERROR;
      }
3391
      if (p == NULL) {
wmmhello's avatar
wmmhello 已提交
3392
        if (taosArrayPush(sortSupport, groupId) == NULL) {
wmmhello's avatar
wmmhello 已提交
3393 3394 3395 3396
          qError("taos push support array error");
          taosArrayDestroy(sortSupport);
          return TSDB_CODE_QRY_APP_ERROR;
        }
wmmhello's avatar
wmmhello 已提交
3397
        if (taosArrayPush(pTableListInfo->pGroupList, &tGroup) == NULL) {
wmmhello's avatar
wmmhello 已提交
3398 3399 3400 3401
          qError("taos push group array error");
          taosArrayDestroy(sortSupport);
          return TSDB_CODE_QRY_APP_ERROR;
        }
3402
      } else {
wmmhello's avatar
wmmhello 已提交
3403
        int32_t pos = TARRAY_ELEM_IDX(sortSupport, p);
3404
        if (taosArrayInsert(sortSupport, pos, groupId) == NULL) {
wmmhello's avatar
wmmhello 已提交
3405 3406 3407 3408
          qError("taos insert support array error");
          taosArrayDestroy(sortSupport);
          return TSDB_CODE_QRY_APP_ERROR;
        }
3409
        if (taosArrayInsert(pTableListInfo->pGroupList, pos, &tGroup) == NULL) {
wmmhello's avatar
wmmhello 已提交
3410 3411 3412 3413 3414
          qError("taos insert group array error");
          taosArrayDestroy(sortSupport);
          return TSDB_CODE_QRY_APP_ERROR;
        }
      }
3415
    } else {
wmmhello's avatar
wmmhello 已提交
3416
      SArray* tGroup = (SArray*)taosArrayGetP(pTableListInfo->pGroupList, index);
3417
      if (taosArrayPush(tGroup, info) == NULL) {
wmmhello's avatar
wmmhello 已提交
3418 3419 3420 3421 3422 3423 3424 3425 3426 3427
        qError("taos push uid array error");
        taosArrayDestroy(sortSupport);
        return TSDB_CODE_QRY_APP_ERROR;
      }
    }
  }
  taosArrayDestroy(sortSupport);
  return TDB_CODE_SUCCESS;
}

3428
bool groupbyTbname(SNodeList* pGroupList) {
3429
  bool bytbname = false;
H
Haojun Liao 已提交
3430
  if (LIST_LENGTH(pGroupList) > 0) {
3431 3432 3433 3434 3435 3436 3437 3438 3439 3440
    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;
}

wmmhello's avatar
wmmhello 已提交
3441 3442
int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, SNodeList* group) {
  if (group == NULL) {
wmmhello's avatar
wmmhello 已提交
3443 3444 3445 3446 3447 3448 3449 3450
    return TDB_CODE_SUCCESS;
  }

  pTableListInfo->map = taosHashInit(32, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
  if (pTableListInfo->map == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

3451 3452
  bool assignUid = groupbyTbname(group);

H
Haojun Liao 已提交
3453
  size_t numOfTables = taosArrayGetSize(pTableListInfo->pTableList);
3454

H
Haojun Liao 已提交
3455 3456 3457
  if (assignUid) {
    for (int32_t i = 0; i < numOfTables; i++) {
      STableKeyInfo* info = taosArrayGet(pTableListInfo->pTableList, i);
3458
      info->groupId = info->uid;
H
Haojun Liao 已提交
3459 3460 3461 3462 3463 3464
      taosHashPut(pTableListInfo->map, &(info->uid), sizeof(uint64_t), &info->groupId, sizeof(uint64_t));
    }
  } else {
    int32_t code = getColInfoResultForGroupby(pHandle->meta, group, pTableListInfo);
    if (code != TSDB_CODE_SUCCESS) {
      return code;
wmmhello's avatar
wmmhello 已提交
3465 3466
    }
  }
3467

3468
  if (pTableListInfo->needSortTableByGroupId) {
H
Haojun Liao 已提交
3469
    return sortTableGroup(pTableListInfo);
3470 3471
  }

wmmhello's avatar
wmmhello 已提交
3472 3473 3474
  return TDB_CODE_SUCCESS;
}

3475 3476 3477 3478 3479 3480 3481 3482 3483 3484 3485 3486 3487 3488 3489 3490 3491
static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pCond) {
  memset(pCond, 0, sizeof(SQueryTableDataCond));

  pCond->order = TSDB_ORDER_ASC;
  pCond->numOfCols = 1;
  pCond->colList = taosMemoryCalloc(1, sizeof(SColumnInfo));
  if (pCond->colList == NULL) {
    terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
    return terrno;
  }

  pCond->colList->colId = 1;
  pCond->colList->type = TSDB_DATA_TYPE_TIMESTAMP;
  pCond->colList->bytes = sizeof(TSKEY);

  pCond->twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
  pCond->suid = uid;
3492
  pCond->type = TIMEWINDOW_RANGE_CONTAINED;
3493
  pCond->startVersion = -1;
L
Liu Jicong 已提交
3494
  pCond->endVersion = -1;
3495 3496 3497 3498

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
3499
SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle,
L
Liu Jicong 已提交
3500 3501
                                  STableListInfo* pTableListInfo, SNode* pTagCond, SNode* pTagIndexCond,
                                  const char* pUser) {
3502 3503
  int32_t type = nodeType(pPhyNode);

X
Xiaoyu Wang 已提交
3504
  if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
3505
    SOperatorInfo* pOperator = NULL;
H
Haojun Liao 已提交
3506
    if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
dengyihao's avatar
dengyihao 已提交
3507
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
H
Haojun Liao 已提交
3508

L
Liu Jicong 已提交
3509 3510 3511
      int32_t code =
          createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort, pHandle,
                                  pTableListInfo, pTagCond, pTagIndexCond, GET_TASKID(pTaskInfo));
3512
      if (code) {
wmmhello's avatar
wmmhello 已提交
3513
        pTaskInfo->code = code;
3514
        qError("failed to createScanTableListInfo, code:%s, %s", tstrerror(code), GET_TASKID(pTaskInfo));
D
dapan1121 已提交
3515 3516
        return NULL;
      }
wmmhello's avatar
wmmhello 已提交
3517

3518
      code = extractTableSchemaInfo(pHandle, &pTableScanNode->scan, pTaskInfo);
S
slzhou 已提交
3519
      if (code) {
3520
        pTaskInfo->code = terrno;
wmmhello's avatar
wmmhello 已提交
3521 3522 3523
        return NULL;
      }

3524
      pOperator = createTableScanOperatorInfo(pTableScanNode, pHandle, pTaskInfo);
3525 3526
      STableScanInfo* pScanInfo = pOperator->info;
      pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
S
slzhou 已提交
3527 3528
    } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
      STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
L
Liu Jicong 已提交
3529 3530 3531
      int32_t                   code =
          createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort, pHandle,
                                  pTableListInfo, pTagCond, pTagIndexCond, GET_TASKID(pTaskInfo));
L
Liu Jicong 已提交
3532
      if (code) {
wmmhello's avatar
wmmhello 已提交
3533
        pTaskInfo->code = code;
H
Haojun Liao 已提交
3534
        qError("failed to createScanTableListInfo, code: %s", tstrerror(code));
wmmhello's avatar
wmmhello 已提交
3535 3536
        return NULL;
      }
3537

3538
      code = extractTableSchemaInfo(pHandle, &pTableScanNode->scan, pTaskInfo);
wmmhello's avatar
wmmhello 已提交
3539 3540 3541 3542
      if (code) {
        pTaskInfo->code = terrno;
        return NULL;
      }
wmmhello's avatar
wmmhello 已提交
3543

3544
      pOperator = createTableMergeScanOperatorInfo(pTableScanNode, pTableListInfo, pHandle, pTaskInfo);
wmmhello's avatar
wmmhello 已提交
3545

3546 3547
      STableScanInfo* pScanInfo = pOperator->info;
      pTaskInfo->cost.pRecoder = &pScanInfo->readRecorder;
H
Haojun Liao 已提交
3548
    } else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == type) {
3549 3550
      pOperator = createExchangeOperatorInfo(pHandle ? pHandle->pMsgCb->clientRpc : NULL, (SExchangePhysiNode*)pPhyNode,
                                             pTaskInfo);
H
Haojun Liao 已提交
3551
    } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN == type) {
5
54liuyao 已提交
3552
      STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
5
54liuyao 已提交
3553
      if (pHandle->vnode) {
L
Liu Jicong 已提交
3554 3555 3556
        int32_t code =
            createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags, pTableScanNode->groupSort,
                                    pHandle, pTableListInfo, pTagCond, pTagIndexCond, GET_TASKID(pTaskInfo));
L
Liu Jicong 已提交
3557
        if (code) {
wmmhello's avatar
wmmhello 已提交
3558
          pTaskInfo->code = code;
H
Haojun Liao 已提交
3559
          qError("failed to createScanTableListInfo, code: %s", tstrerror(code));
wmmhello's avatar
wmmhello 已提交
3560 3561
          return NULL;
        }
L
Liu Jicong 已提交
3562 3563 3564 3565 3566

#ifndef NDEBUG
        int32_t sz = taosArrayGetSize(pTableListInfo->pTableList);
        for (int32_t i = 0; i < sz; i++) {
          STableKeyInfo* pKeyInfo = taosArrayGet(pTableListInfo->pTableList, i);
S
Shengliang Guan 已提交
3567
          qDebug("creating stream task: add table %" PRId64, pKeyInfo->uid);
L
Liu Jicong 已提交
3568 3569
        }
#endif
3570
      }
3571

H
Haojun Liao 已提交
3572
      pTaskInfo->schemaInfo.qsw = extractQueriedColumnSchema(&pTableScanNode->scan);
3573
      pOperator = createStreamScanOperatorInfo(pHandle, pTableScanNode, pTagCond, pTaskInfo);
H
Haojun Liao 已提交
3574
    } else if (QUERY_NODE_PHYSICAL_PLAN_SYSTABLE_SCAN == type) {
L
Liu Jicong 已提交
3575
      SSystemTableScanPhysiNode* pSysScanPhyNode = (SSystemTableScanPhysiNode*)pPhyNode;
3576
      pOperator = createSysTableScanOperatorInfo(pHandle, pSysScanPhyNode, pUser, pTaskInfo);
3577
    } else if (QUERY_NODE_PHYSICAL_PLAN_TAG_SCAN == type) {
X
Xiaoyu Wang 已提交
3578
      STagScanPhysiNode* pScanPhyNode = (STagScanPhysiNode*)pPhyNode;
3579
      int32_t code = getTableList(pHandle->meta, pHandle->vnode, pScanPhyNode, pTagCond, pTagIndexCond, pTableListInfo);
3580
      if (code != TSDB_CODE_SUCCESS) {
3581
        pTaskInfo->code = code;
H
Haojun Liao 已提交
3582
        qError("failed to getTableList, code: %s", tstrerror(code));
3583 3584 3585
        return NULL;
      }

3586
      pOperator = createTagScanOperatorInfo(pHandle, pScanPhyNode, pTableListInfo, pTaskInfo);
3587
    } else if (QUERY_NODE_PHYSICAL_PLAN_BLOCK_DIST_SCAN == type) {
3588
      SBlockDistScanPhysiNode* pBlockNode = (SBlockDistScanPhysiNode*)pPhyNode;
3589 3590 3591
      pTableListInfo->pTableList = taosArrayInit(4, sizeof(STableKeyInfo));

      if (pBlockNode->tableType == TSDB_SUPER_TABLE) {
3592
        int32_t code = vnodeGetAllTableList(pHandle->vnode, pBlockNode->uid, pTableListInfo->pTableList);
3593 3594 3595 3596 3597
        if (code != TSDB_CODE_SUCCESS) {
          pTaskInfo->code = terrno;
          return NULL;
        }
      } else {  // Create one table group.
3598
        STableKeyInfo info = {.uid = pBlockNode->uid, .groupId = 0};
3599 3600 3601 3602
        taosArrayPush(pTableListInfo->pTableList, &info);
      }

      SQueryTableDataCond cond = {0};
L
Liu Jicong 已提交
3603
      int32_t             code = initTableblockDistQueryCond(pBlockNode->suid, &cond);
3604 3605
      if (code != TSDB_CODE_SUCCESS) {
        return NULL;
3606
      }
H
Haojun Liao 已提交
3607 3608 3609

      STsdbReader* pReader = NULL;
      tsdbReaderOpen(pHandle->vnode, &cond, pTableListInfo->pTableList, &pReader, "");
3610 3611
      cleanupQueryTableDataCond(&cond);

3612
      pOperator = createDataBlockInfoScanOperator(pReader, pHandle, cond.suid, pBlockNode, pTaskInfo);
H
Haojun Liao 已提交
3613 3614 3615
    } else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) {
      SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode;

L
Liu Jicong 已提交
3616 3617
      int32_t code = createScanTableListInfo(&pScanNode->scan, pScanNode->pGroupTags, true, pHandle, pTableListInfo,
                                             pTagCond, pTagIndexCond, GET_TASKID(pTaskInfo));
3618 3619 3620 3621
      if (code != TSDB_CODE_SUCCESS) {
        pTaskInfo->code = code;
        return NULL;
      }
3622

3623
      code = extractTableSchemaInfo(pHandle, &pScanNode->scan, pTaskInfo);
3624 3625 3626
      if (code != TSDB_CODE_SUCCESS) {
        pTaskInfo->code = code;
        return NULL;
H
Haojun Liao 已提交
3627 3628
      }

3629
      pOperator = createCacherowsScanOperator(pScanNode, pHandle, pTaskInfo);
3630
    } else if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
3631
      pOperator = createProjectOperatorInfo(NULL, (SProjectPhysiNode*)pPhyNode, pTaskInfo);
H
Haojun Liao 已提交
3632 3633
    } else {
      ASSERT(0);
H
Haojun Liao 已提交
3634
    }
3635 3636 3637 3638 3639

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

3640
    return pOperator;
H
Haojun Liao 已提交
3641 3642
  }

3643 3644
  int32_t num = 0;
  size_t  size = LIST_LENGTH(pPhyNode->pChildren);
H
Haojun Liao 已提交
3645

3646
  SOperatorInfo** ops = taosMemoryCalloc(size, POINTER_BYTES);
dengyihao's avatar
dengyihao 已提交
3647
  for (int32_t i = 0; i < size; ++i) {
3648
    SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, i);
3649
    ops[i] = createOperatorTree(pChildNode, pTaskInfo, pHandle, pTableListInfo, pTagCond, pTagIndexCond, pUser);
3650
    if (ops[i] == NULL) {
H
Haojun Liao 已提交
3651
      taosMemoryFree(ops);
3652 3653
      return NULL;
    }
3654 3655

    ops[i]->resultDataBlockId = pChildNode->pOutputDataBlockDesc->dataBlockId;
3656
  }
H
Haojun Liao 已提交
3657

3658
  SOperatorInfo* pOptr = NULL;
H
Haojun Liao 已提交
3659
  if (QUERY_NODE_PHYSICAL_PLAN_PROJECT == type) {
3660
    pOptr = createProjectOperatorInfo(ops[0], (SProjectPhysiNode*)pPhyNode, pTaskInfo);
3661
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_AGG == type) {
H
Haojun Liao 已提交
3662 3663
    SAggPhysiNode* pAggNode = (SAggPhysiNode*)pPhyNode;
    if (pAggNode->pGroupKeys != NULL) {
H
Haojun Liao 已提交
3664
      pOptr = createGroupOperatorInfo(ops[0], pAggNode, pTaskInfo);
H
Haojun Liao 已提交
3665
    } else {
H
Haojun Liao 已提交
3666
      pOptr = createAggregateOperatorInfo(ops[0], pAggNode, pTaskInfo);
H
Haojun Liao 已提交
3667
    }
3668
  } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type) {
H
Haojun Liao 已提交
3669
    SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
H
Haojun Liao 已提交
3670

H
Haojun Liao 已提交
3671 3672
    bool isStream = (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type);
    pOptr = createIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo, isStream);
3673
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type) {
3674
    pOptr = createStreamIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo);
3675 3676
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) {
    SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode;
3677
    pOptr = createMergeAlignedIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo);
S
shenglian zhou 已提交
3678
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL == type) {
X
Xiaoyu Wang 已提交
3679
    SMergeIntervalPhysiNode* pIntervalPhyNode = (SMergeIntervalPhysiNode*)pPhyNode;
3680
    pOptr = createMergeIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo);
5
54liuyao 已提交
3681
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL == type) {
3682
    int32_t children = 0;
5
54liuyao 已提交
3683 3684
    pOptr = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL == type) {
5
54liuyao 已提交
3685
    int32_t children = pHandle->numOfVgroups;
5
54liuyao 已提交
3686
    pOptr = createStreamFinalIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
H
Haojun Liao 已提交
3687
  } else if (QUERY_NODE_PHYSICAL_PLAN_SORT == type) {
3688
    pOptr = createSortOperatorInfo(ops[0], (SSortPhysiNode*)pPhyNode, pTaskInfo);
S
shenglian zhou 已提交
3689 3690
  } else if (QUERY_NODE_PHYSICAL_PLAN_GROUP_SORT == type) {
    pOptr = createGroupSortOperatorInfo(ops[0], (SGroupSortPhysiNode*)pPhyNode, pTaskInfo);
X
Xiaoyu Wang 已提交
3691
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) {
3692
    SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode;
3693
    pOptr = createMultiwayMergeOperatorInfo(ops, size, pMergePhyNode, pTaskInfo);
3694
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION == type) {
H
Haojun Liao 已提交
3695
    SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
H
Haojun Liao 已提交
3696
    pOptr = createSessionAggOperatorInfo(ops[0], pSessionNode, pTaskInfo);
3697
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION == type) {
3698 3699 3700 3701 3702
    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) {
3703
    int32_t children = pHandle->numOfVgroups;
3704
    pOptr = createStreamFinalSessionAggOperatorInfo(ops[0], pPhyNode, pTaskInfo, children);
H
Haojun Liao 已提交
3705
  } else if (QUERY_NODE_PHYSICAL_PLAN_PARTITION == type) {
3706
    pOptr = createPartitionOperatorInfo(ops[0], (SPartitionPhysiNode*)pPhyNode, pTaskInfo);
3707
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION == type) {
3708
    pOptr = createStreamPartitionOperatorInfo(ops[0], (SStreamPartitionPhysiNode*)pPhyNode, pTaskInfo);
3709
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE == type) {
dengyihao's avatar
dengyihao 已提交
3710
    SStateWinodwPhysiNode* pStateNode = (SStateWinodwPhysiNode*)pPhyNode;
3711
    pOptr = createStatewindowOperatorInfo(ops[0], pStateNode, pTaskInfo);
3712
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE == type) {
5
54liuyao 已提交
3713
    pOptr = createStreamStateAggOperatorInfo(ops[0], pPhyNode, pTaskInfo);
3714
  } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN == type) {
3715
    pOptr = createMergeJoinOperatorInfo(ops, size, (SSortMergeJoinPhysiNode*)pPhyNode, pTaskInfo);
3716
  } else if (QUERY_NODE_PHYSICAL_PLAN_FILL == type) {
H
Haojun Liao 已提交
3717
    pOptr = createFillOperatorInfo(ops[0], (SFillPhysiNode*)pPhyNode, pTaskInfo);
5
54liuyao 已提交
3718 3719
  } else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_FILL == type) {
    pOptr = createStreamFillOperatorInfo(ops[0], (SStreamFillPhysiNode*)pPhyNode, pTaskInfo);
H
Haojun Liao 已提交
3720 3721
  } else if (QUERY_NODE_PHYSICAL_PLAN_INDEF_ROWS_FUNC == type) {
    pOptr = createIndefinitOutputOperatorInfo(ops[0], pPhyNode, pTaskInfo);
3722 3723
  } else if (QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC == type) {
    pOptr = createTimeSliceOperatorInfo(ops[0], pPhyNode, pTaskInfo);
H
Haojun Liao 已提交
3724 3725
  } else {
    ASSERT(0);
H
Haojun Liao 已提交
3726
  }
3727

3728
  taosMemoryFree(ops);
3729 3730 3731 3732
  if (pOptr) {
    pOptr->resultDataBlockId = pPhyNode->pOutputDataBlockDesc->dataBlockId;
  }

3733
  return pOptr;
3734
}
H
Haojun Liao 已提交
3735

L
Liu Jicong 已提交
3736 3737 3738 3739 3740 3741 3742 3743 3744 3745 3746 3747 3748
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 {
3749 3750 3751
    SStreamScanInfo* pInfo = pOperator->info;
    ASSERT(pInfo->pTableScanOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN);
    *ppInfo = pInfo->pTableScanOp->info;
L
Liu Jicong 已提交
3752 3753 3754 3755
    return 0;
  }
}

3756 3757 3758 3759 3760 3761 3762 3763 3764 3765 3766 3767 3768 3769 3770 3771 3772 3773 3774 3775 3776 3777
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;
}

3778
#if 0
L
Liu Jicong 已提交
3779 3780 3781 3782 3783
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;
  }
3784

L
Liu Jicong 已提交
3785 3786 3787 3788
  STableScanPhysiNode* pNode = NULL;
  if (extractTableScanNode(plan->pNode, &pNode) < 0) {
    ASSERT(0);
  }
3789

H
Haojun Liao 已提交
3790
  tsdbReaderClose(pTableScanInfo->dataReader);
3791

L
Liu Jicong 已提交
3792
  STableListInfo info = {0};
H
Haojun Liao 已提交
3793
  pTableScanInfo->dataReader = doCreateDataReader(pNode, pHandle, &info, NULL);
L
Liu Jicong 已提交
3794 3795 3796 3797
  if (pTableScanInfo->dataReader == NULL) {
    ASSERT(0);
    qError("failed to create data reader");
    return TSDB_CODE_QRY_APP_ERROR;
3798
  }
L
Liu Jicong 已提交
3799
  // TODO: set uid and ts to data reader
3800 3801
  return 0;
}
3802
#endif
3803

C
Cary Xu 已提交
3804
int32_t encodeOperator(SOperatorInfo* ops, char** result, int32_t* length, int32_t* nOptrWithVal) {
wmmhello's avatar
wmmhello 已提交
3805
  int32_t code = TDB_CODE_SUCCESS;
3806
  char*   pCurrent = NULL;
wmmhello's avatar
wmmhello 已提交
3807
  int32_t currLength = 0;
3808
  if (ops->fpSet.encodeResultRow) {
C
Cary Xu 已提交
3809
    if (result == NULL || length == NULL || nOptrWithVal == NULL) {
wmmhello's avatar
wmmhello 已提交
3810 3811 3812
      return TSDB_CODE_TSC_INVALID_INPUT;
    }
    code = ops->fpSet.encodeResultRow(ops, &pCurrent, &currLength);
wmmhello's avatar
wmmhello 已提交
3813

3814 3815
    if (code != TDB_CODE_SUCCESS) {
      if (*result != NULL) {
wmmhello's avatar
wmmhello 已提交
3816 3817 3818 3819
        taosMemoryFree(*result);
        *result = NULL;
      }
      return code;
C
Cary Xu 已提交
3820 3821 3822
    } else if (currLength == 0) {
      ASSERT(!pCurrent);
      goto _downstream;
wmmhello's avatar
wmmhello 已提交
3823
    }
wmmhello's avatar
wmmhello 已提交
3824

C
Cary Xu 已提交
3825 3826
    ++(*nOptrWithVal);

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

3829
    if (*result == NULL) {
wmmhello's avatar
wmmhello 已提交
3830
      *result = (char*)taosMemoryCalloc(1, currLength + sizeof(int32_t));
wmmhello's avatar
wmmhello 已提交
3831 3832 3833 3834 3835 3836
      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);
3837
    } else {
wmmhello's avatar
wmmhello 已提交
3838
      int32_t sizePre = *(int32_t*)(*result);
3839
      char*   tmp = (char*)taosMemoryRealloc(*result, sizePre + currLength);
wmmhello's avatar
wmmhello 已提交
3840 3841 3842 3843 3844 3845 3846 3847 3848 3849 3850 3851
      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 已提交
3852 3853
  }

C
Cary Xu 已提交
3854
_downstream:
wmmhello's avatar
wmmhello 已提交
3855
  for (int32_t i = 0; i < ops->numOfDownstream; ++i) {
C
Cary Xu 已提交
3856
    code = encodeOperator(ops->pDownstream[i], result, length, nOptrWithVal);
3857
    if (code != TDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
3858
      return code;
wmmhello's avatar
wmmhello 已提交
3859 3860
    }
  }
wmmhello's avatar
wmmhello 已提交
3861
  return TDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
3862 3863
}

H
Haojun Liao 已提交
3864
int32_t decodeOperator(SOperatorInfo* ops, const char* result, int32_t length) {
wmmhello's avatar
wmmhello 已提交
3865
  int32_t code = TDB_CODE_SUCCESS;
3866 3867
  if (ops->fpSet.decodeResultRow) {
    if (result == NULL) {
wmmhello's avatar
wmmhello 已提交
3868 3869
      return TSDB_CODE_TSC_INVALID_INPUT;
    }
H
Haojun Liao 已提交
3870

3871
    ASSERT(length == *(int32_t*)result);
H
Haojun Liao 已提交
3872 3873

    const char* data = result + sizeof(int32_t);
L
Liu Jicong 已提交
3874
    code = ops->fpSet.decodeResultRow(ops, (char*)data);
3875
    if (code != TDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
3876 3877
      return code;
    }
wmmhello's avatar
wmmhello 已提交
3878

wmmhello's avatar
wmmhello 已提交
3879
    int32_t totalLength = *(int32_t*)result;
3880 3881
    int32_t dataLength = *(int32_t*)data;

3882
    if (totalLength == dataLength + sizeof(int32_t)) {  // the last data
wmmhello's avatar
wmmhello 已提交
3883 3884
      result = NULL;
      length = 0;
3885
    } else {
wmmhello's avatar
wmmhello 已提交
3886 3887 3888 3889
      result += dataLength;
      *(int32_t*)(result) = totalLength - dataLength;
      length = totalLength - dataLength;
    }
wmmhello's avatar
wmmhello 已提交
3890 3891
  }

wmmhello's avatar
wmmhello 已提交
3892 3893
  for (int32_t i = 0; i < ops->numOfDownstream; ++i) {
    code = decodeOperator(ops->pDownstream[i], result, length);
3894
    if (code != TDB_CODE_SUCCESS) {
wmmhello's avatar
wmmhello 已提交
3895
      return code;
wmmhello's avatar
wmmhello 已提交
3896 3897
    }
  }
wmmhello's avatar
wmmhello 已提交
3898
  return TDB_CODE_SUCCESS;
wmmhello's avatar
wmmhello 已提交
3899 3900
}

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

D
dapan1121 已提交
3904
  switch (pNode->type) {
D
dapan1121 已提交
3905 3906 3907 3908 3909 3910
    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 已提交
3911

D
dapan1121 已提交
3912 3913 3914
      *pParam = pInserterParam;
      break;
    }
D
dapan1121 已提交
3915
    case QUERY_NODE_PHYSICAL_PLAN_DELETE: {
3916
      SDeleterParam* pDeleterParam = taosMemoryCalloc(1, sizeof(SDeleterParam));
D
dapan1121 已提交
3917 3918 3919 3920
      if (NULL == pDeleterParam) {
        return TSDB_CODE_OUT_OF_MEMORY;
      }
      int32_t tbNum = taosArrayGetSize(pTask->tableqinfoList.pTableList);
D
dapan1121 已提交
3921
      pDeleterParam->suid = pTask->tableqinfoList.suid;
D
dapan1121 已提交
3922 3923 3924 3925 3926 3927
      pDeleterParam->pUidList = taosArrayInit(tbNum, sizeof(uint64_t));
      if (NULL == pDeleterParam->pUidList) {
        taosMemoryFree(pDeleterParam);
        return TSDB_CODE_OUT_OF_MEMORY;
      }
      for (int32_t i = 0; i < tbNum; ++i) {
3928
        STableKeyInfo* pTable = taosArrayGet(pTask->tableqinfoList.pTableList, i);
D
dapan1121 已提交
3929 3930 3931 3932 3933 3934 3935 3936 3937 3938 3939 3940 3941
        taosArrayPush(pDeleterParam->pUidList, &pTable->uid);
      }

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

  return TSDB_CODE_SUCCESS;
}

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

H
Haojun Liao 已提交
3946
  int32_t code = TSDB_CODE_SUCCESS;
D
dapan1121 已提交
3947
  *pTaskInfo = createExecTaskInfo(queryId, taskId, model, pPlan->dbFName);
H
Haojun Liao 已提交
3948 3949 3950 3951
  if (*pTaskInfo == NULL) {
    code = TSDB_CODE_QRY_OUT_OF_MEMORY;
    goto _complete;
  }
H
Haojun Liao 已提交
3952

H
Haojun Liao 已提交
3953 3954 3955 3956
  if (pHandle && pHandle->pStateBackend) {
    (*pTaskInfo)->streamInfo.pState = pHandle->pStateBackend;
  }

3957
  (*pTaskInfo)->sql = sql;
D
dapan1121 已提交
3958
  sql = NULL;
H
Haojun Liao 已提交
3959

3960
  (*pTaskInfo)->pSubplan = pPlan;
L
Liu Jicong 已提交
3961 3962
  (*pTaskInfo)->pRoot = createOperatorTree(pPlan->pNode, *pTaskInfo, pHandle, &(*pTaskInfo)->tableqinfoList,
                                           pPlan->pTagCond, pPlan->pTagIndexCond, pPlan->user);
L
Liu Jicong 已提交
3963

D
dapan1121 已提交
3964
  if (NULL == (*pTaskInfo)->pRoot) {
3965
    code = (*pTaskInfo)->code;
D
dapan1121 已提交
3966
    goto _complete;
3967 3968
  }

H
Haojun Liao 已提交
3969 3970
  return code;

H
Haojun Liao 已提交
3971
_complete:
D
dapan1121 已提交
3972
  taosMemoryFree(sql);
H
Haojun Liao 已提交
3973
  doDestroyTask(*pTaskInfo);
H
Haojun Liao 已提交
3974 3975
  terrno = code;
  return code;
H
Haojun Liao 已提交
3976 3977
}

3978
void doDestroyTableList(STableListInfo* pTableqinfoList) {
wmmhello's avatar
wmmhello 已提交
3979 3980
  taosArrayDestroy(pTableqinfoList->pTableList);
  taosHashCleanup(pTableqinfoList->map);
3981 3982
  if (pTableqinfoList->needSortTableByGroupId) {
    for (int32_t i = 0; i < taosArrayGetSize(pTableqinfoList->pGroupList); i++) {
wmmhello's avatar
wmmhello 已提交
3983
      SArray* tmp = taosArrayGetP(pTableqinfoList->pGroupList, i);
3984 3985 3986
      if (tmp == pTableqinfoList->pTableList) {
        continue;
      }
wmmhello's avatar
wmmhello 已提交
3987 3988 3989 3990
      taosArrayDestroy(tmp);
    }
  }
  taosArrayDestroy(pTableqinfoList->pGroupList);
3991

wmmhello's avatar
wmmhello 已提交
3992 3993
  pTableqinfoList->pTableList = NULL;
  pTableqinfoList->map = NULL;
3994 3995
}

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

wmmhello's avatar
wmmhello 已提交
3999
  doDestroyTableList(&pTaskInfo->tableqinfoList);
H
Haojun Liao 已提交
4000
  destroyOperatorInfo(pTaskInfo->pRoot);
4001
  cleanupTableSchemaInfo(&pTaskInfo->schemaInfo);
4002
  cleanupStreamInfo(&pTaskInfo->streamInfo);
4003

D
dapan1121 已提交
4004
  if (!pTaskInfo->localFetch.localExec) {
D
dapan1121 已提交
4005 4006
    nodesDestroyNode((SNode*)pTaskInfo->pSubplan);
  }
4007

wafwerar's avatar
wafwerar 已提交
4008 4009 4010
  taosMemoryFreeClear(pTaskInfo->sql);
  taosMemoryFreeClear(pTaskInfo->id.str);
  taosMemoryFreeClear(pTaskInfo);
4011 4012 4013 4014
}

static int64_t getQuerySupportBufSize(size_t numOfTables) {
  size_t s1 = sizeof(STableQueryInfo);
L
Liu Jicong 已提交
4015 4016
  //  size_t s3 = sizeof(STableCheckInfo);  buffer consumption in tsdb
  return (int64_t)(s1 * 1.5 * numOfTables);
4017 4018 4019 4020 4021 4022 4023
}

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 已提交
4024
    while (1) {
4025 4026 4027 4028 4029 4030 4031 4032 4033 4034 4035 4036 4037 4038 4039 4040 4041 4042 4043 4044 4045 4046 4047 4048 4049 4050
      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 已提交
4051

H
Haojun Liao 已提交
4052
int32_t getOperatorExplainExecInfo(SOperatorInfo* operatorInfo, SArray* pExecInfoList) {
4053
  SExplainExecInfo  execInfo = {0};
H
Haojun Liao 已提交
4054
  SExplainExecInfo* pExplainInfo = taosArrayPush(pExecInfoList, &execInfo);
4055

H
Haojun Liao 已提交
4056 4057 4058 4059 4060
  pExplainInfo->numOfRows = operatorInfo->resultInfo.totalRows;
  pExplainInfo->startupCost = operatorInfo->cost.openCost;
  pExplainInfo->totalCost = operatorInfo->cost.totalCost;
  pExplainInfo->verboseLen = 0;
  pExplainInfo->verboseInfo = NULL;
D
dapan1121 已提交
4061

4062
  if (operatorInfo->fpSet.getExplainFn) {
4063 4064
    int32_t code =
        operatorInfo->fpSet.getExplainFn(operatorInfo, &pExplainInfo->verboseInfo, &pExplainInfo->verboseLen);
D
dapan1121 已提交
4065
    if (code) {
4066
      qError("%s operator getExplainFn failed, code:%s", GET_TASKID(operatorInfo->pTaskInfo), tstrerror(code));
D
dapan1121 已提交
4067 4068 4069
      return code;
    }
  }
dengyihao's avatar
dengyihao 已提交
4070

D
dapan1121 已提交
4071
  int32_t code = 0;
D
dapan1121 已提交
4072
  for (int32_t i = 0; i < operatorInfo->numOfDownstream; ++i) {
H
Haojun Liao 已提交
4073 4074
    code = getOperatorExplainExecInfo(operatorInfo->pDownstream[i], pExecInfoList);
    if (code != TSDB_CODE_SUCCESS) {
4075
      //      taosMemoryFreeClear(*pRes);
D
dapan1121 已提交
4076 4077 4078 4079 4080
      return TSDB_CODE_QRY_OUT_OF_MEMORY;
    }
  }

  return TSDB_CODE_SUCCESS;
D
dapan1121 已提交
4081
}
5
54liuyao 已提交
4082

4083 4084
int32_t setOutputBuf(SStreamState* pState, STimeWindow* win, SResultRow** pResult, int64_t tableGroupId,
                     SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup) {
4085 4086 4087 4088 4089 4090
  SWinKey key = {
      .ts = win->skey,
      .groupId = tableGroupId,
  };
  char*   value = NULL;
  int32_t size = pAggSup->resultRowSize;
5
54liuyao 已提交
4091

4092
  if (streamStateAddIfNotExist(pState, &key, (void**)&value, &size) < 0) {
4093 4094 4095 4096 4097 4098 4099 4100 4101 4102
    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;
}

4103 4104
int32_t releaseOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult) {
  streamStateReleaseBuf(pState, pKey, pResult);
4105 4106 4107
  return TSDB_CODE_SUCCESS;
}

4108 4109
int32_t saveOutputBuf(SStreamState* pState, SWinKey* pKey, SResultRow* pResult, int32_t resSize) {
  streamStatePut(pState, pKey, pResult, resSize);
4110 4111 4112
  return TSDB_CODE_SUCCESS;
}

4113
int32_t buildDataBlockFromGroupRes(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock, SExprSupp* pSup,
4114
                                   SGroupResInfo* pGroupResInfo) {
4115
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
4116 4117 4118 4119 4120 4121 4122 4123 4124 4125 4126 4127 4128 4129 4130
  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 = {
            .ts = *(TSKEY*)pPos->key,
            .groupId = pPos->groupId,
    };
4131
    int32_t code = streamStateGet(pState, &key, &pVal, &size);
4132 4133 4134 4135 4136 4137
    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;
4138
      releaseOutputBuf(pState, &key, pRow);
4139 4140 4141 4142 4143
      continue;
    }

    if (pBlock->info.groupId == 0) {
      pBlock->info.groupId = pPos->groupId;
4144 4145 4146 4147
      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 已提交
4148 4149
      } else {
        pBlock->info.parTbName[0] = 0;
4150
      }
4151 4152 4153
    } else {
      // current value belongs to different group, it can't be packed into one datablock
      if (pBlock->info.groupId != pPos->groupId) {
4154
        releaseOutputBuf(pState, &key, pRow);
4155 4156 4157 4158 4159 4160
        break;
      }
    }

    if (pBlock->info.rows + pRow->numOfRows > pBlock->info.capacity) {
      ASSERT(pBlock->info.rows > 0);
4161
      releaseOutputBuf(pState, &key, pRow);
4162 4163 4164 4165 4166 4167 4168 4169 4170 4171
      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) {
4172 4173 4174 4175
        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);
4176 4177 4178 4179 4180 4181 4182 4183 4184 4185 4186 4187 4188
        }
      } 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 已提交
4189

4190
    pBlock->info.rows += pRow->numOfRows;
4191
    releaseOutputBuf(pState, &key, pRow);
4192 4193 4194 4195
  }
  blockDataUpdateTsWindow(pBlock, 0);
  return TSDB_CODE_SUCCESS;
}
5
54liuyao 已提交
4196 4197 4198 4199 4200 4201 4202

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;
}

4203
int32_t buildSessionResultDataBlock(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
5
54liuyao 已提交
4204
                                    SExprSupp* pSup, SGroupResInfo* pGroupResInfo) {
4205
  SExecTaskInfo*  pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
4206 4207 4208 4209 4210 4211 4212 4213 4214 4215 4216 4217 4218
  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);
4219 4220 4221 4222
    if (code == -1) {
      // coverity scan
      continue;
    }
5
54liuyao 已提交
4223 4224 4225 4226 4227 4228 4229 4230 4231 4232 4233
    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;
4234 4235 4236 4237 4238 4239 4240 4241 4242 4243 4244 4245 4246 4247 4248 4249 4250 4251 4252 4253 4254 4255 4256 4257 4258

      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 已提交
4259 4260 4261 4262 4263 4264 4265 4266 4267 4268 4269 4270 4271 4272 4273 4274 4275 4276 4277 4278 4279 4280 4281 4282 4283 4284 4285 4286 4287 4288 4289 4290 4291 4292 4293 4294 4295 4296 4297 4298 4299 4300 4301 4302 4303
    } 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;
4304
}