timewindowoperator.c 189.4 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
#include "executorimpl.h"
X
Xiaoyu Wang 已提交
16
#include "function.h"
5
54liuyao 已提交
17
#include "functionMgt.h"
5
54liuyao 已提交
18
#include "tcompare.h"
L
Liu Jicong 已提交
19
#include "tdatablock.h"
H
Haojun Liao 已提交
20
#include "tfill.h"
21
#include "ttime.h"
22 23 24 25 26 27

typedef enum SResultTsInterpType {
  RESULT_ROW_START_INTERP = 1,
  RESULT_ROW_END_INTERP = 2,
} SResultTsInterpType;

5
54liuyao 已提交
28 29
#define IS_FINAL_OP(op) ((op)->isFinal)

5
54liuyao 已提交
30 31 32 33 34 35 36
typedef struct SWinRes {
  TSKEY    ts;
  uint64_t groupId;
} SWinRes;

typedef struct SPullWindowInfo {
  STimeWindow window;
L
Liu Jicong 已提交
37
  uint64_t    groupId;
5
54liuyao 已提交
38 39
} SPullWindowInfo;

5
54liuyao 已提交
40
static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator);
5
54liuyao 已提交
41

42 43 44 45 46
static int64_t* extractTsCol(SSDataBlock* pBlock, const SIntervalAggOperatorInfo* pInfo);

static SResultRowPosition addToOpenWindowList(SResultRowInfo* pResultRowInfo, const SResultRow* pResult);
static void doCloseWindow(SResultRowInfo* pResultRowInfo, const SIntervalAggOperatorInfo* pInfo, SResultRow* pResult);

H
Haojun Liao 已提交
47 48 49 50 51 52 53 54 55
///*
// * There are two cases to handle:
// *
// * 1. Query range is not set yet (queryRangeSet = 0). we need to set the query range info, including
// * pQueryAttr->lastKey, pQueryAttr->window.skey, and pQueryAttr->eKey.
// * 2. Query range is set and query is in progress. There may be another result with the same query ranges to be
// *    merged during merge stage. In this case, we need the pTableQueryInfo->lastResRows to decide if there
// *    is a previous result generated or not.
// */
56
// static void setIntervalQueryRange(STableQueryInfo* pTableQueryInfo, TSKEY key, STimeWindow* pQRange) {
H
Haojun Liao 已提交
57 58
//  // do nothing
//}
59

X
Xiaoyu Wang 已提交
60
static TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }
61 62 63

static int32_t setTimeWindowOutputBuf(SResultRowInfo* pResultRowInfo, STimeWindow* win, bool masterscan,
                                      SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
64
                                      int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup,
65 66 67 68 69 70 71 72 73 74 75 76
                                      SExecTaskInfo* pTaskInfo) {
  assert(win->skey <= win->ekey);
  SResultRow* pResultRow = doSetResultOutBufByKey(pAggSup->pResultBuf, pResultRowInfo, (char*)&win->skey, TSDB_KEYSIZE,
                                                  masterscan, tableGroupId, pTaskInfo, true, pAggSup);

  if (pResultRow == NULL) {
    *pResult = NULL;
    return TSDB_CODE_SUCCESS;
  }

  // set time window for current result
  pResultRow->win = (*win);
77

78
  *pResult = pResultRow;
79
  setResultRowInitCtx(pResultRow, pCtx, numOfOutput, rowEntryInfoOffset);
80

81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107
  return TSDB_CODE_SUCCESS;
}

static void updateTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pWin, bool includeEndpoint) {
  int64_t* ts = (int64_t*)pColData->pData;
  int32_t  delta = includeEndpoint ? 1 : 0;

  int64_t duration = pWin->ekey - pWin->skey + delta;
  ts[2] = duration;            // set the duration
  ts[3] = pWin->skey;          // window start key
  ts[4] = pWin->ekey + delta;  // window end key
}

static void doKeepTuple(SWindowRowsSup* pRowSup, int64_t ts) {
  pRowSup->win.ekey = ts;
  pRowSup->prevTs = ts;
  pRowSup->numOfRows += 1;
}

static void doKeepNewWindowStartInfo(SWindowRowsSup* pRowSup, const int64_t* tsList, int32_t rowIndex) {
  pRowSup->startRowIndex = rowIndex;
  pRowSup->numOfRows = 0;
  pRowSup->win.skey = tsList[rowIndex];
}

static FORCE_INLINE int32_t getForwardStepsInBlock(int32_t numOfRows, __block_search_fn_t searchFn, TSKEY ekey,
                                                   int16_t pos, int16_t order, int64_t* pData) {
108
  int32_t forwardRows = 0;
109 110 111 112

  if (order == TSDB_ORDER_ASC) {
    int32_t end = searchFn((char*)&pData[pos], numOfRows - pos, ekey, order);
    if (end >= 0) {
113
      forwardRows = end;
114 115

      if (pData[end + pos] == ekey) {
116
        forwardRows += 1;
117 118 119
      }
    }
  } else {
120
    int32_t end = searchFn((char*)&pData[pos], numOfRows - pos, ekey, order);
121
    if (end >= 0) {
122
      forwardRows = end;
123

124
      if (pData[end + pos] == ekey) {
125
        forwardRows += 1;
126 127
      }
    }
X
Xiaoyu Wang 已提交
128 129 130 131 132 133 134 135
    //    int32_t end = searchFn((char*)pData, pos + 1, ekey, order);
    //    if (end >= 0) {
    //      forwardRows = pos - end;
    //
    //      if (pData[end] == ekey) {
    //        forwardRows += 1;
    //      }
    //    }
136 137
  }

138 139
  assert(forwardRows >= 0);
  return forwardRows;
140 141
}

5
54liuyao 已提交
142
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158
  int32_t midPos = -1;
  int32_t numOfRows;

  if (num <= 0) {
    return -1;
  }

  assert(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC);

  TSKEY*  keyList = (TSKEY*)pValue;
  int32_t firstPos = 0;
  int32_t lastPos = num - 1;

  if (order == TSDB_ORDER_DESC) {
    // find the first position which is smaller than the key
    while (1) {
159 160 161 162 163 164 165 166 167 168 169
      if (key >= keyList[firstPos]) return firstPos;
      if (key == keyList[lastPos]) return lastPos;

      if (key < keyList[lastPos]) {
        lastPos += 1;
        if (lastPos >= num) {
          return -1;
        } else {
          return lastPos;
        }
      }
170 171 172 173 174 175

      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;

      if (key < keyList[midPos]) {
        firstPos = midPos + 1;
176 177
      } else if (key > keyList[midPos]) {
        lastPos = midPos - 1;
178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
      } else {
        break;
      }
    }

  } else {
    // find the first position which is bigger than the key
    while (1) {
      if (key <= keyList[firstPos]) return firstPos;
      if (key == keyList[lastPos]) return lastPos;

      if (key > keyList[lastPos]) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }

      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1u) + firstPos;

      if (key < keyList[midPos]) {
        lastPos = midPos - 1;
      } else if (key > keyList[midPos]) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }

  return midPos;
}

X
Xiaoyu Wang 已提交
213 214
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
                                 __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order) {
215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235
  assert(startPos >= 0 && startPos < pDataBlockInfo->rows);

  int32_t num = -1;
  int32_t step = GET_FORWARD_DIRECTION_FACTOR(order);

  if (order == TSDB_ORDER_ASC) {
    if (ekey < pDataBlockInfo->window.ekey && pPrimaryColumn) {
      num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn);
      if (item != NULL) {
        item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
      }
    } else {
      num = pDataBlockInfo->rows - startPos;
      if (item != NULL) {
        item->lastKey = pDataBlockInfo->window.ekey + step;
      }
    }
  } else {  // desc
    if (ekey > pDataBlockInfo->window.skey && pPrimaryColumn) {
      num = getForwardStepsInBlock(pDataBlockInfo->rows, searchFn, ekey, startPos, order, pPrimaryColumn);
      if (item != NULL) {
236
        item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
237 238
      }
    } else {
239
      num = pDataBlockInfo->rows - startPos;
240
      if (item != NULL) {
241
        item->lastKey = pDataBlockInfo->window.ekey + step;
242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272
      }
    }
  }

  assert(num >= 0);
  return num;
}

static void getNextTimeWindow(SInterval* pInterval, int32_t precision, int32_t order, STimeWindow* tw) {
  int32_t factor = GET_FORWARD_DIRECTION_FACTOR(order);
  if (pInterval->intervalUnit != 'n' && pInterval->intervalUnit != 'y') {
    tw->skey += pInterval->sliding * factor;
    tw->ekey = tw->skey + pInterval->interval - 1;
    return;
  }

  int64_t key = tw->skey, interval = pInterval->interval;
  // convert key to second
  key = convertTimePrecision(key, precision, TSDB_TIME_PRECISION_MILLI) / 1000;

  if (pInterval->intervalUnit == 'y') {
    interval *= 12;
  }

  struct tm tm;
  time_t    t = (time_t)key;
  taosLocalTime(&t, &tm);

  int mon = (int)(tm.tm_year * 12 + tm.tm_mon + interval * factor);
  tm.tm_year = mon / 12;
  tm.tm_mon = mon % 12;
wafwerar's avatar
wafwerar 已提交
273
  tw->skey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, precision);
274 275 276 277

  mon = (int)(mon + interval);
  tm.tm_year = mon / 12;
  tm.tm_mon = mon % 12;
wafwerar's avatar
wafwerar 已提交
278
  tw->ekey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, precision);
279 280 281 282

  tw->ekey -= 1;
}

283 284
void doTimeWindowInterpolation(SArray* pPrevValues, SArray* pDataBlock, TSKEY prevTs, int32_t prevRowIndex, TSKEY curTs,
                               int32_t curRowIndex, TSKEY windowKey, int32_t type, SExprSupp* pSup) {
285
  SqlFunctionCtx* pCtx = pSup->pCtx;
286

287
  int32_t index = 1;
288
  for (int32_t k = 0; k < pSup->numOfExprs; ++k) {
H
Haojun Liao 已提交
289
    if (!fmIsIntervalInterpoFunc(pCtx[k].functionId)) {
290 291 292 293
      pCtx[k].start.key = INT64_MIN;
      continue;
    }

X
Xiaoyu Wang 已提交
294
    SFunctParam*     pParam = &pCtx[k].param[0];
295 296
    SColumnInfoData* pColInfo = taosArrayGet(pDataBlock, pParam->pCol->slotId);

297
    ASSERT(pColInfo->info.type == pParam->pCol->type && curTs != windowKey);
298

299
    double v1 = 0, v2 = 0, v = 0;
300
    if (prevRowIndex == -1) {
301
      SGroupKeys* p = taosArrayGet(pPrevValues, index);
302
      GET_TYPED_DATA(v1, double, pColInfo->info.type, p->pData);
303
    } else {
304
      GET_TYPED_DATA(v1, double, pColInfo->info.type, colDataGetData(pColInfo, prevRowIndex));
305 306
    }

307
    GET_TYPED_DATA(v2, double, pColInfo->info.type, colDataGetData(pColInfo, curRowIndex));
308

309
#if 0
310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328
    if (functionId == FUNCTION_INTERP) {
      if (type == RESULT_ROW_START_INTERP) {
        pCtx[k].start.key = prevTs;
        pCtx[k].start.val = v1;

        pCtx[k].end.key = curTs;
        pCtx[k].end.val = v2;

        if (pColInfo->info.type == TSDB_DATA_TYPE_BINARY || pColInfo->info.type == TSDB_DATA_TYPE_NCHAR) {
          if (prevRowIndex == -1) {
            //            pCtx[k].start.ptr = (char*)pRuntimeEnv->prevRow[index];
          } else {
            pCtx[k].start.ptr = (char*)pColInfo->pData + prevRowIndex * pColInfo->info.bytes;
          }

          pCtx[k].end.ptr = (char*)pColInfo->pData + curRowIndex * pColInfo->info.bytes;
        }
      }
    } else if (functionId == FUNCTION_TWA) {
329 330
#endif

X
Xiaoyu Wang 已提交
331 332 333
    SPoint point1 = (SPoint){.key = prevTs, .val = &v1};
    SPoint point2 = (SPoint){.key = curTs, .val = &v2};
    SPoint point = (SPoint){.key = windowKey, .val = &v};
334

X
Xiaoyu Wang 已提交
335
    taosGetLinearInterpolationVal(&point, TSDB_DATA_TYPE_DOUBLE, &point1, &point2, TSDB_DATA_TYPE_DOUBLE);
336

X
Xiaoyu Wang 已提交
337 338 339 340 341 342
    if (type == RESULT_ROW_START_INTERP) {
      pCtx[k].start.key = point.key;
      pCtx[k].start.val = v;
    } else {
      pCtx[k].end.key = point.key;
      pCtx[k].end.val = v;
343
    }
X
Xiaoyu Wang 已提交
344 345 346

    index += 1;
  }
347
#if 0
348
  }
349
#endif
350 351 352 353 354 355 356 357 358 359 360 361 362 363
}

static void setNotInterpoWindowKey(SqlFunctionCtx* pCtx, int32_t numOfOutput, int32_t type) {
  if (type == RESULT_ROW_START_INTERP) {
    for (int32_t k = 0; k < numOfOutput; ++k) {
      pCtx[k].start.key = INT64_MIN;
    }
  } else {
    for (int32_t k = 0; k < numOfOutput; ++k) {
      pCtx[k].end.key = INT64_MIN;
    }
  }
}

364 365
static bool setTimeWindowInterpolationStartTs(SIntervalAggOperatorInfo* pInfo, int32_t pos, SSDataBlock* pBlock,
                                              const TSKEY* tsCols, STimeWindow* win, SExprSupp* pSup) {
X
Xiaoyu Wang 已提交
366
  bool ascQuery = (pInfo->order == TSDB_ORDER_ASC);
367

368
  TSKEY curTs = tsCols[pos];
369 370

  SGroupKeys* pTsKey = taosArrayGet(pInfo->pPrevValues, 0);
X
Xiaoyu Wang 已提交
371
  TSKEY       lastTs = *(int64_t*)pTsKey->pData;
372 373 374 375 376

  // lastTs == INT64_MIN and pos == 0 means this is the first time window, interpolation is not needed.
  // start exactly from this point, no need to do interpolation
  TSKEY key = ascQuery ? win->skey : win->ekey;
  if (key == curTs) {
377
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP);
378 379 380
    return true;
  }

381 382
  // it is the first time window, no need to do interpolation
  if (pTsKey->isNull && pos == 0) {
383
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP);
384 385
  } else {
    TSKEY prevTs = ((pos == 0) ? lastTs : tsCols[pos - 1]);
386 387
    doTimeWindowInterpolation(pInfo->pPrevValues, pBlock->pDataBlock, prevTs, pos - 1, curTs, pos, key,
                              RESULT_ROW_START_INTERP, pSup);
388 389 390 391 392
  }

  return true;
}

393 394 395
static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, SExprSupp* pSup, int32_t endRowIndex,
                                            SArray* pDataBlock, const TSKEY* tsCols, TSKEY blockEkey,
                                            STimeWindow* win) {
396
  int32_t order = pInfo->order;
397 398

  TSKEY actualEndKey = tsCols[endRowIndex];
399
  TSKEY key = (order == TSDB_ORDER_ASC) ? win->ekey : win->skey;
400 401

  // not ended in current data block, do not invoke interpolation
402
  if ((key > blockEkey && (order == TSDB_ORDER_ASC)) || (key < blockEkey && (order == TSDB_ORDER_DESC))) {
403
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP);
404 405 406
    return false;
  }

407
  // there is actual end point of current time window, no interpolation needs
408
  if (key == actualEndKey) {
409
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP);
410 411 412
    return true;
  }

413
  int32_t nextRowIndex = endRowIndex + 1;
414 415 416
  assert(nextRowIndex >= 0);

  TSKEY nextKey = tsCols[nextRowIndex];
417 418
  doTimeWindowInterpolation(pInfo->pPrevValues, pDataBlock, actualEndKey, endRowIndex, nextKey, nextRowIndex, key,
                            RESULT_ROW_END_INTERP, pSup);
419 420 421
  return true;
}

5
54liuyao 已提交
422 423 424 425 426 427 428 429
bool inSlidingWindow(SInterval* pInterval, STimeWindow* pWin, SDataBlockInfo* pBlockInfo) {
  if (pInterval->interval != pInterval->sliding && (pWin->ekey < pBlockInfo->calWin.skey ||
      pWin->skey > pBlockInfo->calWin.ekey) ) {
    return false;
  }
  return true;
}

430
static int32_t getNextQualifiedWindow(SInterval* pInterval, STimeWindow* pNext, SDataBlockInfo* pDataBlockInfo,
5
54liuyao 已提交
431
                                      TSKEY* primaryKeys, int32_t prevPosition, int32_t order) {
X
Xiaoyu Wang 已提交
432
  bool ascQuery = (order == TSDB_ORDER_ASC);
433 434 435 436 437 438 439 440 441 442

  int32_t precision = pInterval->precision;
  getNextTimeWindow(pInterval, precision, order, pNext);

  // next time window is not in current block
  if ((pNext->skey > pDataBlockInfo->window.ekey && order == TSDB_ORDER_ASC) ||
      (pNext->ekey < pDataBlockInfo->window.skey && order == TSDB_ORDER_DESC)) {
    return -1;
  }

5
54liuyao 已提交
443 444 445 446
  if (!inSlidingWindow(pInterval, pNext, pDataBlockInfo) && order == TSDB_ORDER_ASC) {
    return -1;
  }

447
  TSKEY   skey = ascQuery ? pNext->skey : pNext->ekey;
448 449 450 451
  int32_t startPos = 0;

  // tumbling time window query, a special case of sliding time window query
  if (pInterval->sliding == pInterval->interval && prevPosition != -1) {
452
    startPos = prevPosition + 1;
453
  } else {
454
    if ((skey <= pDataBlockInfo->window.skey && ascQuery) || (skey >= pDataBlockInfo->window.ekey && !ascQuery)) {
455 456
      startPos = 0;
    } else {
457
      startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, skey, order);
458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500
    }
  }

  /* interp query with fill should not skip time window */
  //  if (pQueryAttr->pointInterpQuery && pQueryAttr->fillType != TSDB_FILL_NONE) {
  //    return startPos;
  //  }

  /*
   * This time window does not cover any data, try next time window,
   * this case may happen when the time window is too small
   */
  if (primaryKeys == NULL) {
    if (ascQuery) {
      assert(pDataBlockInfo->window.skey <= pNext->ekey);
    } else {
      assert(pDataBlockInfo->window.ekey >= pNext->skey);
    }
  } else {
    if (ascQuery && primaryKeys[startPos] > pNext->ekey) {
      TSKEY next = primaryKeys[startPos];
      if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
        pNext->skey = taosTimeTruncate(next, pInterval, precision);
        pNext->ekey = taosTimeAdd(pNext->skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
      } else {
        pNext->ekey += ((next - pNext->ekey + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
        pNext->skey = pNext->ekey - pInterval->interval + 1;
      }
    } else if ((!ascQuery) && primaryKeys[startPos] < pNext->skey) {
      TSKEY next = primaryKeys[startPos];
      if (pInterval->intervalUnit == 'n' || pInterval->intervalUnit == 'y') {
        pNext->skey = taosTimeTruncate(next, pInterval, precision);
        pNext->ekey = taosTimeAdd(pNext->skey, pInterval->interval, pInterval->intervalUnit, precision) - 1;
      } else {
        pNext->skey -= ((pNext->skey - next + pInterval->sliding - 1) / pInterval->sliding) * pInterval->sliding;
        pNext->ekey = pNext->skey + pInterval->interval - 1;
      }
    }
  }

  return startPos;
}

501 502
static bool isResultRowInterpolated(SResultRow* pResult, SResultTsInterpType type) {
  ASSERT(pResult != NULL && (type == RESULT_ROW_START_INTERP || type == RESULT_ROW_END_INTERP));
503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518
  if (type == RESULT_ROW_START_INTERP) {
    return pResult->startInterp == true;
  } else {
    return pResult->endInterp == true;
  }
}

static void setResultRowInterpo(SResultRow* pResult, SResultTsInterpType type) {
  assert(pResult != NULL && (type == RESULT_ROW_START_INTERP || type == RESULT_ROW_END_INTERP));
  if (type == RESULT_ROW_START_INTERP) {
    pResult->startInterp = true;
  } else {
    pResult->endInterp = true;
  }
}

519 520
static void doWindowBorderInterpolation(SIntervalAggOperatorInfo* pInfo, SSDataBlock* pBlock, SResultRow* pResult,
                                        STimeWindow* win, int32_t startPos, int32_t forwardRows, SExprSupp* pSup) {
521
  if (!pInfo->timeWindowInterpo) {
522 523 524
    return;
  }

525
  ASSERT(pBlock != NULL);
526 527 528 529 530
  if (pBlock->pDataBlock == NULL) {
    //    tscError("pBlock->pDataBlock == NULL");
    return;
  }

531
  SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
532 533

  TSKEY* tsCols = (TSKEY*)(pColInfo->pData);
534
  bool   done = isResultRowInterpolated(pResult, RESULT_ROW_START_INTERP);
535
  if (!done) {  // it is not interpolated, now start to generated the interpolated value
536
    bool interp = setTimeWindowInterpolationStartTs(pInfo, startPos, pBlock, tsCols, win, pSup);
537 538 539 540
    if (interp) {
      setResultRowInterpo(pResult, RESULT_ROW_START_INTERP);
    }
  } else {
541
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP);
542 543 544 545 546 547 548 549
  }

  // point interpolation does not require the end key time window interpolation.
  //  if (pointInterpQuery) {
  //    return;
  //  }

  // interpolation query does not generate the time window end interpolation
550
  done = isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP);
551
  if (!done) {
552
    int32_t endRowIndex = startPos + forwardRows - 1;
553

554
    TSKEY endKey = (pInfo->order == TSDB_ORDER_ASC) ? pBlock->info.window.ekey : pBlock->info.window.skey;
555
    bool  interp = setTimeWindowInterpolationEndTs(pInfo, pSup, endRowIndex, pBlock->pDataBlock, tsCols, endKey, win);
556 557 558 559
    if (interp) {
      setResultRowInterpo(pResult, RESULT_ROW_END_INTERP);
    }
  } else {
560
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP);
561 562 563
  }
}

564 565
static void saveDataBlockLastRow(SArray* pPrevKeys, const SSDataBlock* pBlock, SArray* pCols) {
  if (pBlock->pDataBlock == NULL) {
566 567 568
    return;
  }

569 570 571 572 573 574 575
  size_t num = taosArrayGetSize(pPrevKeys);
  for (int32_t k = 0; k < num; ++k) {
    SColumn* pc = taosArrayGet(pCols, k);

    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, pc->slotId);

    SGroupKeys* pkey = taosArrayGet(pPrevKeys, k);
X
Xiaoyu Wang 已提交
576
    for (int32_t i = pBlock->info.rows - 1; i >= 0; --i) {
577 578 579 580 581 582 583 584 585 586 587 588 589 590
      if (colDataIsNull_s(pColInfo, i)) {
        continue;
      }

      char* val = colDataGetData(pColInfo, i);
      if (IS_VAR_DATA_TYPE(pkey->type)) {
        memcpy(pkey->pData, val, varDataTLen(val));
        ASSERT(varDataTLen(val) <= pkey->bytes);
      } else {
        memcpy(pkey->pData, val, pkey->bytes);
      }

      break;
    }
591 592 593
  }
}

594 595 596 597
static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t numOfExprs, SResultRowInfo* pResultRowInfo,
                                       SSDataBlock* pBlock, int32_t scanFlag, int64_t* tsCols, SResultRowPosition* p) {
  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;

598
  SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
599
  SExprSupp*                pSup = &pOperatorInfo->exprSupp;
600

601
  int32_t  startPos = 0;
602
  int32_t  numOfOutput = pSup->numOfExprs;
603
  uint64_t groupId = pBlock->info.groupId;
604

605
  SResultRow* pResult = NULL;
606

607 608
  while (1) {
    SListNode* pn = tdListGetHead(pResultRowInfo->openWindow);
609

610 611 612 613
    SResultRowPosition* p1 = (SResultRowPosition*)pn->data;
    if (p->pageId == p1->pageId && p->offset == p1->offset) {
      break;
    }
614

615 616
    SResultRow* pr = getResultRowByPos(pInfo->aggSup.pResultBuf, p1);
    ASSERT(pr->offset == p1->offset && pr->pageId == p1->pageId);
617

618
    if (pr->closed) {
X
Xiaoyu Wang 已提交
619 620
      ASSERT(isResultRowInterpolated(pr, RESULT_ROW_START_INTERP) &&
             isResultRowInterpolated(pr, RESULT_ROW_END_INTERP));
621 622 623
      tdListPopHead(pResultRowInfo->openWindow);
      continue;
    }
624

625
    STimeWindow w = pr->win;
626 627
    int32_t     ret = setTimeWindowOutputBuf(pResultRowInfo, &w, (scanFlag == MAIN_SCAN), &pResult, groupId, pSup->pCtx,
                                             numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
628 629 630 631 632 633
    if (ret != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }

    ASSERT(!isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP));

X
Xiaoyu Wang 已提交
634 635
    SGroupKeys* pTsKey = taosArrayGet(pInfo->pPrevValues, 0);
    int64_t     prevTs = *(int64_t*)pTsKey->pData;
636 637
    doTimeWindowInterpolation(pInfo->pPrevValues, pBlock->pDataBlock, prevTs, -1, tsCols[startPos], startPos, w.ekey,
                              RESULT_ROW_END_INTERP, pSup);
638 639

    setResultRowInterpo(pResult, RESULT_ROW_END_INTERP);
640
    setNotInterpoWindowKey(pSup->pCtx, numOfExprs, RESULT_ROW_START_INTERP);
641

642 643
    doApplyFunctions(pTaskInfo, pSup->pCtx, &w, &pInfo->twAggSup.timeWindowData, startPos, 0, tsCols, pBlock->info.rows,
                     numOfExprs, pInfo->order);
644 645 646 647

    if (isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP)) {
      closeResultRow(pr);
      tdListPopHead(pResultRowInfo->openWindow);
X
Xiaoyu Wang 已提交
648
    } else {  // the remains are can not be closed yet.
649
      break;
650
    }
651
  }
652
}
653

5
54liuyao 已提交
654
void printDataBlock(SSDataBlock* pBlock, const char* flag) {
L
Liu Jicong 已提交
655
  if (pBlock == NULL) {
5
54liuyao 已提交
656 657 658
    qDebug("======printDataBlock Block is Null");
    return;
  }
L
Liu Jicong 已提交
659
  char* pBuf = NULL;
5
54liuyao 已提交
660 661
  qDebug("%s", dumpBlockData(pBlock, flag, &pBuf));
  taosMemoryFree(pBuf);
5
54liuyao 已提交
662 663
}

5
54liuyao 已提交
664
typedef int64_t (*__get_value_fn_t)(void* data, int32_t index);
665

X
Xiaoyu Wang 已提交
666 667 668
int32_t binarySearch(void* keyList, int num, TSKEY key, int order, __get_value_fn_t getValuefn) {
  int firstPos = 0, lastPos = num - 1, midPos = -1;
  int numOfRows = 0;
5
54liuyao 已提交
669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714

  if (num <= 0) return -1;
  if (order == TSDB_ORDER_DESC) {
    // find the first position which is smaller or equal than the key
    while (1) {
      if (key >= getValuefn(keyList, lastPos)) return lastPos;
      if (key == getValuefn(keyList, firstPos)) return firstPos;
      if (key < getValuefn(keyList, firstPos)) return firstPos - 1;

      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;

      if (key < getValuefn(keyList, midPos)) {
        lastPos = midPos - 1;
      } else if (key > getValuefn(keyList, midPos)) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }

  } else {
    // find the first position which is bigger or equal than the key
    while (1) {
      if (key <= getValuefn(keyList, firstPos)) return firstPos;
      if (key == getValuefn(keyList, lastPos)) return lastPos;

      if (key > getValuefn(keyList, lastPos)) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }

      numOfRows = lastPos - firstPos + 1;
      midPos = (numOfRows >> 1) + firstPos;

      if (key < getValuefn(keyList, midPos)) {
        lastPos = midPos - 1;
      } else if (key > getValuefn(keyList, midPos)) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
715 716
  }

5
54liuyao 已提交
717 718 719 720
  return midPos;
}

int64_t getReskey(void* data, int32_t index) {
X
Xiaoyu Wang 已提交
721
  SArray*     res = (SArray*)data;
5
54liuyao 已提交
722 723 724 725
  SResKeyPos* pos = taosArrayGetP(res, index);
  return *(int64_t*)pos->key;
}

726
static int32_t saveResult(int64_t ts, int32_t pageId, int32_t offset, uint64_t groupId, SArray* pUpdated) {
5
54liuyao 已提交
727
  int32_t size = taosArrayGetSize(pUpdated);
5
54liuyao 已提交
728
  int32_t index = binarySearch(pUpdated, size, ts, TSDB_ORDER_DESC, getReskey);
5
54liuyao 已提交
729 730 731 732
  if (index == -1) {
    index = 0;
  } else {
    TSKEY resTs = getReskey(pUpdated, index);
5
54liuyao 已提交
733
    if (resTs < ts) {
5
54liuyao 已提交
734 735 736 737 738
      index++;
    } else {
      return TSDB_CODE_SUCCESS;
    }
  }
H
Haojun Liao 已提交
739

5
54liuyao 已提交
740 741 742 743 744
  SResKeyPos* newPos = taosMemoryMalloc(sizeof(SResKeyPos) + sizeof(uint64_t));
  if (newPos == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  newPos->groupId = groupId;
5
54liuyao 已提交
745 746
  newPos->pos = (SResultRowPosition){.pageId = pageId, .offset = offset};
  *(int64_t*)newPos->key = ts;
X
Xiaoyu Wang 已提交
747
  if (taosArrayInsert(pUpdated, index, &newPos) == NULL) {
5
54liuyao 已提交
748 749 750 751 752
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
753 754 755 756
static int32_t saveResultRow(SResultRow* result, uint64_t groupId, SArray* pUpdated) {
  return saveResult(result->win.skey, result->pageId, result->offset, groupId, pUpdated);
}

5
54liuyao 已提交
757 758 759 760 761 762 763 764 765 766 767
static void removeResult(SArray* pUpdated, TSKEY key) {
  int32_t size = taosArrayGetSize(pUpdated);
  int32_t index = binarySearch(pUpdated, size, key, TSDB_ORDER_DESC, getReskey);
  if (index >= 0 && key == getReskey(pUpdated, index)) {
    taosArrayRemove(pUpdated, index);
  }
}

static void removeResults(SArray* pWins, SArray* pUpdated) {
  int32_t size = taosArrayGetSize(pWins);
  for (int32_t i = 0; i < size; i++) {
768 769
    SWinRes* pW = taosArrayGet(pWins, i);
    removeResult(pUpdated, pW->ts);
5
54liuyao 已提交
770 771 772
  }
}

773
int64_t getWinReskey(void* data, int32_t index) {
774
  SArray*  res = (SArray*)data;
775 776 777 778 779 780 781 782 783
  SWinRes* pos = taosArrayGet(res, index);
  return pos->ts;
}

static void removeDeleteResults(SArray* pUpdated, SArray* pDelWins) {
  int32_t upSize = taosArrayGetSize(pUpdated);
  int32_t delSize = taosArrayGetSize(pDelWins);
  for (int32_t i = 0; i < upSize; i++) {
    SResKeyPos* pResKey = taosArrayGetP(pUpdated, i);
784 785
    int64_t     key = *(int64_t*)pResKey->key;
    int32_t     index = binarySearch(pDelWins, delSize, key, TSDB_ORDER_DESC, getWinReskey);
786 787 788 789 790 791
    if (index >= 0 && key == getWinReskey(pDelWins, index)) {
      taosArrayRemove(pDelWins, index);
    }
  }
}

5
54liuyao 已提交
792 793 794 795 796
bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup) {
  ASSERT(pSup->maxTs == INT64_MIN || pSup->maxTs > 0);
  return pSup->maxTs != INT64_MIN && ts < pSup->maxTs - pSup->waterMark;
}

L
Liu Jicong 已提交
797
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup) { return isOverdue(pWin->ekey, pSup); }
5
54liuyao 已提交
798

5
54liuyao 已提交
799
static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pBlock,
800
                            int32_t scanFlag, SArray* pUpdated) {
801
  SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
802

803
  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
804
  SExprSupp*     pSup = &pOperatorInfo->exprSupp;
805

X
Xiaoyu Wang 已提交
806
  int32_t     startPos = 0;
807
  int32_t     numOfOutput = pSup->numOfExprs;
X
Xiaoyu Wang 已提交
808 809 810 811 812
  int64_t*    tsCols = extractTsCol(pBlock, pInfo);
  uint64_t    tableGroupId = pBlock->info.groupId;
  bool        ascScan = (pInfo->order == TSDB_ORDER_ASC);
  TSKEY       ts = getStartTsKey(&pBlock->info.window, tsCols);
  SResultRow* pResult = NULL;
813

814
  STimeWindow win = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->order);
L
Liu Jicong 已提交
815
  int32_t     ret = TSDB_CODE_SUCCESS;
5
54liuyao 已提交
816
  if ((!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) && inSlidingWindow(&pInfo->interval, &win, &pBlock->info)) {
L
Liu Jicong 已提交
817 818
    ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx,
                                 numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
5
54liuyao 已提交
819 820 821
    if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
822

L
Liu Jicong 已提交
823
    if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM && pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
5
54liuyao 已提交
824
      saveResultRow(pResult, tableGroupId, pUpdated);
5
54liuyao 已提交
825
    }
826 827
  }

X
Xiaoyu Wang 已提交
828 829 830
  TSKEY   ekey = ascScan ? win.ekey : win.skey;
  int32_t forwardRows =
      getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
831
  ASSERT(forwardRows > 0);
832 833

  // prev time window not interpolation yet.
834
  if (pInfo->timeWindowInterpo) {
835 836
    SResultRowPosition pos = addToOpenWindowList(pResultRowInfo, pResult);
    doInterpUnclosedTimeWindow(pOperatorInfo, numOfOutput, pResultRowInfo, pBlock, scanFlag, tsCols, &pos);
837 838

    // restore current time window
839 840
    ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx,
                                 numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
841 842 843 844
    if (ret != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }

845
    // window start key interpolation
846
    doWindowBorderInterpolation(pInfo, pBlock, pResult, &win, startPos, forwardRows, pSup);
847
  }
848

5
54liuyao 已提交
849
  if ((!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) && inSlidingWindow(&pInfo->interval, &win, &pBlock->info)) {
5
54liuyao 已提交
850 851 852 853
    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &win, true);
    doApplyFunctions(pTaskInfo, pSup->pCtx, &win, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols,
                     pBlock->info.rows, numOfOutput, pInfo->order);
  }
854 855

  doCloseWindow(pResultRowInfo, pInfo, pResult);
856 857 858

  STimeWindow nextWin = win;
  while (1) {
859
    int32_t prevEndPos = forwardRows - 1 + startPos;
860
    startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, pInfo->order);
861 862 863
    if (startPos < 0) {
      break;
    }
5
54liuyao 已提交
864
    if (pInfo->ignoreExpiredData && isCloseWindow(&nextWin, &pInfo->twAggSup)) {
5
54liuyao 已提交
865 866 867 868 869
      ekey = ascScan ? nextWin.ekey : nextWin.skey;
      forwardRows =
          getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
      continue;
    }
870 871

    // null data, failed to allocate more memory buffer
X
Xiaoyu Wang 已提交
872
    int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
873
                                          pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
874 875 876 877
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }

L
Liu Jicong 已提交
878
    if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM && pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
5
54liuyao 已提交
879
      saveResultRow(pResult, tableGroupId, pUpdated);
880 881
    }

X
Xiaoyu Wang 已提交
882
    ekey = ascScan ? nextWin.ekey : nextWin.skey;
883
    forwardRows =
884
        getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->order);
885 886

    // window start(end) key interpolation
887
    doWindowBorderInterpolation(pInfo, pBlock, pResult, &nextWin, startPos, forwardRows, pSup);
888 889

    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true);
890 891
    doApplyFunctions(pTaskInfo, pSup->pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols,
                     pBlock->info.rows, numOfOutput, pInfo->order);
892
    doCloseWindow(pResultRowInfo, pInfo, pResult);
893 894 895
  }

  if (pInfo->timeWindowInterpo) {
896
    saveDataBlockLastRow(pInfo->pPrevValues, pBlock, pInfo->pInterpCols);
897
  }
898 899 900 901 902 903 904 905 906 907 908 909
}

void doCloseWindow(SResultRowInfo* pResultRowInfo, const SIntervalAggOperatorInfo* pInfo, SResultRow* pResult) {
  // current result is done in computing final results.
  if (pInfo->timeWindowInterpo && isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP)) {
    closeResultRow(pResult);
    tdListPopHead(pResultRowInfo->openWindow);
  }
}

SResultRowPosition addToOpenWindowList(SResultRowInfo* pResultRowInfo, const SResultRow* pResult) {
  SResultRowPosition pos = (SResultRowPosition){.pageId = pResult->pageId, .offset = pResult->offset};
X
Xiaoyu Wang 已提交
910
  SListNode*         pn = tdListGetTail(pResultRowInfo->openWindow);
911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930
  if (pn == NULL) {
    tdListAppend(pResultRowInfo->openWindow, &pos);
    return pos;
  }

  SResultRowPosition* px = (SResultRowPosition*)pn->data;
  if (px->pageId != pos.pageId || px->offset != pos.offset) {
    tdListAppend(pResultRowInfo->openWindow, &pos);
  }

  return pos;
}

int64_t* extractTsCol(SSDataBlock* pBlock, const SIntervalAggOperatorInfo* pInfo) {
  TSKEY* tsCols = NULL;
  if (pBlock->pDataBlock != NULL) {
    SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
    tsCols = (int64_t*)pColDataInfo->pData;

    if (tsCols != NULL) {
5
54liuyao 已提交
931
       blockDataUpdateTsWindow(pBlock, pInfo->primaryTsIndex);
932 933 934 935
    }
  }

  return tsCols;
936 937 938 939 940 941 942
}

static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
  if (OPTR_IS_OPENED(pOperator)) {
    return TSDB_CODE_SUCCESS;
  }

L
Liu Jicong 已提交
943
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
944
  SIntervalAggOperatorInfo* pInfo = pOperator->info;
945
  SExprSupp*                pSup = &pOperator->exprSupp;
946

947 948
  int32_t scanFlag = MAIN_SCAN;

X
Xiaoyu Wang 已提交
949
  int64_t        st = taosGetTimestampUs();
950 951 952
  SOperatorInfo* downstream = pOperator->pDownstream[0];

  while (1) {
953
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
954 955 956 957
    if (pBlock == NULL) {
      break;
    }

958 959
    getTableScanInfo(pOperator, &pInfo->order, &scanFlag);

960
    if (pInfo->scalarSupp.pExprInfo != NULL) {
L
Liu Jicong 已提交
961 962
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
963 964
    }

965
    // the pDataBlock are always the same one, no need to call this again
966
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, scanFlag, true);
H
Haojun Liao 已提交
967
    hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag, NULL);
968 969 970
  }

  closeAllResultRows(&pInfo->binfo.resultRowInfo);
971
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->order);
972
  OPTR_SET_OPENED(pOperator);
973 974

  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
975 976 977
  return TSDB_CODE_SUCCESS;
}

978 979 980 981 982 983 984 985 986 987 988 989
static bool compareVal(const char* v, const SStateKeys* pKey) {
  if (IS_VAR_DATA_TYPE(pKey->type)) {
    if (varDataLen(v) != varDataLen(pKey->pData)) {
      return false;
    } else {
      return strncmp(varDataVal(v), varDataVal(pKey->pData), varDataLen(v)) == 0;
    }
  } else {
    return memcmp(pKey->pData, v, pKey->bytes) == 0;
  }
}

990
static void doStateWindowAggImpl(SOperatorInfo* pOperator, SStateWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {
L
Liu Jicong 已提交
991
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
992
  SExprSupp*     pSup = &pOperator->exprSupp;
993

994
  SColumnInfoData* pStateColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->stateCol.slotId);
995 996 997
  int64_t          gid = pBlock->info.groupId;

  bool    masterScan = true;
998
  int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
999 1000
  int16_t bytes = pStateColInfoData->info.bytes;

1001
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
1002 1003 1004 1005 1006
  TSKEY*           tsList = (TSKEY*)pColInfoData->pData;

  SWindowRowsSup* pRowSup = &pInfo->winSup;
  pRowSup->numOfRows = 0;

1007
  struct SColumnDataAgg* pAgg = NULL;
1008
  for (int32_t j = 0; j < pBlock->info.rows; ++j) {
X
Xiaoyu Wang 已提交
1009
    pAgg = (pBlock->pBlockAgg != NULL) ? pBlock->pBlockAgg[pInfo->stateCol.slotId] : NULL;
1010
    if (colDataIsNull(pStateColInfoData, pBlock->info.rows, j, pAgg)) {
1011 1012 1013 1014 1015 1016
      continue;
    }

    char* val = colDataGetData(pStateColInfoData, j);

    if (!pInfo->hasKey) {
1017 1018 1019 1020 1021 1022 1023
      // todo extract method
      if (IS_VAR_DATA_TYPE(pInfo->stateKey.type)) {
        varDataCopy(pInfo->stateKey.pData, val);
      } else {
        memcpy(pInfo->stateKey.pData, val, bytes);
      }

1024 1025 1026 1027
      pInfo->hasKey = true;

      doKeepNewWindowStartInfo(pRowSup, tsList, j);
      doKeepTuple(pRowSup, tsList[j]);
1028
    } else if (compareVal(val, &pInfo->stateKey)) {
1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039
      doKeepTuple(pRowSup, tsList[j]);
      if (j == 0 && pRowSup->startRowIndex != 0) {
        pRowSup->startRowIndex = 0;
      }
    } else {  // a new state window started
      SResultRow* pResult = NULL;

      // keep the time window for the closed time window.
      STimeWindow window = pRowSup->win;

      pRowSup->win.ekey = pRowSup->win.skey;
1040 1041
      int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx,
                                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1042 1043 1044 1045 1046
      if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
        longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
      }

      updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &window, false);
1047
      doApplyFunctions(pTaskInfo, pSup->pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1048 1049 1050 1051 1052
                       pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);

      // here we start a new session window
      doKeepNewWindowStartInfo(pRowSup, tsList, j);
      doKeepTuple(pRowSup, tsList[j]);
1053 1054 1055 1056 1057 1058 1059

      // todo extract method
      if (IS_VAR_DATA_TYPE(pInfo->stateKey.type)) {
        varDataCopy(pInfo->stateKey.pData, val);
      } else {
        memcpy(pInfo->stateKey.pData, val, bytes);
      }
1060 1061 1062 1063 1064
    }
  }

  SResultRow* pResult = NULL;
  pRowSup->win.ekey = tsList[pBlock->info.rows - 1];
1065 1066
  int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1067 1068 1069 1070 1071
  if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
    longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
  }

  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, false);
1072
  doApplyFunctions(pTaskInfo, pSup->pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1073 1074 1075
                   pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
}

1076
static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
1077 1078 1079 1080 1081
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SStateWindowOperatorInfo* pInfo = pOperator->info;
1082

1083 1084
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
  SExprSupp*     pSup = &pOperator->exprSupp;
1085

1086
  SOptrBasicInfo* pBInfo = &pInfo->binfo;
1087 1088

  if (pOperator->status == OP_RES_TO_RETURN) {
1089
    while (1) {
1090 1091
      doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
      doFilter(pInfo->pCondition, pBInfo->pRes);
1092

1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103
      bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
      if (!hasRemain) {
        doSetOperatorCompleted(pOperator);
        break;
      }

      if (pBInfo->pRes->info.rows > 0) {
        break;
      }
    }
    pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1104
    return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1105 1106
  }

1107
  int32_t order = TSDB_ORDER_ASC;
1108
  int64_t st = taosGetTimestampUs();
1109 1110 1111

  SOperatorInfo* downstream = pOperator->pDownstream[0];
  while (1) {
1112
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1113 1114 1115 1116
    if (pBlock == NULL) {
      break;
    }

1117
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
1118 1119
    blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);

1120 1121 1122
    doStateWindowAggImpl(pOperator, pInfo, pBlock);
  }

X
Xiaoyu Wang 已提交
1123
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
1124

1125 1126 1127
  pOperator->status = OP_RES_TO_RETURN;
  closeAllResultRows(&pBInfo->resultRowInfo);

1128
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
1129
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
1130
  while (1) {
1131 1132
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
    doFilter(pInfo->pCondition, pBInfo->pRes);
1133

1134 1135 1136 1137 1138
    bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
    if (!hasRemain) {
      doSetOperatorCompleted(pOperator);
      break;
    }
1139

1140 1141 1142 1143 1144
    if (pBInfo->pRes->info.rows > 0) {
      break;
    }
  }
  pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1145
  return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1146 1147
}

1148
static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) {
1149
  SIntervalAggOperatorInfo* pInfo = pOperator->info;
L
Liu Jicong 已提交
1150
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
1151 1152 1153 1154 1155 1156 1157 1158

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

  SSDataBlock* pBlock = pInfo->binfo.pRes;

  if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM) {
1159
    return pOperator->fpSet.getStreamResFn(pOperator);
1160 1161 1162 1163 1164 1165 1166
  } else {
    pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
    if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
      return NULL;
    }

    blockDataEnsureCapacity(pBlock, pOperator->resultInfo.capacity);
1167 1168 1169
    while (1) {
      doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
      doFilter(pInfo->pCondition, pBlock);
1170

1171 1172 1173 1174 1175
      bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
      if (!hasRemain) {
        doSetOperatorCompleted(pOperator);
        break;
      }
1176

1177 1178 1179
      if (pBlock->info.rows > 0) {
        break;
      }
1180 1181
    }

1182 1183 1184
    size_t rows = pBlock->info.rows;
    pOperator->resultInfo.totalRows += rows;

X
Xiaoyu Wang 已提交
1185
    return (rows == 0) ? NULL : pBlock;
1186 1187 1188 1189
  }
}

// todo merged with the build group result.
1190
static void finalizeUpdatedResult(int32_t numOfOutput, SDiskbasedBuf* pBuf, SArray* pUpdateList,
1191
                                  int32_t* rowEntryInfoOffset) {
1192 1193 1194 1195 1196 1197 1198
  size_t num = taosArrayGetSize(pUpdateList);

  for (int32_t i = 0; i < num; ++i) {
    SResKeyPos* pPos = taosArrayGetP(pUpdateList, i);

    SFilePage*  bufPage = getBufPage(pBuf, pPos->pos.pageId);
    SResultRow* pRow = (SResultRow*)((char*)bufPage + pPos->pos.offset);
1199

1200
    for (int32_t j = 0; j < numOfOutput; ++j) {
1201
      SResultRowEntryInfo* pEntry = getResultEntryInfo(pRow, j, rowEntryInfoOffset);
1202 1203
      if (pRow->numOfRows < pEntry->numOfRes) {
        pRow->numOfRows = pEntry->numOfRes;
1204 1205 1206 1207 1208 1209
      }
    }

    releaseBufPage(pBuf, bufPage);
  }
}
5
54liuyao 已提交
1210
static void setInverFunction(SqlFunctionCtx* pCtx, int32_t num, EStreamType type) {
L
Liu Jicong 已提交
1211
  for (int i = 0; i < num; i++) {
5
54liuyao 已提交
1212 1213
    if (type == STREAM_INVERT) {
      fmSetInvertFunc(pCtx[i].functionId, &(pCtx[i].fpSet));
L
Liu Jicong 已提交
1214
    } else if (type == STREAM_NORMAL) {
5
54liuyao 已提交
1215 1216 1217 1218
      fmSetNormalFunc(pCtx[i].functionId, &(pCtx[i].fpSet));
    }
  }
}
5
54liuyao 已提交
1219

1220
void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SExprSupp* pSup, int32_t numOfOutput) {
X
Xiaoyu Wang 已提交
1221
  SResultRow*     pResult = getResultRowByPos(pResultBuf, p1);
1222
  SqlFunctionCtx* pCtx = pSup->pCtx;
5
54liuyao 已提交
1223
  for (int32_t i = 0; i < numOfOutput; ++i) {
1224
    pCtx[i].resultInfo = getResultEntryInfo(pResult, i, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
1225 1226 1227 1228 1229 1230 1231 1232 1233
    struct SResultRowEntryInfo* pResInfo = pCtx[i].resultInfo;
    if (fmIsWindowPseudoColumnFunc(pCtx[i].functionId)) {
      continue;
    }
    pResInfo->initialized = false;
    if (pCtx[i].functionId != -1) {
      pCtx[i].fpSet.init(&pCtx[i], pResInfo);
    }
  }
5
54liuyao 已提交
1234 1235 1236
  SFilePage* bufPage = getBufPage(pResultBuf, p1->pageId);
  setBufPageDirty(bufPage, true);
  releaseBufPage(pResultBuf, bufPage);
5
54liuyao 已提交
1237 1238
}

5
54liuyao 已提交
1239
bool doClearWindow(SAggSupporter* pAggSup, SExprSupp* pSup, char* pData, int16_t bytes, uint64_t groupId,
X
Xiaoyu Wang 已提交
1240
                   int32_t numOfOutput) {
1241
  SET_RES_WINDOW_KEY(pAggSup->keyBuf, pData, bytes, groupId);
5
54liuyao 已提交
1242
  SResultRowPosition* p1 =
1243
      (SResultRowPosition*)taosHashGet(pAggSup->pResultRowHashTable, pAggSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
1244 1245
  if (!p1) {
    // window has been closed
5
54liuyao 已提交
1246
    return false;
1247
  }
1248
  doClearWindowImpl(p1, pAggSup->pResultBuf, pSup, numOfOutput);
5
54liuyao 已提交
1249
  return true;
5
54liuyao 已提交
1250 1251
}

1252 1253 1254 1255 1256 1257 1258 1259 1260
bool doDeleteIntervalWindow(SAggSupporter* pAggSup, TSKEY ts, uint64_t groupId) {
  size_t bytes = sizeof(TSKEY);
  SET_RES_WINDOW_KEY(pAggSup->keyBuf, &ts, bytes, groupId);
  SResultRowPosition* p1 =
      (SResultRowPosition*)taosHashGet(pAggSup->pResultRowHashTable, pAggSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
  if (!p1) {
    // window has been closed
    return false;
  }
5
54liuyao 已提交
1261
  // SFilePage* bufPage = getBufPage(pAggSup->pResultBuf, p1->pageId);
1262 1263 1264 1265 1266 1267 1268
  // dBufSetBufPageRecycled(pAggSup->pResultBuf, bufPage);
  taosHashRemove(pAggSup->pResultRowHashTable, pAggSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
  return true;
}

void doDeleteSpecifyIntervalWindow(SAggSupporter* pAggSup, SSDataBlock* pBlock, SArray* pUpWins, SInterval* pInterval) {
  SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
1269
  TSKEY*           tsStarts = (TSKEY*)pStartCol->pData;
1270
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
1271
  uint64_t*        groupIds = (uint64_t*)pGroupCol->pData;
1272 1273 1274
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    SResultRowInfo dumyInfo;
    dumyInfo.cur.pageId = -1;
1275
    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsStarts[i], pInterval, TSDB_ORDER_ASC);
1276 1277 1278 1279 1280 1281 1282 1283
    doDeleteIntervalWindow(pAggSup, win.skey, groupIds[i]);
    if (pUpWins) {
      SWinRes winRes = {.ts = win.skey, .groupId = groupIds[i]};
      taosArrayPush(pUpWins, &winRes);
    }
  }
}

1284 1285
static void doClearWindows(SAggSupporter* pAggSup, SExprSupp* pSup1, SInterval* pInterval, int32_t numOfOutput,
                           SSDataBlock* pBlock, SArray* pUpWins) {
5
54liuyao 已提交
1286
  SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
5
54liuyao 已提交
1287 1288 1289
  TSKEY*           tsCols = (TSKEY*)pTsCol->pData;
  uint64_t*        pGpDatas = NULL;
  if (pBlock->info.type == STREAM_RETRIEVE) {
5
54liuyao 已提交
1290
    SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
L
Liu Jicong 已提交
1291
    pGpDatas = (uint64_t*)pGpCol->pData;
5
54liuyao 已提交
1292
  }
L
Liu Jicong 已提交
1293
  int32_t step = 0;
5
54liuyao 已提交
1294 1295 1296 1297 1298 1299 1300
  int32_t startPos = 0;
  SResultRowInfo dumyInfo;
  dumyInfo.cur.pageId = -1;
  STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsCols[0], pInterval, TSDB_ORDER_ASC);
  while (1) {
    step = getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, win.ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
    uint64_t winGpId = pGpDatas ? pGpDatas[startPos] : pBlock->info.groupId;
1301
    bool     res = doClearWindow(pAggSup, pSup1, (char*)&win.skey, sizeof(TKEY), winGpId, numOfOutput);
5
54liuyao 已提交
1302
    if (pUpWins && res) {
1303 1304
      SWinRes winRes = {.ts = win.skey, .groupId = winGpId};
      taosArrayPush(pUpWins, &winRes);
1305
    }
5
54liuyao 已提交
1306 1307 1308 1309 1310
    int32_t prevEndPos = step - 1 + startPos;
    startPos = getNextQualifiedWindow(pInterval, &win, &pBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC);
    if (startPos < 0) {
      break;
    }
5
54liuyao 已提交
1311 1312
  }
}
1313

5
54liuyao 已提交
1314 1315 1316 1317 1318 1319 1320
static int32_t getAllIntervalWindow(SHashObj* pHashMap, SArray* resWins) {
  void*  pIte = NULL;
  size_t keyLen = 0;
  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
    void*    key = taosHashGetKey(pIte, &keyLen);
    uint64_t groupId = *(uint64_t*)key;
    ASSERT(keyLen == GET_RES_WINDOW_KEY_LEN(sizeof(TSKEY)));
1321
    TSKEY               ts = *(int64_t*)((char*)key + sizeof(uint64_t));
5
54liuyao 已提交
1322
    SResultRowPosition* pPos = (SResultRowPosition*)pIte;
1323
    int32_t             code = saveResult(ts, pPos->pageId, pPos->offset, groupId, resWins);
5
54liuyao 已提交
1324 1325 1326 1327 1328 1329 1330
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  }
  return TSDB_CODE_SUCCESS;
}

1331 1332 1333
static int32_t closeIntervalWindow(SHashObj* pHashMap, STimeWindowAggSupp* pSup, SInterval* pInterval,
                                   SHashObj* pPullDataMap, SArray* closeWins, SArray* pRecyPages,
                                   SDiskbasedBuf* pDiscBuf) {
X
Xiaoyu Wang 已提交
1334
  void*  pIte = NULL;
5
54liuyao 已提交
1335
  size_t keyLen = 0;
X
Xiaoyu Wang 已提交
1336 1337 1338
  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
    void*    key = taosHashGetKey(pIte, &keyLen);
    uint64_t groupId = *(uint64_t*)key;
5
54liuyao 已提交
1339
    ASSERT(keyLen == GET_RES_WINDOW_KEY_LEN(sizeof(TSKEY)));
X
Xiaoyu Wang 已提交
1340
    TSKEY          ts = *(int64_t*)((char*)key + sizeof(uint64_t));
5
54liuyao 已提交
1341 1342
    SResultRowInfo dumyInfo;
    dumyInfo.cur.pageId = -1;
1343
    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, ts, pInterval, TSDB_ORDER_ASC);
L
Liu Jicong 已提交
1344 1345 1346 1347
    SWinRes     winRe = {
            .ts = win.skey,
            .groupId = groupId,
    };
5
54liuyao 已提交
1348
    void* chIds = taosHashGet(pPullDataMap, &winRe, sizeof(SWinRes));
5
54liuyao 已提交
1349
    if (isCloseWindow(&win, pSup)) {
5
54liuyao 已提交
1350
      if (chIds && pPullDataMap) {
L
Liu Jicong 已提交
1351
        SArray* chAy = *(SArray**)chIds;
5
54liuyao 已提交
1352
        int32_t size = taosArrayGetSize(chAy);
S
Shengliang Guan 已提交
1353
        qInfo("window %" PRId64 " wait child size:%d", win.skey, size);
5
54liuyao 已提交
1354
        for (int32_t i = 0; i < size; i++) {
S
Shengliang Guan 已提交
1355
          qInfo("window %" PRId64 " wait chid id:%d", win.skey, *(int32_t*)taosArrayGet(chAy, i));
5
54liuyao 已提交
1356 1357 1358
        }
        continue;
      } else if (pPullDataMap) {
S
Shengliang Guan 已提交
1359
        qInfo("close window %" PRId64, win.skey);
5
54liuyao 已提交
1360
      }
5
54liuyao 已提交
1361 1362 1363 1364 1365 1366
      SResultRowPosition* pPos = (SResultRowPosition*)pIte;
      if (pSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
        int32_t code = saveResult(ts, pPos->pageId, pPos->offset, groupId, closeWins);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
1367 1368 1369
        ASSERT(pRecyPages != NULL);
        taosArrayPush(pRecyPages, &pPos->pageId);
      } else {
5
54liuyao 已提交
1370
        // SFilePage* bufPage = getBufPage(pDiscBuf, pPos->pageId);
1371
        // dBufSetBufPageRecycled(pDiscBuf, bufPage);
5
54liuyao 已提交
1372
      }
5
54liuyao 已提交
1373 1374 1375
      char keyBuf[GET_RES_WINDOW_KEY_LEN(sizeof(TSKEY))];
      SET_RES_WINDOW_KEY(keyBuf, &ts, sizeof(TSKEY), groupId);
      taosHashRemove(pHashMap, keyBuf, keyLen);
5
54liuyao 已提交
1376 1377 1378 1379 1380
    }
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
1381 1382 1383
static void closeChildIntervalWindow(SArray* pChildren, TSKEY maxTs) {
  int32_t size = taosArrayGetSize(pChildren);
  for (int32_t i = 0; i < size; i++) {
L
Liu Jicong 已提交
1384
    SOperatorInfo*                    pChildOp = taosArrayGetP(pChildren, i);
5
54liuyao 已提交
1385 1386 1387
    SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info;
    ASSERT(pChInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);
    pChInfo->twAggSup.maxTs = TMAX(pChInfo->twAggSup.maxTs, maxTs);
1388 1389
    closeIntervalWindow(pChInfo->aggSup.pResultRowHashTable, &pChInfo->twAggSup, &pChInfo->interval, NULL, NULL, NULL,
                        pChInfo->aggSup.pResultBuf);
1390 1391 1392 1393 1394 1395 1396
  }
}

static void freeAllPages(SArray* pageIds, SDiskbasedBuf* pDiskBuf) {
  int32_t size = taosArrayGetSize(pageIds);
  for (int32_t i = 0; i < size; i++) {
    int32_t pageId = *(int32_t*)taosArrayGet(pageIds, i);
5
54liuyao 已提交
1397
    // SFilePage* bufPage = getBufPage(pDiskBuf, pageId);
1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419
    // dBufSetBufPageRecycled(pDiskBuf, bufPage);
  }
  taosArrayClear(pageIds);
}

static void doBuildDeleteResult(SArray* pWins, int32_t* index, SSDataBlock* pBlock) {
  blockDataCleanup(pBlock);
  int32_t size = taosArrayGetSize(pWins);
  if (*index == size) {
    *index = 0;
    taosArrayClear(pWins);
    return;
  }
  blockDataEnsureCapacity(pBlock, size - *index);
  SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, DELETE_GROUPID_COLUMN_INDEX);
  for (int32_t i = *index; i < size; i++) {
    SWinRes* pWin = taosArrayGet(pWins, i);
    colDataAppend(pTsCol, pBlock->info.rows, (const char*)&pWin->ts, false);
    colDataAppend(pGroupCol, pBlock->info.rows, (const char*)&pWin->groupId, false);
    pBlock->info.rows++;
    (*index)++;
5
54liuyao 已提交
1420 1421 1422
  }
}

1423
static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
1424
  SIntervalAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
1425
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
1426 1427

  pInfo->order = TSDB_ORDER_ASC;
1428
  SExprSupp* pSup = &pOperator->exprSupp;
1429 1430 1431 1432 1433 1434

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

  if (pOperator->status == OP_RES_TO_RETURN) {
1435
    doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
1436 1437 1438 1439
    if (pInfo->pDelRes->info.rows > 0) {
      return pInfo->pDelRes;
    }

1440
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
1441
    if (pInfo->binfo.pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) {
1442
      pOperator->status = OP_EXEC_DONE;
1443
      freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
1444 1445 1446 1447 1448 1449
    }
    return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
  }

  SOperatorInfo* downstream = pOperator->pDownstream[0];

1450
  SArray* pUpdated = taosArrayInit(4, POINTER_BYTES);  // SResKeyPos
1451
  while (1) {
1452
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1453 1454 1455
    if (pBlock == NULL) {
      break;
    }
1456
    printDataBlock(pBlock, "single interval recv");
1457

5
54liuyao 已提交
1458
    if (pBlock->info.type == STREAM_CLEAR) {
1459 1460
      doClearWindows(&pInfo->aggSup, &pOperator->exprSupp, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock,
                     NULL);
1461
      qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
5
54liuyao 已提交
1462
      continue;
1463 1464
    }
    if (pBlock->info.type == STREAM_DELETE_DATA) {
1465 1466
      doDeleteSpecifyIntervalWindow(&pInfo->aggSup, pBlock, pInfo->pDelWins, &pInfo->interval);
      continue;
1467
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
1468 1469
      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
      continue;
5
54liuyao 已提交
1470
    }
1471

1472 1473 1474 1475 1476
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }

1477 1478 1479
    // The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the
    // caller. Note that all the time window are not close till now.
    // the pDataBlock are always the same one, no need to call this again
1480
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
1481
    if (pInfo->invertible) {
1482
      setInverFunction(pSup->pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.type);
1483 1484
    }

5
54liuyao 已提交
1485
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
H
Haojun Liao 已提交
1486
    hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, MAIN_SCAN, pUpdated);
1487
  }
1488
  pOperator->status = OP_RES_TO_RETURN;
1489 1490
  closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, NULL, pUpdated,
                      pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
1491

1492
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
1493 1494
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
1495 1496 1497 1498 1499
  removeDeleteResults(pUpdated, pInfo->pDelWins);
  doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
  if (pInfo->pDelRes->info.rows > 0) {
    return pInfo->pDelRes;
  }
1500
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
1501
  printDataBlock(pInfo->binfo.pRes, "single interval");
1502 1503 1504 1505 1506
  return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
}

static void destroyStateWindowOperatorInfo(void* param, int32_t numOfOutput) {
  SStateWindowOperatorInfo* pInfo = (SStateWindowOperatorInfo*)param;
1507
  cleanupBasicInfo(&pInfo->binfo);
1508
  taosMemoryFreeClear(pInfo->stateKey.pData);
1509

D
dapan1121 已提交
1510
  taosMemoryFreeClear(param);
1511 1512 1513
}

void destroyIntervalOperatorInfo(void* param, int32_t numOfOutput) {
1514
  SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)param;
1515
  cleanupBasicInfo(&pInfo->binfo);
1516
  cleanupAggSup(&pInfo->aggSup);
1517
  taosArrayDestroy(pInfo->pRecycledPages);
1518

D
dapan1121 已提交
1519
  taosMemoryFreeClear(param);
1520 1521
}

5
54liuyao 已提交
1522
void destroyStreamFinalIntervalOperatorInfo(void* param, int32_t numOfOutput) {
X
Xiaoyu Wang 已提交
1523
  SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)param;
1524
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
1525
  cleanupAggSup(&pInfo->aggSup);
L
Liu Jicong 已提交
1526
  // it should be empty.
5
54liuyao 已提交
1527 1528 1529
  taosHashCleanup(pInfo->pPullDataMap);
  taosArrayDestroy(pInfo->pPullWins);
  blockDataDestroy(pInfo->pPullDataRes);
1530
  taosArrayDestroy(pInfo->pRecycledPages);
5
54liuyao 已提交
1531

1532 1533 1534 1535
  if (pInfo->pChildren) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, i);
1536
      destroyStreamFinalIntervalOperatorInfo(pChildOp->info, numOfOutput);
1537 1538 1539 1540
      taosMemoryFreeClear(pChildOp->info);
      taosMemoryFreeClear(pChildOp);
    }
  }
1541
  nodesDestroyNode((SNode*)pInfo->pPhyNode);
1542

D
dapan1121 已提交
1543
  taosMemoryFreeClear(param);
5
54liuyao 已提交
1544 1545
}

1546
static bool allInvertible(SqlFunctionCtx* pFCtx, int32_t numOfCols) {
5
54liuyao 已提交
1547 1548 1549 1550 1551 1552 1553 1554
  for (int32_t i = 0; i < numOfCols; i++) {
    if (!fmIsInvertible(pFCtx[i].functionId)) {
      return false;
    }
  }
  return true;
}

1555
static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SIntervalAggOperatorInfo* pInfo) {
1556 1557
  // the primary timestamp column
  bool needed = false;
1558 1559
  pInfo->pInterpCols = taosArrayInit(4, sizeof(SColumn));
  pInfo->pPrevValues = taosArrayInit(4, sizeof(SGroupKeys));
1560

X
Xiaoyu Wang 已提交
1561
  {  // ts column
1562 1563
    SColumn c = {0};
    c.colId = 1;
1564
    c.slotId = pInfo->primaryTsIndex;
1565 1566
    c.type = TSDB_DATA_TYPE_TIMESTAMP;
    c.bytes = sizeof(int64_t);
1567
    taosArrayPush(pInfo->pInterpCols, &c);
1568 1569

    SGroupKeys key = {0};
X
Xiaoyu Wang 已提交
1570 1571 1572 1573
    key.bytes = c.bytes;
    key.type = c.type;
    key.isNull = true;  // to denote no value is assigned yet
    key.pData = taosMemoryCalloc(1, c.bytes);
1574
    taosArrayPush(pInfo->pPrevValues, &key);
1575 1576
  }

X
Xiaoyu Wang 已提交
1577
  for (int32_t i = 0; i < numOfCols; ++i) {
1578 1579
    SExprInfo* pExpr = pCtx[i].pExpr;

H
Haojun Liao 已提交
1580
    if (fmIsIntervalInterpoFunc(pCtx[i].functionId)) {
1581 1582 1583
      SFunctParam* pParam = &pExpr->base.pParam[0];

      SColumn c = *pParam->pCol;
1584
      taosArrayPush(pInfo->pInterpCols, &c);
1585 1586 1587
      needed = true;

      SGroupKeys key = {0};
X
Xiaoyu Wang 已提交
1588 1589
      key.bytes = c.bytes;
      key.type = c.type;
1590
      key.isNull = false;
X
Xiaoyu Wang 已提交
1591
      key.pData = taosMemoryCalloc(1, c.bytes);
1592
      taosArrayPush(pInfo->pPrevValues, &key);
1593 1594 1595 1596 1597 1598
    }
  }

  return needed;
}

1599 1600 1601 1602 1603 1604
void increaseTs(SqlFunctionCtx* pCtx) {
  if (pCtx[0].pExpr->pExpr->_function.pFunctNode->funcType == FUNCTION_TYPE_WSTARTTS) {
    pCtx[0].increase = true;
  }
}

1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626
SSDataBlock* createDeleteBlock() {
  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
  pBlock->info.hasVarCol = false;
  pBlock->info.groupId = 0;
  pBlock->info.rows = 0;
  pBlock->info.type = STREAM_DELETE_RESULT;
  pBlock->info.rowSize = sizeof(TSKEY) + sizeof(uint64_t);

  pBlock->pDataBlock = taosArrayInit(2, sizeof(SColumnInfoData));
  SColumnInfoData infoData = {0};
  infoData.info.type = TSDB_DATA_TYPE_TIMESTAMP;
  infoData.info.bytes = sizeof(TSKEY);
  // window start ts
  taosArrayPush(pBlock->pDataBlock, &infoData);

  infoData.info.type = TSDB_DATA_TYPE_UBIGINT;
  infoData.info.bytes = sizeof(uint64_t);
  taosArrayPush(pBlock->pDataBlock, &infoData);

  return pBlock;
}

1627 1628
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
L
Liu Jicong 已提交
1629 1630
                                          STimeWindowAggSupp* pTwAggSupp, SIntervalPhysiNode* pPhyNode,
                                          SExecTaskInfo* pTaskInfo, bool isStream) {
1631
  SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
L
Liu Jicong 已提交
1632
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1633 1634 1635 1636
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

L
Liu Jicong 已提交
1637 1638 1639
  pInfo->win = pTaskInfo->window;
  pInfo->order = TSDB_ORDER_ASC;
  pInfo->interval = *pInterval;
L
Liu Jicong 已提交
1640
  pInfo->execModel = pTaskInfo->execModel;
L
Liu Jicong 已提交
1641
  pInfo->twAggSup = *pTwAggSupp;
5
54liuyao 已提交
1642
  pInfo->ignoreExpiredData = pPhyNode->window.igExpired;
1643
  pInfo->pCondition = pPhyNode->window.node.pConditions;
1644 1645 1646 1647

  if (pPhyNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pPhyNode->window.pExprs, NULL, &numOfScalar);
L
Liu Jicong 已提交
1648
    int32_t    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
1649 1650 1651 1652
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }
1653

1654 1655
  pInfo->primaryTsIndex = primaryTsSlotId;

1656 1657
  SExprSupp* pSup = &pOperator->exprSupp;

1658 1659
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  initResultSizeInfo(pOperator, 4096);
1660

1661 1662 1663
  int32_t code = initAggInfo(pSup, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
  initBasicInfo(&pInfo->binfo, pResBlock);

1664 1665
  if (isStream) {
    ASSERT(numOfCols > 0);
1666
    increaseTs(pSup->pCtx);
1667
  }
1668

1669
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);
1670

1671
  pInfo->invertible = allInvertible(pSup->pCtx, numOfCols);
X
Xiaoyu Wang 已提交
1672
  pInfo->invertible = false;  // Todo(liuyao): Dependent TSDB API
1673

1674
  pInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, numOfCols, pInfo);
1675 1676
  if (pInfo->timeWindowInterpo) {
    pInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition));
H
Haojun Liao 已提交
1677 1678 1679
    if (pInfo->binfo.resultRowInfo.openWindow == NULL) {
      goto _error;
    }
1680
  }
1681 1682 1683
  pInfo->pRecycledPages = taosArrayInit(4, sizeof(int32_t));
  pInfo->pDelWins = taosArrayInit(4, sizeof(SWinRes));
  pInfo->delIndex = 0;
5
54liuyao 已提交
1684
  // pInfo->pDelRes = createPullDataBlock(); todo(liuyao) for delete
1685 1686
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
1687

1688
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
1689

X
Xiaoyu Wang 已提交
1690 1691 1692 1693
  pOperator->name = "TimeIntervalAggOperator";
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL;
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
1694
  pOperator->exprSupp.pExprInfo = pExprInfo;
X
Xiaoyu Wang 已提交
1695
  pOperator->pTaskInfo = pTaskInfo;
1696
  pOperator->exprSupp.numOfExprs = numOfCols;
X
Xiaoyu Wang 已提交
1697
  pOperator->info = pInfo;
1698 1699 1700 1701 1702 1703 1704 1705 1706 1707 1708

  pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, doStreamIntervalAgg, NULL,
                                         destroyIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);

  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

L
Liu Jicong 已提交
1709
_error:
1710 1711 1712 1713 1714 1715 1716 1717 1718
  destroyIntervalOperatorInfo(pInfo, numOfCols);
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}

SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                                SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
wmmhello's avatar
wmmhello 已提交
1719
                                                STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo) {
1720
  SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
L
Liu Jicong 已提交
1721
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1722 1723 1724 1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

  pInfo->order = TSDB_ORDER_ASC;
  pInfo->interval = *pInterval;
  pInfo->execModel = OPTR_EXEC_MODEL_STREAM;
  pInfo->win = pTaskInfo->window;
  pInfo->twAggSup = *pTwAggSupp;
  pInfo->primaryTsIndex = primaryTsSlotId;

  int32_t numOfRows = 4096;
  size_t  keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;

  initResultSizeInfo(pOperator, numOfRows);
1737 1738
  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
  initBasicInfo(&pInfo->binfo, pResBlock);
1739 1740
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);

wmmhello's avatar
wmmhello 已提交
1741
  if (code != TSDB_CODE_SUCCESS) {
1742 1743 1744
    goto _error;
  }

1745
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
1746 1747

  pOperator->name = "StreamTimeIntervalAggOperator";
X
Xiaoyu Wang 已提交
1748
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL;
1749
  pOperator->blocking = true;
1750
  pOperator->status = OP_NOT_OPENED;
1751
  pOperator->exprSupp.pExprInfo = pExprInfo;
1752
  pOperator->pTaskInfo = pTaskInfo;
1753
  pOperator->exprSupp.numOfExprs = numOfCols;
1754 1755 1756 1757 1758 1759 1760 1761 1762 1763 1764 1765
  pOperator->info = pInfo;

  pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doStreamIntervalAgg, doStreamIntervalAgg, NULL,
                                         destroyIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);

  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

L
Liu Jicong 已提交
1766
_error:
1767 1768 1769 1770 1771 1772 1773 1774 1775 1776
  destroyIntervalOperatorInfo(pInfo, numOfCols);
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}

// todo handle multiple tables cases.
static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperatorInfo* pInfo, SSDataBlock* pBlock) {
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1777
  SExprSupp*     pSup = &pOperator->exprSupp;
1778

1779
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
1780 1781

  bool    masterScan = true;
1782
  int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
1783 1784 1785 1786 1787 1788 1789 1790 1791 1792 1793 1794 1795 1796 1797 1798 1799 1800 1801 1802 1803 1804 1805 1806 1807 1808 1809 1810 1811 1812 1813
  int64_t gid = pBlock->info.groupId;

  int64_t gap = pInfo->gap;

  if (!pInfo->reptScan) {
    pInfo->reptScan = true;
    pInfo->winSup.prevTs = INT64_MIN;
  }

  SWindowRowsSup* pRowSup = &pInfo->winSup;
  pRowSup->numOfRows = 0;

  // In case of ascending or descending order scan data, only one time window needs to be kepted for each table.
  TSKEY* tsList = (TSKEY*)pColInfoData->pData;
  for (int32_t j = 0; j < pBlock->info.rows; ++j) {
    if (pInfo->winSup.prevTs == INT64_MIN) {
      doKeepNewWindowStartInfo(pRowSup, tsList, j);
      doKeepTuple(pRowSup, tsList[j]);
    } else if (tsList[j] - pRowSup->prevTs <= gap && (tsList[j] - pRowSup->prevTs) >= 0) {
      // The gap is less than the threshold, so it belongs to current session window that has been opened already.
      doKeepTuple(pRowSup, tsList[j]);
      if (j == 0 && pRowSup->startRowIndex != 0) {
        pRowSup->startRowIndex = 0;
      }
    } else {  // start a new session window
      SResultRow* pResult = NULL;

      // keep the time window for the closed time window.
      STimeWindow window = pRowSup->win;

      pRowSup->win.ekey = pRowSup->win.skey;
1814 1815
      int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx,
                                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1816 1817 1818 1819 1820 1821
      if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
        longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
      }

      // pInfo->numOfRows data belong to the current session window
      updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &window, false);
1822
      doApplyFunctions(pTaskInfo, pSup->pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1823 1824 1825 1826 1827 1828 1829 1830 1831 1832
                       pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);

      // here we start a new session window
      doKeepNewWindowStartInfo(pRowSup, tsList, j);
      doKeepTuple(pRowSup, tsList[j]);
    }
  }

  SResultRow* pResult = NULL;
  pRowSup->win.ekey = tsList[pBlock->info.rows - 1];
1833 1834
  int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1835 1836 1837 1838 1839
  if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
    longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
  }

  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, false);
1840
  doApplyFunctions(pTaskInfo, pSup->pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1841 1842 1843
                   pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
}

1844
static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
1845 1846 1847 1848 1849 1850
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SSessionAggOperatorInfo* pInfo = pOperator->info;
  SOptrBasicInfo*          pBInfo = &pInfo->binfo;
1851
  SExprSupp*               pSup = &pOperator->exprSupp;
1852 1853

  if (pOperator->status == OP_RES_TO_RETURN) {
1854
    while (1) {
1855 1856 1857 1858 1859 1860 1861 1862
      doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
      doFilter(pInfo->pCondition, pBInfo->pRes);

      bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
      if (!hasRemain) {
        doSetOperatorCompleted(pOperator);
        break;
      }
1863

1864 1865 1866 1867 1868
      if (pBInfo->pRes->info.rows > 0) {
        break;
      }
    }
    pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1869
    return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1870 1871
  }

1872 1873 1874
  int64_t st = taosGetTimestampUs();
  int32_t order = TSDB_ORDER_ASC;

1875 1876 1877
  SOperatorInfo* downstream = pOperator->pDownstream[0];

  while (1) {
1878
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1879 1880 1881 1882 1883
    if (pBlock == NULL) {
      break;
    }

    // the pDataBlock are always the same one, no need to call this again
1884
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
1885 1886
    blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);

1887 1888 1889
    doSessionWindowAggImpl(pOperator, pInfo, pBlock);
  }

1890 1891
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;

1892 1893 1894 1895
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
  closeAllResultRows(&pBInfo->resultRowInfo);

1896
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
1897
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
1898
  while (1) {
1899 1900
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
    doFilter(pInfo->pCondition, pBInfo->pRes);
1901

1902 1903 1904 1905 1906
    bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
    if (!hasRemain) {
      doSetOperatorCompleted(pOperator);
      break;
    }
1907

1908 1909 1910 1911 1912
    if (pBInfo->pRes->info.rows > 0) {
      break;
    }
  }
  pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1913
  return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1914 1915
}

1916
static void doKeepPrevRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock* pBlock, int32_t rowIndex) {
H
Haojun Liao 已提交
1917
  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
1918 1919
  for (int32_t i = 0; i < numOfCols; ++i) {
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
H
Haojun Liao 已提交
1920 1921 1922 1923 1924 1925

    // null data should not be kept since it can not be used to perform interpolation
    if (!colDataIsNull_s(pColInfoData, i)) {
      SGroupKeys* pkey = taosArrayGet(pSliceInfo->pPrevRow, i);

      pkey->isNull = false;
1926
      char* val = colDataGetData(pColInfoData, rowIndex);
H
Haojun Liao 已提交
1927 1928 1929 1930 1931
      memcpy(pkey->pData, val, pkey->bytes);
    }
  }
}

1932 1933 1934 1935 1936 1937 1938 1939 1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969 1970 1971 1972 1973 1974 1975 1976 1977 1978 1979 1980 1981 1982 1983 1984 1985 1986 1987 1988 1989 1990 1991 1992 1993 1994 1995 1996 1997 1998 1999 2000 2001 2002 2003 2004 2005 2006 2007 2008 2009 2010 2011 2012 2013 2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024 2025 2026 2027 2028 2029 2030 2031
static void genInterpolationResult(STimeSliceOperatorInfo* pSliceInfo, SExprSupp* pExprSup, SSDataBlock* pBlock,
                                   int32_t rowIndex, SSDataBlock* pResBlock) {
  int32_t rows = pResBlock->info.rows;

  // todo set the correct primary timestamp column

  // output the result
  for (int32_t j = 0; j < pExprSup->numOfExprs; ++j) {
    SExprInfo* pExprInfo = &pExprSup->pExprInfo[j];
    int32_t    dstSlot = pExprInfo->base.resSchema.slotId;
    int32_t    srcSlot = pExprInfo->base.pParam[0].pCol->slotId;

    SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
    SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);

    switch (pSliceInfo->fillType) {
      case TSDB_FILL_NULL:
        colDataAppendNULL(pDst, rows);
        break;

      case TSDB_FILL_SET_VALUE: {
        SVariant* pVar = &pSliceInfo->pFillColInfo[j].fillVal;

        if (pDst->info.type == TSDB_DATA_TYPE_FLOAT) {
          float v = 0;
          GET_TYPED_DATA(v, float, pVar->nType, &pVar->i);
          colDataAppend(pDst, rows, (char*)&v, false);
        } else if (pDst->info.type == TSDB_DATA_TYPE_DOUBLE) {
          double v = 0;
          GET_TYPED_DATA(v, double, pVar->nType, &pVar->i);
          colDataAppend(pDst, rows, (char*)&v, false);
        } else if (IS_SIGNED_NUMERIC_TYPE(pDst->info.type)) {
          int64_t v = 0;
          GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i);
          colDataAppend(pDst, rows, (char*)&v, false);
        }
      } break;

      case TSDB_FILL_LINEAR:
#if 0
        if (pCtx->start.key == INT64_MIN || pCtx->start.key > pCtx->startTs
                    || pCtx->end.key == INT64_MIN || pCtx->end.key < pCtx->startTs) {
//                  goto interp_exit;
                }

              double v1 = -1, v2 = -1;
              GET_TYPED_DATA(v1, double, pCtx->inputType, &pCtx->start.val);
              GET_TYPED_DATA(v2, double, pCtx->inputType, &pCtx->end.val);

              SPoint point1 = {.key = ts, .val = &v1};
              SPoint point2 = {.key = nextTs, .val = &v2};
              SPoint point  = {.key = pCtx->startTs, .val = pCtx->pOutput};

              int32_t srcType = pCtx->inputType;
              if (isNull((char *)&pCtx->start.val, srcType) || isNull((char *)&pCtx->end.val, srcType)) {
                setNull(pCtx->pOutput, srcType, pCtx->inputBytes);
              } else {
                bool exceedMax = false, exceedMin = false;
                taosGetLinearInterpolationVal(&point, pCtx->outputType, &point1, &point2, TSDB_DATA_TYPE_DOUBLE, &exceedMax, &exceedMin);
                if (exceedMax || exceedMin) {
                  __compar_fn_t func = getComparFunc((int32_t)pCtx->inputType, 0);
                  if (func(&pCtx->start.val, &pCtx->end.val) <= 0) {
                    COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, exceedMax ? &pCtx->start.val : &pCtx->end.val);
                  } else {
                    COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, exceedMax ? &pCtx->end.val : &pCtx->start.val);
                  }
                }
              }
#endif
        break;

      case TSDB_FILL_PREV: {
        SGroupKeys* pkey = taosArrayGet(pSliceInfo->pPrevRow, srcSlot);
        colDataAppend(pDst, rows, pkey->pData, false);
      } break;

      case TSDB_FILL_NEXT: {
        char* p = colDataGetData(pSrc, rowIndex);
        colDataAppend(pDst, rows, p, colDataIsNull_s(pSrc, rowIndex));
      } break;

      case TSDB_FILL_NONE:
      default:
        break;
    }
  }

  pResBlock->info.rows += 1;
}

static int32_t initPrevRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pBlock) {
  if (pInfo->pPrevRow != NULL) {
    return TSDB_CODE_SUCCESS;
  }

  pInfo->pPrevRow = taosArrayInit(4, sizeof(SGroupKeys));
  if (pInfo->pPrevRow == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

2032
  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
2033 2034 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046
  for (int32_t i = 0; i < numOfCols; ++i) {
    SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);

    SGroupKeys key = {0};
    key.bytes = pColInfo->info.bytes;
    key.type = pColInfo->info.type;
    key.isNull = false;
    key.pData = taosMemoryCalloc(1, pColInfo->info.bytes);
    taosArrayPush(pInfo->pPrevRow, &key);
  }

  return TSDB_CODE_SUCCESS;
}

H
Haojun Liao 已提交
2047
static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
2048 2049 2050 2051
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

2052 2053
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

2054
  STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
2055 2056
  SSDataBlock*            pResBlock = pSliceInfo->pRes;
  SExprSupp*              pSup = &pOperator->exprSupp;
H
Haojun Liao 已提交
2057

2058 2059
  blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);

2060 2061 2062 2063 2064 2065 2066 2067
  //  if (pOperator->status == OP_RES_TO_RETURN) {
  //    //    doBuildResultDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pIntervalInfo->pRes);
  //    if (pResBlock->info.rows == 0 || !hasDataInGroupInfo(&pSliceInfo->groupResInfo)) {
  //      doSetOperatorCompleted(pOperator);
  //    }
  //
  //    return pResBlock;
  //  }
2068

2069 2070
  int32_t        order = TSDB_ORDER_ASC;
  SInterval*     pInterval = &pSliceInfo->interval;
2071 2072
  SOperatorInfo* downstream = pOperator->pDownstream[0];

H
Haojun Liao 已提交
2073
  int32_t numOfRows = 0;
2074
  while (1) {
2075
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
2076 2077 2078 2079
    if (pBlock == NULL) {
      break;
    }

2080 2081 2082 2083 2084
    int32_t code = initPrevRowsKeeper(pSliceInfo, pBlock);
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
    }

2085
    // the pDataBlock are always the same one, no need to call this again
2086
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, order, MAIN_SCAN, true);
H
Haojun Liao 已提交
2087

2088
    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
2089 2090
    for (int32_t i = 0; i < pBlock->info.rows; ++i) {
      int64_t ts = *(int64_t*)colDataGetData(pTsCol, i);
H
Haojun Liao 已提交
2091 2092

      if (ts == pSliceInfo->current) {
2093
        for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
2094
          SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j];
2095 2096
          int32_t    dstSlot = pExprInfo->base.resSchema.slotId;
          int32_t    srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
H
Haojun Liao 已提交
2097 2098

          SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
2099
          SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
H
Haojun Liao 已提交
2100 2101 2102 2103 2104

          char* v = colDataGetData(pSrc, i);
          colDataAppend(pDst, numOfRows, v, false);
        }

2105
        pResBlock->info.rows += 1;
2106
        doKeepPrevRows(pSliceInfo, pBlock, i);
H
Haojun Liao 已提交
2107

2108 2109
        pSliceInfo->current =
            taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
H
Haojun Liao 已提交
2110 2111 2112 2113
        if (pSliceInfo->current > pSliceInfo->win.ekey) {
          doSetOperatorCompleted(pOperator);
          break;
        }
2114 2115 2116 2117

        if (pResBlock->info.rows >= pResBlock->info.capacity) {
          break;
        }
H
Haojun Liao 已提交
2118
      } else if (ts < pSliceInfo->current) {
2119
        if (i < pBlock->info.rows - 1) {
2120
          int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
H
Haojun Liao 已提交
2121
          if (nextTs > pSliceInfo->current) {
2122 2123 2124
            while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
              genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
              pSliceInfo->current =
H
Haojun Liao 已提交
2125
                  taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
2126
              if (pResBlock->info.rows >= pResBlock->info.capacity) {
H
Haojun Liao 已提交
2127 2128
                break;
              }
H
Haojun Liao 已提交
2129
            }
2130 2131 2132 2133

            if (pSliceInfo->current > pSliceInfo->win.ekey) {
              doSetOperatorCompleted(pOperator);
              break;
H
Haojun Liao 已提交
2134 2135
            }
          } else {
H
Haojun Liao 已提交
2136
            // ignore current row, and do nothing
H
Haojun Liao 已提交
2137 2138
          }
        } else {  // it is the last row of current block
2139 2140 2141 2142 2143 2144 2145
          doKeepPrevRows(pSliceInfo, pBlock, i);
        }
      } else {  // ts > pSliceInfo->current
        while (pSliceInfo->current < ts && pSliceInfo->current <= pSliceInfo->win.ekey) {
          genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
          pSliceInfo->current =
              taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
2146 2147 2148
          if (pResBlock->info.rows >= pResBlock->info.capacity) {
            break;
          }
2149 2150 2151 2152 2153
        }

        if (pSliceInfo->current > pSliceInfo->win.ekey) {
          doSetOperatorCompleted(pOperator);
          break;
H
Haojun Liao 已提交
2154 2155 2156
        }
      }
    }
2157 2158 2159 2160
  }

  // restore the value
  setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
H
Haojun Liao 已提交
2161
  if (pResBlock->info.rows == 0) {
2162 2163 2164
    pOperator->status = OP_EXEC_DONE;
  }

H
Haojun Liao 已提交
2165 2166 2167
  return pResBlock->info.rows == 0 ? NULL : pResBlock;
}

2168
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo) {
2169 2170 2171 2172 2173 2174
  STimeSliceOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STimeSliceOperatorInfo));
  SOperatorInfo*          pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pOperator == NULL || pInfo == NULL) {
    goto _error;
  }

2175
  SInterpFuncPhysiNode* pInterpPhyNode = (SInterpFuncPhysiNode*)pPhyNode;
2176
  SExprSupp*            pSup = &pOperator->exprSupp;
2177

2178
  int32_t    numOfExprs = 0;
2179
  SExprInfo* pExprInfo = createExprInfo(pInterpPhyNode->pFuncs, NULL, &numOfExprs);
2180
  int32_t    code = initExprSupp(pSup, pExprInfo, numOfExprs);
H
Haojun Liao 已提交
2181 2182 2183 2184
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

2185
  if (pInterpPhyNode->pExprs != NULL) {
2186
    int32_t    num = 0;
2187 2188 2189 2190 2191 2192 2193 2194 2195 2196
    SExprInfo* pScalarExprInfo = createExprInfo(pInterpPhyNode->pExprs, NULL, &num);
    code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, num);
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }

  pInfo->tsCol = extractColumnFromColumnNode((SColumnNode*)pInterpPhyNode->pTimeSeries);
  pInfo->fillType = convertFillType(pInterpPhyNode->fillMode);
  initResultSizeInfo(pOperator, 4096);
2197

2198 2199 2200
  pInfo->pFillColInfo = createFillColInfo(pExprInfo, numOfExprs, (SNodeListNode*)pInterpPhyNode->pFillValues);
  pInfo->pRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
  pInfo->win = pInterpPhyNode->timeRange;
2201
  pInfo->interval.interval = pInterpPhyNode->interval;
2202
  pInfo->current = pInfo->win.skey;
H
Haojun Liao 已提交
2203

2204
  pOperator->name = "TimeSliceOperator";
2205
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC;
2206 2207 2208 2209
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
2210

2211 2212
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doTimeslice, NULL, NULL, destroyBasicOperatorInfo, NULL, NULL, NULL);
2213

H
Haojun Liao 已提交
2214
  code = appendDownstream(pOperator, &downstream, 1);
2215 2216
  return pOperator;

L
Liu Jicong 已提交
2217
_error:
2218 2219 2220 2221 2222 2223 2224
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
  return NULL;
}

SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
2225
                                             SSDataBlock* pResBlock, STimeWindowAggSupp* pTwAggSup, int32_t tsSlotId,
2226
                                             SColumn* pStateKeyCol, SNode* pCondition, SExecTaskInfo* pTaskInfo) {
2227 2228 2229 2230 2231 2232
  SStateWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStateWindowOperatorInfo));
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

2233 2234 2235 2236
  pInfo->stateCol = *pStateKeyCol;
  pInfo->stateKey.type = pInfo->stateCol.type;
  pInfo->stateKey.bytes = pInfo->stateCol.bytes;
  pInfo->stateKey.pData = taosMemoryCalloc(1, pInfo->stateCol.bytes);
2237
  pInfo->pCondition = pCondition;
2238 2239 2240 2241
  if (pInfo->stateKey.pData == NULL) {
    goto _error;
  }

2242 2243 2244
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;

  initResultSizeInfo(pOperator, 4096);
2245 2246 2247
  initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExpr, numOfCols, keyBufSize, pTaskInfo->id.str);
  initBasicInfo(&pInfo->binfo, pResBlock);

2248
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
2249

L
Liu Jicong 已提交
2250
  pInfo->twAggSup = *pTwAggSup;
2251 2252
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

X
Xiaoyu Wang 已提交
2253 2254
  pInfo->tsSlotId = tsSlotId;
  pOperator->name = "StateWindowOperator";
2255
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE;
X
Xiaoyu Wang 已提交
2256 2257
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
2258 2259
  pOperator->exprSupp.pExprInfo = pExpr;
  pOperator->exprSupp.numOfExprs = numOfCols;
X
Xiaoyu Wang 已提交
2260 2261
  pOperator->pTaskInfo = pTaskInfo;
  pOperator->info = pInfo;
2262 2263 2264 2265 2266 2267 2268

  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStateWindowAgg, NULL, NULL,
                                         destroyStateWindowOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);

  int32_t code = appendDownstream(pOperator, &downstream, 1);
  return pOperator;

L
Liu Jicong 已提交
2269
_error:
2270 2271 2272 2273 2274 2275
  pTaskInfo->code = TSDB_CODE_SUCCESS;
  return NULL;
}

void destroySWindowOperatorInfo(void* param, int32_t numOfOutput) {
  SSessionAggOperatorInfo* pInfo = (SSessionAggOperatorInfo*)param;
2276
  cleanupBasicInfo(&pInfo->binfo);
2277

D
dapan1121 已提交
2278
  taosMemoryFreeClear(param);
2279 2280 2281
}

SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
L
Liu Jicong 已提交
2282
                                            SSDataBlock* pResBlock, int64_t gap, int32_t tsSlotId,
2283 2284
                                            STimeWindowAggSupp* pTwAggSupp, SNode* pCondition,
                                            SExecTaskInfo* pTaskInfo) {
2285 2286 2287 2288 2289 2290
  SSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSessionAggOperatorInfo));
  SOperatorInfo*           pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

2291 2292
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  initResultSizeInfo(pOperator, 4096);
2293

2294
  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
2295 2296 2297 2298
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

2299 2300
  initBasicInfo(&pInfo->binfo, pResBlock);

2301
  pInfo->twAggSup = *pTwAggSupp;
2302
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
2303 2304
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

L
Liu Jicong 已提交
2305 2306 2307 2308 2309
  pInfo->tsSlotId = tsSlotId;
  pInfo->gap = gap;
  pInfo->binfo.pRes = pResBlock;
  pInfo->winSup.prevTs = INT64_MIN;
  pInfo->reptScan = false;
2310
  pInfo->pCondition = pCondition;
L
Liu Jicong 已提交
2311
  pOperator->name = "SessionWindowAggOperator";
2312
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION;
2313
  pOperator->blocking = true;
L
Liu Jicong 已提交
2314
  pOperator->status = OP_NOT_OPENED;
2315 2316
  pOperator->exprSupp.pExprInfo = pExprInfo;
  pOperator->exprSupp.numOfExprs = numOfCols;
L
Liu Jicong 已提交
2317
  pOperator->info = pInfo;
2318 2319 2320 2321 2322 2323 2324 2325

  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doSessionWindowAgg, NULL, NULL,
                                         destroySWindowOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
  pOperator->pTaskInfo = pTaskInfo;

  code = appendDownstream(pOperator, &downstream, 1);
  return pOperator;

L
Liu Jicong 已提交
2326
_error:
2327 2328 2329 2330 2331 2332 2333 2334
  if (pInfo != NULL) {
    destroySWindowOperatorInfo(pInfo, numOfCols);
  }

  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
L
Liu Jicong 已提交
2335
}
5
54liuyao 已提交
2336

5
54liuyao 已提交
2337 2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354
void compactFunctions(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int32_t numOfOutput,
                      SExecTaskInfo* pTaskInfo) {
  for (int32_t k = 0; k < numOfOutput; ++k) {
    if (fmIsWindowPseudoColumnFunc(pDestCtx[k].functionId)) {
      continue;
    }
    int32_t code = TSDB_CODE_SUCCESS;
    if (functionNeedToExecute(&pDestCtx[k]) && pDestCtx[k].fpSet.combine != NULL) {
      code = pDestCtx[k].fpSet.combine(&pDestCtx[k], &pSourceCtx[k]);
      if (code != TSDB_CODE_SUCCESS) {
        qError("%s apply functions error, code: %s", GET_TASKID(pTaskInfo), tstrerror(code));
        pTaskInfo->code = code;
        longjmp(pTaskInfo->env, code);
      }
    }
  }
}

2355 2356 2357 2358 2359 2360 2361 2362
bool hasIntervalWindow(SAggSupporter* pSup, TSKEY ts, uint64_t groupId) {
  int32_t bytes = sizeof(TSKEY);
  SET_RES_WINDOW_KEY(pSup->keyBuf, &ts, bytes, groupId);
  SResultRowPosition* p1 =
      (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf, GET_RES_WINDOW_KEY_LEN(bytes));
  return p1 != NULL;
}

2363 2364
static void rebuildIntervalWindow(SStreamFinalIntervalOperatorInfo* pInfo, SExprSupp* pSup, SArray* pWinArray,
                                  int32_t groupId, int32_t numOfOutput, SExecTaskInfo* pTaskInfo, SArray* pUpdated) {
5
54liuyao 已提交
2365
  int32_t size = taosArrayGetSize(pWinArray);
5
54liuyao 已提交
2366 2367 2368
  if (!pInfo->pChildren) {
    return;
  }
5
54liuyao 已提交
2369
  for (int32_t i = 0; i < size; i++) {
2370 2371 2372 2373 2374
    SWinRes*    pWinRes = taosArrayGet(pWinArray, i);
    SResultRow* pCurResult = NULL;
    STimeWindow ParentWin = {.skey = pWinRes->ts, .ekey = pWinRes->ts + 1};
    setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &ParentWin, true, &pCurResult, pWinRes->groupId, pSup->pCtx,
                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
5
54liuyao 已提交
2375
    int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
2376
    bool    find = true;
5
54liuyao 已提交
2377 2378 2379
    for (int32_t j = 0; j < numOfChildren; j++) {
      SOperatorInfo*            pChildOp = taosArrayGetP(pInfo->pChildren, j);
      SIntervalAggOperatorInfo* pChInfo = pChildOp->info;
2380
      SExprSupp*                pChildSup = &pChildOp->exprSupp;
2381 2382 2383 2384
      if (!hasIntervalWindow(&pChInfo->aggSup, pWinRes->ts, pWinRes->groupId)) {
        continue;
      }
      find = true;
2385
      SResultRow* pChResult = NULL;
2386 2387 2388
      setTimeWindowOutputBuf(&pChInfo->binfo.resultRowInfo, &ParentWin, true, &pChResult, pWinRes->groupId,
                             pChildSup->pCtx, pChildSup->numOfExprs, pChildSup->rowEntryInfoOffset, &pChInfo->aggSup,
                             pTaskInfo);
2389
      compactFunctions(pSup->pCtx, pChildSup->pCtx, numOfOutput, pTaskInfo);
5
54liuyao 已提交
2390
    }
2391 2392 2393
    if (find && pUpdated) {
      saveResultRow(pCurResult, pWinRes->groupId, pUpdated);
    }
5
54liuyao 已提交
2394 2395 2396 2397 2398
  }
}

bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup) {
  SET_RES_WINDOW_KEY(pSup->keyBuf, &pWin->skey, sizeof(int64_t), groupId);
2399 2400
  SResultRowPosition* p1 = (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf,
                                                            GET_RES_WINDOW_KEY_LEN(sizeof(int64_t)));
5
54liuyao 已提交
2401 2402 2403
  return p1 == NULL;
}

L
Liu Jicong 已提交
2404 2405 2406 2407
int32_t getNexWindowPos(SInterval* pInterval, SDataBlockInfo* pBlockInfo, TSKEY* tsCols, int32_t startPos, TSKEY eKey,
                        STimeWindow* pNextWin) {
  int32_t forwardRows =
      getNumOfRowsInTimeWindow(pBlockInfo, tsCols, startPos, eKey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
5
54liuyao 已提交
2408 2409 2410 2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421
  int32_t prevEndPos = forwardRows - 1 + startPos;
  return getNextQualifiedWindow(pInterval, pNextWin, pBlockInfo, tsCols, prevEndPos, TSDB_ORDER_ASC);
}

void addPullWindow(SHashObj* pMap, SWinRes* pWinRes, int32_t size) {
  SArray* childIds = taosArrayInit(8, sizeof(int32_t));
  for (int32_t i = 0; i < size; i++) {
    taosArrayPush(childIds, &i);
  }
  taosHashPut(pMap, pWinRes, sizeof(SWinRes), &childIds, sizeof(void*));
}

static int32_t getChildIndex(SSDataBlock* pBlock) { return pBlock->info.childId; }

5
54liuyao 已提交
2422 2423 2424 2425 2426 2427
STimeWindow getFinalTimeWindow(int64_t ts, SInterval* pInterval) {
  STimeWindow w = {.skey = ts, .ekey = INT64_MAX};
  w.ekey = taosTimeAdd(w.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
  return w;
}

5
54liuyao 已提交
2428
static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBlock, uint64_t tableGroupId,
S
shenglian zhou 已提交
2429
                           SArray* pUpdated) {
5
54liuyao 已提交
2430
  SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)pOperatorInfo->info;
X
Xiaoyu Wang 已提交
2431 2432
  SResultRowInfo*                   pResultRowInfo = &(pInfo->binfo.resultRowInfo);
  SExecTaskInfo*                    pTaskInfo = pOperatorInfo->pTaskInfo;
2433 2434
  SExprSupp*                        pSup = &pOperatorInfo->exprSupp;
  int32_t                           numOfOutput = pSup->numOfExprs;
X
Xiaoyu Wang 已提交
2435 2436 2437 2438 2439
  int32_t                           step = 1;
  bool                              ascScan = true;
  TSKEY*                            tsCols = NULL;
  SResultRow*                       pResult = NULL;
  int32_t                           forwardRows = 0;
5
54liuyao 已提交
2440

5
54liuyao 已提交
2441 2442 2443
  ASSERT(pSDataBlock->pDataBlock != NULL);
  SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
  tsCols = (int64_t*)pColDataInfo->pData;
5
54liuyao 已提交
2444

X
Xiaoyu Wang 已提交
2445 2446
  int32_t     startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1);
  TSKEY       ts = getStartTsKey(&pSDataBlock->info.window, tsCols);
5
54liuyao 已提交
2447 2448 2449 2450 2451 2452
  STimeWindow nextWin = {0};
  if (IS_FINAL_OP(pInfo)) {
    nextWin = getFinalTimeWindow(ts, &pInfo->interval);
  } else {
    nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->order);
  }
5
54liuyao 已提交
2453
  while (1) {
5
54liuyao 已提交
2454
    bool isClosed = isCloseWindow(&nextWin, &pInfo->twAggSup);
5
54liuyao 已提交
2455
    if ((pInfo->ignoreExpiredData && isClosed) || !inSlidingWindow(&pInfo->interval, &nextWin, &pSDataBlock->info)) {
5
54liuyao 已提交
2456 2457 2458 2459 2460 2461 2462
      startPos = getNexWindowPos(&pInfo->interval, &pSDataBlock->info, tsCols, startPos, nextWin.ekey, &nextWin);
      if (startPos < 0) {
        break;
      }
      continue;
    }
    if (IS_FINAL_OP(pInfo) && isClosed && pInfo->pChildren) {
L
Liu Jicong 已提交
2463 2464 2465 2466 2467
      bool    ignore = true;
      SWinRes winRes = {
          .ts = nextWin.skey,
          .groupId = tableGroupId,
      };
5
54liuyao 已提交
2468 2469 2470 2471 2472 2473 2474 2475 2476 2477
      void* chIds = taosHashGet(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
      if (isDeletedWindow(&nextWin, tableGroupId, &pInfo->aggSup) && !chIds) {
        SPullWindowInfo pull = {.window = nextWin, .groupId = tableGroupId};
        // add pull data request
        taosArrayPush(pInfo->pPullWins, &pull);
        addPullWindow(pInfo->pPullDataMap, &winRes, taosArrayGetSize(pInfo->pChildren));
      } else {
        int32_t index = -1;
        SArray* chArray = NULL;
        if (chIds) {
L
Liu Jicong 已提交
2478
          chArray = *(void**)chIds;
5
54liuyao 已提交
2479 2480 2481
          int32_t chId = getChildIndex(pSDataBlock);
          index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
        }
L
Liu Jicong 已提交
2482
        if (index != -1 && pSDataBlock->info.type == STREAM_PULL_DATA) {
5
54liuyao 已提交
2483 2484 2485 2486 2487 2488
          taosArrayRemove(chArray, index);
          if (taosArrayGetSize(chArray) == 0) {
            // pull data is over
            taosHashRemove(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
          }
        }
L
Liu Jicong 已提交
2489
        if (index == -1 || pSDataBlock->info.type == STREAM_PULL_DATA) {
5
54liuyao 已提交
2490 2491 2492 2493 2494 2495 2496 2497 2498 2499 2500
          ignore = false;
        }
      }

      if (ignore) {
        startPos = getNexWindowPos(&pInfo->interval, &pSDataBlock->info, tsCols, startPos, nextWin.ekey, &nextWin);
        if (startPos < 0) {
          break;
        }
        continue;
      }
5
54liuyao 已提交
2501
    }
5
54liuyao 已提交
2502

2503 2504
    int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, true, &pResult, tableGroupId, pSup->pCtx,
                                          numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
5
54liuyao 已提交
2505 2506 2507
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
5
54liuyao 已提交
2508

5
54liuyao 已提交
2509 2510 2511 2512
    if (IS_FINAL_OP(pInfo)) {
      forwardRows = 1;
    } else {
      forwardRows = getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, nextWin.ekey, binarySearchForKey, NULL,
S
shenglian zhou 已提交
2513
                                           TSDB_ORDER_ASC);
5
54liuyao 已提交
2514
    }
5
54liuyao 已提交
2515
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdated) {
5
54liuyao 已提交
2516
      saveResultRow(pResult, tableGroupId, pUpdated);
5
54liuyao 已提交
2517
    }
5
54liuyao 已提交
2518
    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true);
2519 2520
    doApplyFunctions(pTaskInfo, pSup->pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols,
                     pSDataBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
2521
    int32_t prevEndPos = (forwardRows - 1) * step + startPos;
2522
    ASSERT(pSDataBlock->info.window.skey > 0 && pSDataBlock->info.window.ekey > 0);
5
54liuyao 已提交
2523 2524 2525 2526 2527 2528 2529
    startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, pInfo->order);
    if (startPos < 0) {
      break;
    }
  }
}

5
54liuyao 已提交
2530 2531 2532 2533
static void clearStreamIntervalOperator(SStreamFinalIntervalOperatorInfo* pInfo) {
  taosHashClear(pInfo->aggSup.pResultRowHashTable);
  clearDiskbasedBuf(pInfo->aggSup.pResultBuf);
  cleanupResultRowInfo(&pInfo->binfo.resultRowInfo);
2534
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
5
54liuyao 已提交
2535 2536
}

5
54liuyao 已提交
2537 2538 2539 2540
static void clearSpecialDataBlock(SSDataBlock* pBlock) {
  if (pBlock->info.rows <= 0) {
    return;
  }
5
54liuyao 已提交
2541 2542 2543
  blockDataCleanup(pBlock);
}

2544
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex) {
5
54liuyao 已提交
2545 2546
  // ASSERT(pDest->info.capacity >= pSource->info.rows);
  blockDataEnsureCapacity(pDest, pSource->info.rows);
5
54liuyao 已提交
2547
  clearSpecialDataBlock(pDest);
5
54liuyao 已提交
2548 2549
  SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, 0);
  SColumnInfoData* pSourceCol = taosArrayGet(pSource->pDataBlock, tsColIndex);
2550

5
54liuyao 已提交
2551
  // copy timestamp column
2552 2553
  colDataAssign(pDestCol, pSourceCol, pSource->info.rows, &pDest->info);
  for (int32_t i = 1; i < taosArrayGetSize(pDest->pDataBlock); i++) {
5
54liuyao 已提交
2554 2555 2556
    SColumnInfoData* pCol = taosArrayGet(pDest->pDataBlock, i);
    colDataAppendNNULL(pCol, 0, pSource->info.rows);
  }
2557

5
54liuyao 已提交
2558
  pDest->info.rows = pSource->info.rows;
2559 2560
  pDest->info.groupId = pSource->info.groupId;
  pDest->info.type = pSource->info.type;
5
54liuyao 已提交
2561 2562 2563
  blockDataUpdateTsWindow(pDest, 0);
}

5
54liuyao 已提交
2564 2565 2566 2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577
static bool needBreak(SStreamFinalIntervalOperatorInfo* pInfo) {
  int32_t size = taosArrayGetSize(pInfo->pPullWins);
  if (pInfo->pullIndex < size) {
    return true;
  }
  return false;
}

static void doBuildPullDataBlock(SArray* array, int32_t* pIndex, SSDataBlock* pBlock) {
  clearSpecialDataBlock(pBlock);
  int32_t size = taosArrayGetSize(array);
  if (size - (*pIndex) == 0) {
    return;
  }
L
Liu Jicong 已提交
2578
  blockDataEnsureCapacity(pBlock, size - (*pIndex));
5
54liuyao 已提交
2579 2580
  ASSERT(3 <= taosArrayGetSize(pBlock->pDataBlock));
  for (; (*pIndex) < size; (*pIndex)++) {
L
Liu Jicong 已提交
2581
    SPullWindowInfo* pWin = taosArrayGet(array, (*pIndex));
5
54liuyao 已提交
2582
    SColumnInfoData* pStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
5
54liuyao 已提交
2583 2584
    colDataAppend(pStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false);

5
54liuyao 已提交
2585
    SColumnInfoData* pEndTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
5
54liuyao 已提交
2586 2587
    colDataAppend(pEndTs, pBlock->info.rows, (const char*)&pWin->window.ekey, false);

5
54liuyao 已提交
2588
    SColumnInfoData* pGroupId = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
5
54liuyao 已提交
2589 2590 2591 2592 2593 2594 2595 2596 2597 2598
    colDataAppend(pGroupId, pBlock->info.rows, (const char*)&pWin->groupId, false);
    pBlock->info.rows++;
  }
  if ((*pIndex) == size) {
    *pIndex = 0;
    taosArrayClear(array);
  }
  blockDataUpdateTsWindow(pBlock, 0);
}

L
Liu Jicong 已提交
2599
void processPullOver(SSDataBlock* pBlock, SHashObj* pMap) {
5
54liuyao 已提交
2600
  SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
2601
  TSKEY*           tsData = (TSKEY*)pStartCol->pData;
5
54liuyao 已提交
2602
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
L
Liu Jicong 已提交
2603 2604
  uint64_t*        groupIdData = (uint64_t*)pGroupCol->pData;
  int32_t          chId = getChildIndex(pBlock);
5
54liuyao 已提交
2605 2606
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    SWinRes winRes = {.ts = tsData[i], .groupId = groupIdData[i]};
L
Liu Jicong 已提交
2607
    void*   chIds = taosHashGet(pMap, &winRes, sizeof(SWinRes));
5
54liuyao 已提交
2608
    if (chIds) {
L
Liu Jicong 已提交
2609
      SArray* chArray = *(SArray**)chIds;
5
54liuyao 已提交
2610 2611 2612 2613 2614 2615 2616 2617 2618 2619 2620
      int32_t index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
      if (index != -1) {
        taosArrayRemove(chArray, index);
        if (taosArrayGetSize(chArray) == 0) {
          // pull data is over
          taosHashRemove(pMap, &winRes, sizeof(SWinRes));
        }
      }
    }
  }
}
5
54liuyao 已提交
2621

5
54liuyao 已提交
2622 2623
static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
  SStreamFinalIntervalOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
2624
  SOperatorInfo*                    downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
2625
  SArray*                           pUpdated = taosArrayInit(4, POINTER_BYTES);
5
54liuyao 已提交
2626
  TSKEY                             maxTs = INT64_MIN;
5
54liuyao 已提交
2627

2628 2629
  SExprSupp* pSup = &pOperator->exprSupp;

5
54liuyao 已提交
2630 2631 2632 2633
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2634
    if (pInfo->binfo.pRes->info.rows == 0) {
5
54liuyao 已提交
2635
      pOperator->status = OP_EXEC_DONE;
5
54liuyao 已提交
2636 2637 2638
      if (!IS_FINAL_OP(pInfo)) {
        // semi interval operator clear disk buffer
        clearStreamIntervalOperator(pInfo);
2639 2640
      } else {
        freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2641
      }
5
54liuyao 已提交
2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655
      return NULL;
    }
    printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
    return pInfo->binfo.pRes;
  } else {
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
    if (pInfo->binfo.pRes->info.rows != 0) {
      printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
      return pInfo->binfo.pRes;
    }
    if (pInfo->pUpdateRes->info.rows != 0 && pInfo->returnUpdate) {
      pInfo->returnUpdate = false;
      ASSERT(!IS_FINAL_OP(pInfo));
      printDataBlock(pInfo->pUpdateRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
5
54liuyao 已提交
2656 2657
      // process the rest of the data
      return pInfo->pUpdateRes;
5
54liuyao 已提交
2658
    }
5
54liuyao 已提交
2659 2660 2661 2662 2663 2664 2665
    doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
    if (pInfo->pPullDataRes->info.rows != 0) {
      // process the rest of the data
      ASSERT(IS_FINAL_OP(pInfo));
      printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
      return pInfo->pPullDataRes;
    }
2666 2667 2668 2669 2670 2671
    doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
    if (pInfo->pDelRes->info.rows != 0) {
      // process the rest of the data
      printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
2672 2673 2674 2675 2676
  }

  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
5
54liuyao 已提交
2677
      clearSpecialDataBlock(pInfo->pUpdateRes);
2678
      removeDeleteResults(pUpdated, pInfo->pDelWins);
5
54liuyao 已提交
2679
      pOperator->status = OP_RES_TO_RETURN;
5
54liuyao 已提交
2680
      qInfo("%s return data", IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
5
54liuyao 已提交
2681 2682
      break;
    }
5
54liuyao 已提交
2683 2684
    printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval Final recv" : "interval  Semi recv");
    maxTs = TMAX(maxTs, pBlock->info.window.ekey);
2685

L
Liu Jicong 已提交
2686
    if (pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_PULL_DATA ||
L
Liu Jicong 已提交
2687
        pBlock->info.type == STREAM_INVALID) {
5
54liuyao 已提交
2688 2689
      pInfo->binfo.pRes->info.type = pBlock->info.type;
    } else if (pBlock->info.type == STREAM_CLEAR) {
2690
      SArray* pUpWins = taosArrayInit(8, sizeof(SWinRes));
2691
      doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock, pUpWins);
5
54liuyao 已提交
2692
      if (IS_FINAL_OP(pInfo)) {
L
Liu Jicong 已提交
2693 2694
        int32_t                           childIndex = getChildIndex(pBlock);
        SOperatorInfo*                    pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
2695
        SStreamFinalIntervalOperatorInfo* pChildInfo = pChildOp->info;
L
Liu Jicong 已提交
2696
        SExprSupp*                        pChildSup = &pChildOp->exprSupp;
2697

2698 2699 2700
        doClearWindows(&pChildInfo->aggSup, pChildSup, &pChildInfo->interval, pChildSup->numOfExprs, pBlock, NULL);
        rebuildIntervalWindow(pInfo, pSup, pUpWins, pInfo->binfo.pRes->info.groupId, pOperator->exprSupp.numOfExprs,
                              pOperator->pTaskInfo, NULL);
5
54liuyao 已提交
2701 2702
        taosArrayDestroy(pUpWins);
        continue;
2703
      }
5
54liuyao 已提交
2704 2705
      removeResults(pUpWins, pUpdated);
      copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
2706
      pInfo->returnUpdate = true;
2707
      taosArrayDestroy(pUpWins);
5
54liuyao 已提交
2708
      break;
2709 2710 2711 2712 2713 2714 2715 2716 2717
    } else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT) {
      doDeleteSpecifyIntervalWindow(&pInfo->aggSup, pBlock, pInfo->pDelWins, &pInfo->interval);
      if (IS_FINAL_OP(pInfo)) {
        int32_t                           childIndex = getChildIndex(pBlock);
        SOperatorInfo*                    pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
        SStreamFinalIntervalOperatorInfo* pChildInfo = pChildOp->info;
        SExprSupp*                        pChildSup = &pChildOp->exprSupp;
        doDeleteSpecifyIntervalWindow(&pChildInfo->aggSup, pBlock, NULL, &pChildInfo->interval);
        rebuildIntervalWindow(pInfo, pSup, pInfo->pDelWins, pInfo->binfo.pRes->info.groupId,
2718
                              pOperator->exprSupp.numOfExprs, pOperator->pTaskInfo, pUpdated);
2719 2720 2721 2722
        continue;
      }
      removeResults(pInfo->pDelWins, pUpdated);
      break;
5
54liuyao 已提交
2723
    } else if (pBlock->info.type == STREAM_GET_ALL && IS_FINAL_OP(pInfo)) {
5
54liuyao 已提交
2724 2725
      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
      continue;
5
54liuyao 已提交
2726
    } else if (pBlock->info.type == STREAM_RETRIEVE && !IS_FINAL_OP(pInfo)) {
2727
      SArray* pUpWins = taosArrayInit(8, sizeof(SWinRes));
5
54liuyao 已提交
2728
      doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock, pUpWins);
5
54liuyao 已提交
2729 2730 2731 2732 2733 2734
      removeResults(pUpWins, pUpdated);
      taosArrayDestroy(pUpWins);
      if (taosArrayGetSize(pUpdated) > 0) {
        break;
      }
      continue;
L
Liu Jicong 已提交
2735 2736
    } else if (pBlock->info.type == STREAM_PULL_OVER && IS_FINAL_OP(pInfo)) {
      processPullOver(pBlock, pInfo->pPullDataMap);
5
54liuyao 已提交
2737
      continue;
5
54liuyao 已提交
2738
    }
5
54liuyao 已提交
2739

5
54liuyao 已提交
2740 2741 2742 2743
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
2744
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
5
54liuyao 已提交
2745
    doHashInterval(pOperator, pBlock, pBlock->info.groupId, pUpdated);
5
54liuyao 已提交
2746
    if (IS_FINAL_OP(pInfo)) {
S
shenglian zhou 已提交
2747
      int32_t chIndex = getChildIndex(pBlock);
5
54liuyao 已提交
2748 2749 2750 2751 2752 2753 2754
      int32_t size = taosArrayGetSize(pInfo->pChildren);
      // if chIndex + 1 - size > 0, add new child
      for (int32_t i = 0; i < chIndex + 1 - size; i++) {
        SOperatorInfo* pChildOp = createStreamFinalIntervalOperatorInfo(NULL, pInfo->pPhyNode, pOperator->pTaskInfo, 0);
        if (!pChildOp) {
          longjmp(pOperator->pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
        }
2755 2756
        SStreamFinalIntervalOperatorInfo* pTmpInfo = pChildOp->info;
        pTmpInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
5
54liuyao 已提交
2757 2758
        taosArrayPush(pInfo->pChildren, &pChildOp);
      }
S
shenglian zhou 已提交
2759
      SOperatorInfo*                    pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
5
54liuyao 已提交
2760
      SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info;
2761
      setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, pChInfo->order, MAIN_SCAN, true);
5
54liuyao 已提交
2762
      doHashInterval(pChildOp, pBlock, pBlock->info.groupId, NULL);
5
54liuyao 已提交
2763 2764 2765 2766

      if (needBreak(pInfo)) {
        break;
      }
5
54liuyao 已提交
2767 2768
    }
  }
S
shenglian zhou 已提交
2769

5
54liuyao 已提交
2770
  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
5
54liuyao 已提交
2771
  if (IS_FINAL_OP(pInfo)) {
2772 2773
    closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pInfo->pPullDataMap,
                        pUpdated, pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2774
    closeChildIntervalWindow(pInfo->pChildren, pInfo->twAggSup.maxTs);
5
54liuyao 已提交
2775 2776
  }

2777
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
2778 2779 2780
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2781 2782 2783 2784 2785 2786 2787 2788 2789
  if (pInfo->binfo.pRes->info.rows != 0) {
    printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
    return pInfo->binfo.pRes;
  }

  if (pInfo->pUpdateRes->info.rows != 0 && pInfo->returnUpdate) {
    pInfo->returnUpdate = false;
    ASSERT(!IS_FINAL_OP(pInfo));
    printDataBlock(pInfo->pUpdateRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
5
54liuyao 已提交
2790 2791 2792
    // process the rest of the data
    return pInfo->pUpdateRes;
  }
5
54liuyao 已提交
2793 2794 2795 2796 2797 2798 2799 2800

  doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
  if (pInfo->pPullDataRes->info.rows != 0) {
    // process the rest of the data
    ASSERT(IS_FINAL_OP(pInfo));
    printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
    return pInfo->pPullDataRes;
  }
2801 2802 2803 2804 2805 2806 2807

  doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
  if (pInfo->pDelRes->info.rows != 0) {
    // process the rest of the data
    printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
    return pInfo->pDelRes;
  }
5
54liuyao 已提交
2808 2809 2810 2811 2812 2813 2814 2815 2816 2817 2818 2819 2820 2821 2822 2823 2824 2825 2826 2827 2828 2829 2830 2831 2832 2833
  // ASSERT(false);
  return NULL;
}

SSDataBlock* createPullDataBlock() {
  SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
  pBlock->info.hasVarCol = false;
  pBlock->info.groupId = 0;
  pBlock->info.rows = 0;
  pBlock->info.type = STREAM_RETRIEVE;
  pBlock->info.rowSize = sizeof(TSKEY) + sizeof(TSKEY) + sizeof(uint64_t);

  pBlock->pDataBlock = taosArrayInit(3, sizeof(SColumnInfoData));
  SColumnInfoData infoData = {0};
  infoData.info.type = TSDB_DATA_TYPE_TIMESTAMP;
  infoData.info.bytes = sizeof(TSKEY);
  // window start ts
  taosArrayPush(pBlock->pDataBlock, &infoData);
  // window end ts
  taosArrayPush(pBlock->pDataBlock, &infoData);

  infoData.info.type = TSDB_DATA_TYPE_UBIGINT;
  infoData.info.bytes = sizeof(uint64_t);
  taosArrayPush(pBlock->pDataBlock, &infoData);

  return pBlock;
5
54liuyao 已提交
2834 2835
}

S
shenglian zhou 已提交
2836 2837 2838
SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                     SExecTaskInfo* pTaskInfo, int32_t numOfChild) {
  SIntervalPhysiNode*               pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
5
54liuyao 已提交
2839
  SStreamFinalIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamFinalIntervalOperatorInfo));
S
shenglian zhou 已提交
2840
  SOperatorInfo*                    pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
2841 2842 2843
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
2844

5
54liuyao 已提交
2845
  pInfo->order = TSDB_ORDER_ASC;
S
shenglian zhou 已提交
2846 2847 2848 2849 2850 2851 2852 2853
  pInfo->interval = (SInterval){.interval = pIntervalPhyNode->interval,
                                .sliding = pIntervalPhyNode->sliding,
                                .intervalUnit = pIntervalPhyNode->intervalUnit,
                                .slidingUnit = pIntervalPhyNode->slidingUnit,
                                .offset = pIntervalPhyNode->offset,
                                .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision};
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pIntervalPhyNode->window.watermark,
5
54liuyao 已提交
2854 2855
      .calTrigger = pIntervalPhyNode->window.triggerType,
      .maxTs = INT64_MIN,
S
shenglian zhou 已提交
2856
  };
2857
  ASSERT(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY);
5
54liuyao 已提交
2858 2859 2860
  pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
2861 2862 2863 2864 2865 2866 2867 2868 2869
  if (pIntervalPhyNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pIntervalPhyNode->window.pExprs, NULL, &numOfScalar);
    int32_t    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }

S
shenglian zhou 已提交
2870 2871
  int32_t      numOfCols = 0;
  SExprInfo*   pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
5
54liuyao 已提交
2872
  SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
2873 2874 2875 2876

  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
  initBasicInfo(&pInfo->binfo, pResBlock);

2877
  ASSERT(numOfCols > 0);
2878
  increaseTs(pOperator->exprSupp.pCtx);
5
54liuyao 已提交
2879 2880 2881 2882
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2883
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
5
54liuyao 已提交
2884 2885
  pInfo->pChildren = NULL;
  if (numOfChild > 0) {
2886
    pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
5
54liuyao 已提交
2887 2888 2889
    for (int32_t i = 0; i < numOfChild; i++) {
      SOperatorInfo* pChildOp = createStreamFinalIntervalOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
      if (pChildOp) {
2890 2891
        SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info;
        pChInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
5
54liuyao 已提交
2892 2893 2894 2895 2896 2897
        taosArrayPush(pInfo->pChildren, &pChildOp);
        continue;
      }
      goto _error;
    }
  }
S
shenglian zhou 已提交
2898
  pInfo->pUpdateRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
5
54liuyao 已提交
2899
  pInfo->pUpdateRes->info.type = STREAM_CLEAR;
5
54liuyao 已提交
2900
  blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
5
54liuyao 已提交
2901 2902
  pInfo->returnUpdate = false;

2903
  pInfo->pPhyNode = (SPhysiNode*)nodesCloneNode((SNode*)pPhyNode);
5
54liuyao 已提交
2904

5
54liuyao 已提交
2905 2906 2907 2908
  if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL) {
    pInfo->isFinal = true;
    pOperator->name = "StreamFinalIntervalOperator";
  } else {
5
54liuyao 已提交
2909
    // semi interval operator does not catch result
5
54liuyao 已提交
2910 2911 2912 2913
    pInfo->isFinal = false;
    pOperator->name = "StreamSemiIntervalOperator";
  }

5
54liuyao 已提交
2914
  if (!IS_FINAL_OP(pInfo) || numOfChild == 0) {
5
54liuyao 已提交
2915 2916
    pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
  }
5
54liuyao 已提交
2917 2918 2919 2920 2921
  pInfo->pPullWins = taosArrayInit(8, sizeof(SPullWindowInfo));
  pInfo->pullIndex = 0;
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  pInfo->pPullDataMap = taosHashInit(64, hashFn, false, HASH_NO_LOCK);
  pInfo->pPullDataRes = createPullDataBlock();
5
54liuyao 已提交
2922
  pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired;
5
54liuyao 已提交
2923
  // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete
2924 2925
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
2926 2927
  pInfo->delIndex = 0;
  pInfo->pDelWins = taosArrayInit(4, sizeof(SWinRes));
5
54liuyao 已提交
2928

5
54liuyao 已提交
2929
  pOperator->operatorType = pPhyNode->type;
5
54liuyao 已提交
2930 2931
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
2932
  pOperator->exprSupp.pExprInfo = pExprInfo;
5
54liuyao 已提交
2933
  pOperator->pTaskInfo = pTaskInfo;
2934
  pOperator->exprSupp.numOfExprs = numOfCols;
5
54liuyao 已提交
2935 2936
  pOperator->info = pInfo;

S
shenglian zhou 已提交
2937 2938 2939
  pOperator->fpSet =
      createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, NULL, destroyStreamFinalIntervalOperatorInfo,
                          aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
2940 2941 2942 2943 2944 2945 2946 2947 2948 2949 2950 2951 2952 2953

  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

_error:
  destroyStreamFinalIntervalOperatorInfo(pInfo, numOfCols);
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
5
54liuyao 已提交
2954
}
5
54liuyao 已提交
2955 2956 2957

void destroyStreamAggSupporter(SStreamAggSupporter* pSup) {
  taosMemoryFreeClear(pSup->pKeyBuf);
2958
  void** pIte = NULL;
2959
  while ((pIte = taosHashIterate(pSup->pResultRows, pIte)) != NULL) {
2960
    SArray* pWins = (SArray*)(*pIte);
2961 2962 2963
    taosArrayDestroy(pWins);
  }
  taosHashCleanup(pSup->pResultRows);
5
54liuyao 已提交
2964 2965 2966 2967 2968
  destroyDiskbasedBuf(pSup->pResultBuf);
}

void destroyStreamSessionAggOperatorInfo(void* param, int32_t numOfOutput) {
  SStreamSessionAggOperatorInfo* pInfo = (SStreamSessionAggOperatorInfo*)param;
2969
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
2970 2971
  destroyStreamAggSupporter(&pInfo->streamAggSup);
  cleanupGroupResInfo(&pInfo->groupResInfo);
2972 2973 2974
  if (pInfo->pChildren != NULL) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
X
Xiaoyu Wang 已提交
2975
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, i);
2976 2977 2978 2979 2980 2981
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
      destroyStreamSessionAggOperatorInfo(pChInfo, numOfOutput);
      taosMemoryFreeClear(pChild);
      taosMemoryFreeClear(pChInfo);
    }
  }
2982

D
dapan1121 已提交
2983
  taosMemoryFreeClear(param);
5
54liuyao 已提交
2984 2985
}

2986 2987
int32_t initBasicInfoEx(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfCols,
                        SSDataBlock* pResultBlock) {
2988 2989 2990 2991 2992
  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

2993
  initBasicInfo(pBasicInfo, pResultBlock);
2994

5
54liuyao 已提交
2995
  for (int32_t i = 0; i < numOfCols; ++i) {
2996
    pSup->pCtx[i].pBuf = NULL;
5
54liuyao 已提交
2997
  }
2998

2999
  ASSERT(numOfCols > 0);
3000
  increaseTs(pSup->pCtx);
5
54liuyao 已提交
3001 3002 3003 3004 3005 3006 3007 3008
  return TSDB_CODE_SUCCESS;
}

void initDummyFunction(SqlFunctionCtx* pDummy, SqlFunctionCtx* pCtx, int32_t nums) {
  for (int i = 0; i < nums; i++) {
    pDummy[i].functionId = pCtx[i].functionId;
  }
}
X
Xiaoyu Wang 已提交
3009 3010
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int64_t gap, int64_t waterMark,
                    uint8_t type) {
5
54liuyao 已提交
3011
  ASSERT(downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
3012
  SStreamScanInfo* pScanInfo = downstream->info;
X
Xiaoyu Wang 已提交
3013
  pScanInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = pAggSup, .gap = gap, .parentType = type};
5
54liuyao 已提交
3014
  pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, waterMark);
5
54liuyao 已提交
3015 3016
}

3017 3018
int32_t initSessionAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx,
                                int32_t numOfOutput) {
3019 3020 3021
  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SResultWindowInfo));
}

3022 3023 3024 3025 3026 3027 3028
SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                  SExecTaskInfo* pTaskInfo) {
  SSessionWinodwPhysiNode*       pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
  int32_t                        numOfCols = 0;
  SExprInfo*                     pExprInfo = createExprInfo(pSessionNode->window.pFuncs, NULL, &numOfCols);
  SSDataBlock*                   pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
  int32_t                        code = TSDB_CODE_OUT_OF_MEMORY;
X
Xiaoyu Wang 已提交
3029
  SStreamSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamSessionAggOperatorInfo));
3030
  SOperatorInfo*                 pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
3031 3032 3033 3034 3035
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
3036 3037 3038 3039 3040 3041 3042 3043
  if (pSessionNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pSessionNode->window.pExprs, NULL, &numOfScalar);
    int32_t    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }
3044
  SExprSupp* pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3045

3046
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
5
54liuyao 已提交
3047 3048 3049
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
3050

3051
  code = initSessionAggSupporter(&pInfo->streamAggSup, "StreamSessionAggOperatorInfo", pSup->pCtx, numOfCols);
5
54liuyao 已提交
3052 3053 3054
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
X
Xiaoyu Wang 已提交
3055

5
54liuyao 已提交
3056 3057 3058 3059
  pInfo->pDummyCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfCols, sizeof(SqlFunctionCtx));
  if (pInfo->pDummyCtx == NULL) {
    goto _error;
  }
3060
  initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols);
5
54liuyao 已提交
3061

3062 3063
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pSessionNode->window.watermark, .calTrigger = pSessionNode->window.triggerType, .maxTs = INT64_MIN};
H
Haojun Liao 已提交
3064 3065

  initResultRowInfo(&pInfo->binfo.resultRowInfo);
5
54liuyao 已提交
3066 3067
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

5
54liuyao 已提交
3068 3069 3070 3071
  pInfo->primaryTsIndex = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
  if (pSessionNode->window.pTsEnd) {
    pInfo->endTsIndex = ((SColumnNode*)pSessionNode->window.pTsEnd)->slotId;
  }
3072
  pInfo->gap = pSessionNode->gap;
5
54liuyao 已提交
3073 3074 3075 3076 3077
  pInfo->binfo.pRes = pResBlock;
  pInfo->order = TSDB_ORDER_ASC;
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  pInfo->pStDeleted = taosHashInit(64, hashFn, true, HASH_NO_LOCK);
  pInfo->pDelIterator = NULL;
5
54liuyao 已提交
3078
  // pInfo->pDelRes = createPullDataBlock();
3079 3080
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
3081
  pInfo->pChildren = NULL;
5
54liuyao 已提交
3082 3083
  pInfo->isFinal = false;
  pInfo->pPhyNode = pPhyNode;
5
54liuyao 已提交
3084
  pInfo->ignoreExpiredData = pSessionNode->window.igExpired;
5
54liuyao 已提交
3085
  pInfo->returnDelete = false;
5
54liuyao 已提交
3086 3087

  pOperator->name = "StreamSessionWindowAggOperator";
3088
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
5
54liuyao 已提交
3089 3090
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
3091 3092
  pOperator->exprSupp.pExprInfo = pExprInfo;
  pOperator->exprSupp.numOfExprs = numOfCols;
5
54liuyao 已提交
3093
  pOperator->info = pInfo;
X
Xiaoyu Wang 已提交
3094 3095 3096
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, NULL, destroyStreamSessionAggOperatorInfo,
                          aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
3097
  pOperator->pTaskInfo = pTaskInfo;
5
54liuyao 已提交
3098 3099 3100 3101
  if (downstream) {
    initDownStream(downstream, &pInfo->streamAggSup, pInfo->gap, pInfo->twAggSup.waterMark, pOperator->operatorType);
    code = appendDownstream(pOperator, &downstream, 1);
  }
5
54liuyao 已提交
3102 3103 3104 3105 3106 3107 3108 3109 3110 3111 3112 3113 3114 3115
  return pOperator;

_error:
  if (pInfo != NULL) {
    destroyStreamSessionAggOperatorInfo(pInfo, numOfCols);
  }

  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}

int64_t getSessionWindowEndkey(void* data, int32_t index) {
X
Xiaoyu Wang 已提交
3116
  SArray*            pWinInfos = (SArray*)data;
5
54liuyao 已提交
3117 3118 3119
  SResultWindowInfo* pWin = taosArrayGet(pWinInfos, index);
  return pWin->win.ekey;
}
5
54liuyao 已提交
3120 3121

bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap) {
5
54liuyao 已提交
3122
  if (ts + gap >= pWin->skey && ts - gap <= pWin->ekey) {
5
54liuyao 已提交
3123 3124 3125 3126 3127
    return true;
  }
  return false;
}

3128
bool isInWindow(SResultWindowInfo* pWinInfo, TSKEY ts, int64_t gap) { return isInTimeWindow(&pWinInfo->win, ts, gap); }
5
54liuyao 已提交
3129

X
Xiaoyu Wang 已提交
3130 3131
static SResultWindowInfo* insertNewSessionWindow(SArray* pWinInfos, TSKEY ts, int32_t index) {
  SResultWindowInfo win = {.pos.offset = -1, .pos.pageId = -1, .win.skey = ts, .win.ekey = ts, .isOutput = false};
5
54liuyao 已提交
3132 3133 3134 3135
  return taosArrayInsert(pWinInfos, index, &win);
}

static SResultWindowInfo* addNewSessionWindow(SArray* pWinInfos, TSKEY ts) {
X
Xiaoyu Wang 已提交
3136
  SResultWindowInfo win = {.pos.offset = -1, .pos.pageId = -1, .win.skey = ts, .win.ekey = ts, .isOutput = false};
5
54liuyao 已提交
3137 3138 3139
  return taosArrayPush(pWinInfos, &win);
}

3140
SArray* getWinInfos(SStreamAggSupporter* pAggSup, uint64_t groupId) {
3141
  void**  ite = taosHashGet(pAggSup->pResultRows, &groupId, sizeof(uint64_t));
3142 3143 3144
  SArray* pWinInfos = NULL;
  if (ite == NULL) {
    pWinInfos = taosArrayInit(1024, pAggSup->valueSize);
3145
    taosHashPut(pAggSup->pResultRows, &groupId, sizeof(uint64_t), &pWinInfos, sizeof(void*));
3146 3147 3148 3149 3150 3151
  } else {
    pWinInfos = *ite;
  }
  return pWinInfos;
}

5
54liuyao 已提交
3152 3153
// don't add new window
SResultWindowInfo* getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
3154
                                       int64_t gap, int32_t* pIndex) {
5
54liuyao 已提交
3155 3156 3157 3158 3159 3160 3161 3162
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;

  int32_t size = taosArrayGetSize(pWinInfos);
  if (size == 0) {
    return NULL;
  }
  // find the first position which is smaller than the key
3163
  int32_t            index = binarySearch(pWinInfos, size, startTs, TSDB_ORDER_DESC, getSessionWindowEndkey);
5
54liuyao 已提交
3164 3165 3166 3167 3168 3169 3170 3171 3172 3173 3174 3175 3176 3177 3178
  SResultWindowInfo* pWin = NULL;
  if (index >= 0) {
    pWin = taosArrayGet(pWinInfos, index);
    if (isInWindow(pWin, startTs, gap)) {
      *pIndex = index;
      return pWin;
    }
  }

  if (index + 1 < size) {
    pWin = taosArrayGet(pWinInfos, index + 1);
    if (isInWindow(pWin, startTs, gap)) {
      *pIndex = index + 1;
      return pWin;
    } else if (endTs != INT64_MIN && isInWindow(pWin, endTs, gap)) {
3179
      *pIndex = index + 1;
5
54liuyao 已提交
3180 3181 3182 3183 3184 3185 3186
      return pWin;
    }
  }

  return NULL;
}

3187 3188
SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
                                        int64_t gap, int32_t* pIndex) {
3189 3190 3191
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;

5
54liuyao 已提交
3192 3193
  int32_t size = taosArrayGetSize(pWinInfos);
  if (size == 0) {
5
54liuyao 已提交
3194
    *pIndex = 0;
5
54liuyao 已提交
3195
    return addNewSessionWindow(pWinInfos, startTs);
5
54liuyao 已提交
3196 3197
  }
  // find the first position which is smaller than the key
5
54liuyao 已提交
3198
  int32_t            index = binarySearch(pWinInfos, size, startTs, TSDB_ORDER_DESC, getSessionWindowEndkey);
5
54liuyao 已提交
3199 3200 3201
  SResultWindowInfo* pWin = NULL;
  if (index >= 0) {
    pWin = taosArrayGet(pWinInfos, index);
5
54liuyao 已提交
3202
    if (isInWindow(pWin, startTs, gap)) {
5
54liuyao 已提交
3203 3204 3205 3206 3207 3208 3209
      *pIndex = index;
      return pWin;
    }
  }

  if (index + 1 < size) {
    pWin = taosArrayGet(pWinInfos, index + 1);
5
54liuyao 已提交
3210
    if (isInWindow(pWin, startTs, gap)) {
5
54liuyao 已提交
3211 3212
      *pIndex = index + 1;
      return pWin;
5
54liuyao 已提交
3213 3214 3215
    } else if (endTs != INT64_MIN && isInWindow(pWin, endTs, gap)) {
      *pIndex = index;
      return pWin;
5
54liuyao 已提交
3216 3217 3218 3219 3220
    }
  }

  if (index == size - 1) {
    *pIndex = taosArrayGetSize(pWinInfos);
5
54liuyao 已提交
3221
    return addNewSessionWindow(pWinInfos, startTs);
5
54liuyao 已提交
3222
  }
5
54liuyao 已提交
3223
  *pIndex = index + 1;
5
54liuyao 已提交
3224
  return insertNewSessionWindow(pWinInfos, startTs, index + 1);
5
54liuyao 已提交
3225 3226
}

3227 3228
int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs, TSKEY* pEndTs, int32_t rows,
                                int32_t start, int64_t gap, SHashObj* pStDeleted) {
5
54liuyao 已提交
3229
  for (int32_t i = start; i < rows; ++i) {
3230
    if (!isInWindow(pWinInfo, pStartTs[i], gap) && (!pEndTs || !isInWindow(pWinInfo, pEndTs[i], gap))) {
5
54liuyao 已提交
3231 3232
      return i - start;
    }
5
54liuyao 已提交
3233
    if (pWinInfo->win.skey > pStartTs[i]) {
5
54liuyao 已提交
3234 3235 3236 3237
      if (pStDeleted && pWinInfo->isOutput) {
        taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY));
        pWinInfo->isOutput = false;
      }
5
54liuyao 已提交
3238 3239 3240 3241 3242
      pWinInfo->win.skey = pStartTs[i];
    }
    pWinInfo->win.ekey = TMAX(pWinInfo->win.ekey, pStartTs[i]);
    if (pEndTs) {
      pWinInfo->win.ekey = TMAX(pWinInfo->win.ekey, pEndTs[i]);
5
54liuyao 已提交
3243 3244 3245 3246 3247
    }
  }
  return rows - start;
}

X
Xiaoyu Wang 已提交
3248
static int32_t setWindowOutputBuf(SResultWindowInfo* pWinInfo, SResultRow** pResult, SqlFunctionCtx* pCtx,
3249
                                  uint64_t groupId, int32_t numOfOutput, int32_t* rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3250
                                  SStreamAggSupporter* pAggSup, SExecTaskInfo* pTaskInfo) {
5
54liuyao 已提交
3251 3252
  assert(pWinInfo->win.skey <= pWinInfo->win.ekey);
  // too many time window in query
3253
  int32_t size = taosArrayGetSize(pAggSup->pCurWins);
5
54liuyao 已提交
3254 3255 3256
  if (size > MAX_INTERVAL_TIME_WINDOW) {
    longjmp(pTaskInfo->env, TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW);
  }
X
Xiaoyu Wang 已提交
3257

5
54liuyao 已提交
3258
  if (pWinInfo->pos.pageId == -1) {
3259
    *pResult = getNewResultRow(pAggSup->pResultBuf, groupId, pAggSup->resultRowSize);
5
54liuyao 已提交
3260 3261 3262 3263 3264 3265 3266 3267 3268 3269 3270 3271 3272 3273 3274 3275 3276 3277
    if (*pResult == NULL) {
      return TSDB_CODE_OUT_OF_MEMORY;
    }
    initResultRow(*pResult);

    // add a new result set for a new group
    pWinInfo->pos.pageId = (*pResult)->pageId;
    pWinInfo->pos.offset = (*pResult)->offset;
  } else {
    *pResult = getResultRowByPos(pAggSup->pResultBuf, &pWinInfo->pos);
    if (!(*pResult)) {
      qError("getResultRowByPos return NULL, TID:%s", GET_TASKID(pTaskInfo));
      return TSDB_CODE_FAILED;
    }
  }

  // set time window for current result
  (*pResult)->win = pWinInfo->win;
3278
  setResultRowInitCtx(*pResult, pCtx, numOfOutput, rowEntryInfoOffset);
5
54liuyao 已提交
3279 3280 3281
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
3282 3283 3284
static int32_t doOneWindowAggImpl(int32_t tsColId, SOptrBasicInfo* pBinfo, SStreamAggSupporter* pAggSup,
                                  SColumnInfoData* pTimeWindowData, SSDataBlock* pSDataBlock,
                                  SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex, int32_t winRows,
3285
                                  int32_t numOutput, SOperatorInfo* pOperator) {
3286
  SExprSupp*     pSup = &pOperator->exprSupp;
3287 3288
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

X
Xiaoyu Wang 已提交
3289 3290
  SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, tsColId);
  TSKEY*           tsCols = (int64_t*)pColDataInfo->pData;
3291 3292
  int32_t          code = setWindowOutputBuf(pCurWin, pResult, pSup->pCtx, pSDataBlock->info.groupId, numOutput,
                                             pSup->rowEntryInfoOffset, pAggSup, pTaskInfo);
5
54liuyao 已提交
3293 3294 3295
  if (code != TSDB_CODE_SUCCESS || (*pResult) == NULL) {
    return TSDB_CODE_QRY_OUT_OF_MEMORY;
  }
5
54liuyao 已提交
3296
  updateTimeWindowInfo(pTimeWindowData, &pCurWin->win, false);
3297
  doApplyFunctions(pTaskInfo, pSup->pCtx, &pCurWin->win, pTimeWindowData, startIndex, winRows, tsCols,
X
Xiaoyu Wang 已提交
3298
                   pSDataBlock->info.rows, numOutput, TSDB_ORDER_ASC);
5
54liuyao 已提交
3299 3300 3301
  SFilePage* bufPage = getBufPage(pAggSup->pResultBuf, pCurWin->pos.pageId);
  setBufPageDirty(bufPage, true);
  releaseBufPage(pAggSup->pResultBuf, bufPage);
5
54liuyao 已提交
3302 3303 3304
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
3305 3306
static int32_t doOneWindowAgg(SStreamSessionAggOperatorInfo* pInfo, SSDataBlock* pSDataBlock,
                              SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex, int32_t winRows,
3307
                              int32_t numOutput, SOperatorInfo* pOperator) {
X
Xiaoyu Wang 已提交
3308
  return doOneWindowAggImpl(pInfo->primaryTsIndex, &pInfo->binfo, &pInfo->streamAggSup, &pInfo->twAggSup.timeWindowData,
3309
                            pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pOperator);
5
54liuyao 已提交
3310 3311
}

X
Xiaoyu Wang 已提交
3312 3313
static int32_t doOneStateWindowAgg(SStreamStateAggOperatorInfo* pInfo, SSDataBlock* pSDataBlock,
                                   SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex,
3314
                                   int32_t winRows, int32_t numOutput, SOperatorInfo* pOperator) {
X
Xiaoyu Wang 已提交
3315
  return doOneWindowAggImpl(pInfo->primaryTsIndex, &pInfo->binfo, &pInfo->streamAggSup, &pInfo->twAggSup.timeWindowData,
3316
                            pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pOperator);
5
54liuyao 已提交
3317 3318
}

5
54liuyao 已提交
3319 3320
int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) {
  SResultWindowInfo* pCurWin = taosArrayGet(pWinInfos, startIndex);
X
Xiaoyu Wang 已提交
3321
  int32_t            size = taosArrayGetSize(pWinInfos);
5
54liuyao 已提交
3322 3323 3324 3325 3326 3327 3328 3329 3330 3331 3332
  // Just look for the window behind StartIndex
  for (int32_t i = startIndex + 1; i < size; i++) {
    SResultWindowInfo* pWinInfo = taosArrayGet(pWinInfos, i);
    if (!isInWindow(pCurWin, pWinInfo->win.skey, gap)) {
      return i - startIndex - 1;
    }
  }

  return size - startIndex - 1;
}

5
54liuyao 已提交
3333
void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, uint64_t groupId,
3334
                       int32_t numOfOutput, SHashObj* pStUpdated, SHashObj* pStDeleted, SOperatorInfo* pOperator) {
3335
  SExprSupp*     pSup = &pOperator->exprSupp;
3336 3337
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

3338
  SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pCurWins, startIndex);
X
Xiaoyu Wang 已提交
3339
  SResultRow*        pCurResult = NULL;
3340
  setWindowOutputBuf(pCurWin, &pCurResult, pSup->pCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3341
                     &pInfo->streamAggSup, pTaskInfo);
5
54liuyao 已提交
3342
  num += startIndex + 1;
3343
  ASSERT(num <= taosArrayGetSize(pInfo->streamAggSup.pCurWins));
5
54liuyao 已提交
3344 3345
  // Just look for the window behind StartIndex
  for (int32_t i = startIndex + 1; i < num; i++) {
3346
    SResultWindowInfo* pWinInfo = taosArrayGet(pInfo->streamAggSup.pCurWins, i);
X
Xiaoyu Wang 已提交
3347
    SResultRow*        pWinResult = NULL;
3348
    setWindowOutputBuf(pWinInfo, &pWinResult, pInfo->pDummyCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3349
                       &pInfo->streamAggSup, pTaskInfo);
5
54liuyao 已提交
3350
    pCurWin->win.ekey = TMAX(pCurWin->win.ekey, pWinInfo->win.ekey);
3351
    compactFunctions(pSup->pCtx, pInfo->pDummyCtx, numOfOutput, pTaskInfo);
5
54liuyao 已提交
3352 3353 3354 3355 3356
    taosHashRemove(pStUpdated, &pWinInfo->pos, sizeof(SResultRowPosition));
    if (pWinInfo->isOutput) {
      taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY));
      pWinInfo->isOutput = false;
    }
3357
    taosArrayRemove(pInfo->streamAggSup.pCurWins, i);
5
54liuyao 已提交
3358 3359
    SFilePage* tmpPage = getBufPage(pInfo->streamAggSup.pResultBuf, pWinInfo->pos.pageId);
    releaseBufPage(pInfo->streamAggSup.pResultBuf, tmpPage);
5
54liuyao 已提交
3360
  }
5
54liuyao 已提交
3361 3362 3363 3364
  SFilePage* bufPage = getBufPage(pInfo->streamAggSup.pResultBuf, pCurWin->pos.pageId);
  ASSERT(num > 0);
  setBufPageDirty(bufPage, true);
  releaseBufPage(pInfo->streamAggSup.pResultBuf, bufPage);
5
54liuyao 已提交
3365 3366
}

3367 3368
static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SHashObj* pStUpdated,
                                   SHashObj* pStDeleted, bool hasEndTs) {
X
Xiaoyu Wang 已提交
3369
  SExecTaskInfo*                 pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
3370
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
3371
  bool                           masterScan = true;
3372
  int32_t                        numOfOutput = pOperator->exprSupp.numOfExprs;
5
54liuyao 已提交
3373
  uint64_t                       groupId = pSDataBlock->info.groupId;
X
Xiaoyu Wang 已提交
3374 3375 3376 3377 3378
  int64_t                        gap = pInfo->gap;
  int64_t                        code = TSDB_CODE_SUCCESS;

  int32_t     step = 1;
  bool        ascScan = true;
5
54liuyao 已提交
3379 3380
  TSKEY*      startTsCols = NULL;
  TSKEY*      endTsCols = NULL;
5
54liuyao 已提交
3381
  SResultRow* pResult = NULL;
X
Xiaoyu Wang 已提交
3382
  int32_t     winRows = 0;
5
54liuyao 已提交
3383

5
54liuyao 已提交
3384 3385 3386 3387 3388 3389
  ASSERT(pSDataBlock->pDataBlock);
  SColumnInfoData* pStartTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
  startTsCols = (int64_t*)pStartTsCol->pData;
  SColumnInfoData* pEndTsCol = NULL;
  if (hasEndTs) {
    pEndTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->endTsIndex);
5
54liuyao 已提交
3390
  } else {
5
54liuyao 已提交
3391
    pEndTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
3392
  }
5
54liuyao 已提交
3393
  endTsCols = (int64_t*)pEndTsCol->pData;
X
Xiaoyu Wang 已提交
3394

5
54liuyao 已提交
3395
  SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
X
Xiaoyu Wang 已提交
3396
  for (int32_t i = 0; i < pSDataBlock->info.rows;) {
5
54liuyao 已提交
3397
    if (pInfo->ignoreExpiredData && isOverdue(endTsCols[i], &pInfo->twAggSup)) {
5
54liuyao 已提交
3398 3399 3400
      i++;
      continue;
    }
X
Xiaoyu Wang 已提交
3401
    int32_t            winIndex = 0;
3402 3403 3404
    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, startTsCols[i], endTsCols[i], groupId, gap, &winIndex);
    winRows =
        updateSessionWindowInfo(pCurWin, startTsCols, endTsCols, pSDataBlock->info.rows, i, pInfo->gap, pStDeleted);
3405
    code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pOperator);
5
54liuyao 已提交
3406 3407 3408
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
5
54liuyao 已提交
3409

3410
    int32_t winNum = getNumCompactWindow(pAggSup->pCurWins, winIndex, gap);
5
54liuyao 已提交
3411
    if (winNum > 0) {
3412
      compactTimeWindow(pInfo, winIndex, winNum, groupId, numOfOutput, pStUpdated, pStDeleted, pOperator);
5
54liuyao 已提交
3413
    }
5
54liuyao 已提交
3414
    pCurWin->isClosed = false;
5
54liuyao 已提交
3415
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pStUpdated) {
5
54liuyao 已提交
3416 3417
      SWinRes value = {.ts = pCurWin->win.skey, .groupId = groupId};
      code = taosHashPut(pStUpdated, &pCurWin->pos, sizeof(SResultRowPosition), &value, sizeof(SWinRes));
5
54liuyao 已提交
3418 3419 3420 3421
      if (code != TSDB_CODE_SUCCESS) {
        longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
      }
      pCurWin->isOutput = true;
5
54liuyao 已提交
3422 3423 3424 3425 3426
    }
    i += winRows;
  }
}

5
54liuyao 已提交
3427 3428 3429 3430 3431
void deleteWindow(SArray* pWinInfos, int32_t index) {
  ASSERT(index >= 0 && index < taosArrayGetSize(pWinInfos));
  taosArrayRemove(pWinInfos, index);
}

3432
static void doDeleteTimeWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, int64_t gap, SArray* result) {
5
54liuyao 已提交
3433
  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
3434
  TSKEY*           startDatas = (TSKEY*)pStartTsCol->pData;
5
54liuyao 已提交
3435
  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
3436
  TSKEY*           endDatas = (TSKEY*)pEndTsCol->pData;
5
54liuyao 已提交
3437
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
3438
  uint64_t*        gpDatas = (uint64_t*)pGroupCol->pData;
5
54liuyao 已提交
3439 3440
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    int32_t winIndex = 0;
3441 3442
    while (1) {
      SResultWindowInfo* pCurWin = getCurSessionWindow(pAggSup, startDatas[i], endDatas[i], gpDatas[i], gap, &winIndex);
5
54liuyao 已提交
3443 3444 3445 3446 3447 3448 3449 3450 3451 3452 3453
      if (!pCurWin) {
        break;
      }
      deleteWindow(pAggSup->pCurWins, winIndex);
      if (result) {
        taosArrayPush(result, pCurWin);
      }
    }
  }
}

3454 3455
static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SExprSupp* pSup, SSDataBlock* pBlock, int32_t tsIndex,
                                  int32_t numOfOutput, int64_t gap, SArray* result) {
5
54liuyao 已提交
3456
  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, tsIndex);
X
Xiaoyu Wang 已提交
3457 3458
  TSKEY*           tsCols = (TSKEY*)pColDataInfo->pData;
  int32_t          step = 0;
5
54liuyao 已提交
3459
  for (int32_t i = 0; i < pBlock->info.rows; i += step) {
X
Xiaoyu Wang 已提交
3460
    int32_t            winIndex = 0;
3461
    SResultWindowInfo* pCurWin =
5
54liuyao 已提交
3462 3463
        getCurSessionWindow(pAggSup, tsCols[i], INT64_MIN, pBlock->info.groupId, gap, &winIndex);
    if (!pCurWin || pCurWin->pos.pageId == -1) {
5
54liuyao 已提交
3464
      // window has been closed.
5
54liuyao 已提交
3465
      step = 1;
5
54liuyao 已提交
3466 3467
      continue;
    }
5
54liuyao 已提交
3468 3469
    step = updateSessionWindowInfo(pCurWin, tsCols, NULL, pBlock->info.rows, i, gap, NULL);
    ASSERT(isInWindow(pCurWin, tsCols[i], gap));
3470
    doClearWindowImpl(&pCurWin->pos, pAggSup->pResultBuf, pSup, numOfOutput);
3471 3472 3473
    if (result) {
      taosArrayPush(result, pCurWin);
    }
5
54liuyao 已提交
3474 3475 3476
  }
}

5
54liuyao 已提交
3477
static int32_t copyUpdateResult(SHashObj* pStUpdated, SArray* pUpdated) {
X
Xiaoyu Wang 已提交
3478
  void*  pData = NULL;
5
54liuyao 已提交
3479
  size_t keyLen = 0;
X
Xiaoyu Wang 已提交
3480
  while ((pData = taosHashIterate(pStUpdated, pData)) != NULL) {
5
54liuyao 已提交
3481 3482 3483 3484 3485 3486
    void* key = taosHashGetKey(pData, &keyLen);
    ASSERT(keyLen == sizeof(SResultRowPosition));
    SResKeyPos* pos = taosMemoryMalloc(sizeof(SResKeyPos) + sizeof(uint64_t));
    if (pos == NULL) {
      return TSDB_CODE_QRY_OUT_OF_MEMORY;
    }
5
54liuyao 已提交
3487
    pos->groupId = ((SWinRes*)pData)->groupId;
5
54liuyao 已提交
3488
    pos->pos = *(SResultRowPosition*)key;
5
54liuyao 已提交
3489
    *(int64_t*)pos->key = ((SWinRes*)pData)->ts;
5
54liuyao 已提交
3490 3491
    taosArrayPush(pUpdated, &pos);
  }
5
54liuyao 已提交
3492
  taosArraySort(pUpdated, resultrowComparAsc);
5
54liuyao 已提交
3493 3494 3495 3496 3497
  return TSDB_CODE_SUCCESS;
}

void doBuildDeleteDataBlock(SHashObj* pStDeleted, SSDataBlock* pBlock, void** Ite) {
  blockDataCleanup(pBlock);
3498 3499 3500 3501 3502
  int32_t size = taosHashGetSize(pStDeleted);
  if (size == 0) {
    return;
  }
  blockDataEnsureCapacity(pBlock, size);
5
54liuyao 已提交
3503
  size_t keyLen = 0;
X
Xiaoyu Wang 已提交
3504
  while (((*Ite) = taosHashIterate(pStDeleted, *Ite)) != NULL) {
5
54liuyao 已提交
3505
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
X
Xiaoyu Wang 已提交
3506
    colDataAppend(pColInfoData, pBlock->info.rows, *Ite, false);
3507
    for (int32_t i = 1; i < taosArrayGetSize(pBlock->pDataBlock); i++) {
5
54liuyao 已提交
3508
      pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
X
Xiaoyu Wang 已提交
3509
      colDataAppendNULL(pColInfoData, pBlock->info.rows);
5
54liuyao 已提交
3510 3511 3512 3513 3514 3515 3516 3517 3518 3519 3520
    }
    pBlock->info.rows += 1;
    if (pBlock->info.rows + 1 >= pBlock->info.capacity) {
      break;
    }
  }
  if ((*Ite) == NULL) {
    taosHashClear(pStDeleted);
  }
}

X
Xiaoyu Wang 已提交
3521
static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWinArray, int32_t groupId,
3522
                              int32_t numOfOutput, SOperatorInfo* pOperator) {
3523
  SExprSupp*     pSup = &pOperator->exprSupp;
3524 3525
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

3526 3527
  int32_t size = taosArrayGetSize(pWinArray);
  ASSERT(pInfo->pChildren);
3528

3529 3530
  for (int32_t i = 0; i < size; i++) {
    SResultWindowInfo* pParentWin = taosArrayGet(pWinArray, i);
X
Xiaoyu Wang 已提交
3531
    SResultRow*        pCurResult = NULL;
3532
    setWindowOutputBuf(pParentWin, &pCurResult, pSup->pCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3533
                       &pInfo->streamAggSup, pTaskInfo);
3534 3535
    int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
    for (int32_t j = 0; j < numOfChildren; j++) {
X
Xiaoyu Wang 已提交
3536
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, j);
3537
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
3538
      SArray*                        pChWins = getWinInfos(&pChInfo->streamAggSup, groupId);
X
Xiaoyu Wang 已提交
3539 3540
      int32_t                        chWinSize = taosArrayGetSize(pChWins);
      int32_t index = binarySearch(pChWins, chWinSize, pParentWin->win.skey, TSDB_ORDER_DESC, getSessionWindowEndkey);
5
54liuyao 已提交
3541
      if (index < 0) {
3542
        index = 0;
5
54liuyao 已提交
3543 3544
      }
      for (int32_t k = index; k < chWinSize; k++) {
5
54liuyao 已提交
3545 3546
        SResultWindowInfo* pChWin = taosArrayGet(pChWins, k);
        if (pParentWin->win.skey <= pChWin->win.skey && pChWin->win.ekey <= pParentWin->win.ekey) {
3547
          SResultRow* pChResult = NULL;
5
54liuyao 已提交
3548
          setWindowOutputBuf(pChWin, &pChResult, pChild->exprSupp.pCtx, groupId, numOfOutput,
3549 3550
                             pChild->exprSupp.rowEntryInfoOffset, &pChInfo->streamAggSup, pTaskInfo);
          compactFunctions(pSup->pCtx, pChild->exprSupp.pCtx, numOfOutput, pTaskInfo);
5
54liuyao 已提交
3551 3552
          SFilePage* bufPage = getBufPage(pChInfo->streamAggSup.pResultBuf, pChWin->pos.pageId);
          releaseBufPage(pChInfo->streamAggSup.pResultBuf, bufPage);
3553
          continue;
5
54liuyao 已提交
3554 3555
        } else if (!pChWin->isClosed) {
          break;
3556 3557 3558
        }
      }
    }
5
54liuyao 已提交
3559 3560 3561 3562
    SFilePage* bufPage = getBufPage(pInfo->streamAggSup.pResultBuf, pParentWin->pos.pageId);
    ASSERT(size > 0);
    setBufPageDirty(bufPage, true);
    releaseBufPage(pInfo->streamAggSup.pResultBuf, bufPage);
3563 3564 3565
  }
}

X
Xiaoyu Wang 已提交
3566
typedef SResultWindowInfo* (*__get_win_info_)(void*);
5
54liuyao 已提交
3567 3568
SResultWindowInfo* getResWinForSession(void* pData) { return (SResultWindowInfo*)pData; }
SResultWindowInfo* getResWinForState(void* pData) { return &((SStateWindowInfo*)pData)->winInfo; }
5
54liuyao 已提交
3569

3570 3571
int32_t closeSessionWindow(SHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SArray* pClosed, __get_win_info_ fn,
                           bool delete) {
5
54liuyao 已提交
3572
  // Todo(liuyao) save window to tdb
3573
  void** pIte = NULL;
5
54liuyao 已提交
3574
  size_t keyLen = 0;
3575
  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
5
54liuyao 已提交
3576
    uint64_t* pGroupId = taosHashGetKey(pIte, &keyLen);
3577 3578
    SArray*   pWins = (SArray*)(*pIte);
    int32_t   size = taosArrayGetSize(pWins);
3579 3580 3581
    for (int32_t i = 0; i < size; i++) {
      void*              pWin = taosArrayGet(pWins, i);
      SResultWindowInfo* pSeWin = fn(pWin);
5
54liuyao 已提交
3582
      if (isCloseWindow(&pSeWin->win, pTwSup)) {
3583 3584
        if (!pSeWin->isClosed) {
          pSeWin->isClosed = true;
5
54liuyao 已提交
3585
          if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE && pClosed) {
5
54liuyao 已提交
3586
            int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, *pGroupId, pClosed);
5
54liuyao 已提交
3587 3588 3589
            if (code != TSDB_CODE_SUCCESS) {
              return code;
            }
3590 3591
            pSeWin->isOutput = true;
          }
5
54liuyao 已提交
3592
          if (delete) {
5
54liuyao 已提交
3593
            deleteWindow(pWins, i);
5
54liuyao 已提交
3594 3595 3596
            i--;
            size = taosArrayGetSize(pWins);
          }
5
54liuyao 已提交
3597
        }
3598
        continue;
5
54liuyao 已提交
3599
      }
3600
      break;
5
54liuyao 已提交
3601 3602 3603 3604 3605
    }
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3606 3607 3608 3609 3610 3611 3612 3613 3614 3615
static void closeChildSessionWindow(SArray* pChildren, TSKEY maxTs, bool delete) {
  int32_t size = taosArrayGetSize(pChildren);
  for (int32_t i = 0; i < size; i++) {
    SOperatorInfo*                 pChildOp = taosArrayGetP(pChildren, i);
    SStreamSessionAggOperatorInfo* pChInfo = pChildOp->info;
    pChInfo->twAggSup.maxTs = TMAX(pChInfo->twAggSup.maxTs, maxTs);
    closeSessionWindow(pChInfo->streamAggSup.pResultRows, &pChInfo->twAggSup, NULL, getResWinForSession, delete);
  }
}

3616
int32_t getAllSessionWindow(SHashObj* pHashMap, SArray* pClosed, __get_win_info_ fn) {
3617
  void** pIte = NULL;
3618
  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
3619
    SArray* pWins = (SArray*)(*pIte);
3620 3621 3622 3623 3624 3625 3626 3627
    int32_t size = taosArrayGetSize(pWins);
    for (int32_t i = 0; i < size; i++) {
      void*              pWin = taosArrayGet(pWins, i);
      SResultWindowInfo* pSeWin = fn(pWin);
      if (!pSeWin->isClosed) {
        int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, 0, pClosed);
        pSeWin->isOutput = true;
      }
5
54liuyao 已提交
3628 3629 3630 3631 3632
    }
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3633 3634 3635 3636 3637 3638 3639 3640
static void copyDeleteWindowInfo(SArray* pResWins, SHashObj* pStDeleted) {
  int32_t size = taosArrayGetSize(pResWins);
  for (int32_t i = 0; i < size; i++) {
    SResultWindowInfo* pWinInfo = taosArrayGet(pResWins, i);
    taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY));
  }
}

5
54liuyao 已提交
3641
static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
5
54liuyao 已提交
3642
  SExprSupp*                     pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3643
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
3644
  SOptrBasicInfo*                pBInfo = &pInfo->binfo;
5
54liuyao 已提交
3645 3646 3647 3648
  TSKEY                          maxTs = INT64_MIN;
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
5
54liuyao 已提交
3649 3650
    doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
    if (pInfo->pDelRes->info.rows > 0) {
L
Liu Jicong 已提交
3651
      printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3652 3653
      return pInfo->pDelRes;
    }
X
Xiaoyu Wang 已提交
3654
    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
3655
    if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) {
5
54liuyao 已提交
3656 3657
      doSetOperatorCompleted(pOperator);
    }
L
Liu Jicong 已提交
3658
    printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3659 3660 3661
    return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
  }

X
Xiaoyu Wang 已提交
3662 3663
  _hash_fn_t     hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  SHashObj*      pStUpdated = taosHashInit(64, hashFn, true, HASH_NO_LOCK);
5
54liuyao 已提交
3664
  SOperatorInfo* downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
3665
  SArray*        pUpdated = taosArrayInit(16, POINTER_BYTES);
5
54liuyao 已提交
3666 3667 3668 3669 3670
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
      break;
    }
5
54liuyao 已提交
3671
    printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "Final  Session Recv" : "Single Session Recv");
3672

5
54liuyao 已提交
3673
    if (pBlock->info.type == STREAM_CLEAR) {
X
Xiaoyu Wang 已提交
3674
      SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
3675 3676
      doClearSessionWindows(&pInfo->streamAggSup, &pOperator->exprSupp, pBlock, 0, pOperator->exprSupp.numOfExprs, 0,
                            pWins);
5
54liuyao 已提交
3677 3678
      if (IS_FINAL_OP(pInfo)) {
        int32_t                        childIndex = getChildIndex(pBlock);
X
Xiaoyu Wang 已提交
3679
        SOperatorInfo*                 pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
3680
        SStreamSessionAggOperatorInfo* pChildInfo = pChildOp->info;
3681
        doClearSessionWindows(&pChildInfo->streamAggSup, &pChildOp->exprSupp, pBlock, 0, pChildOp->exprSupp.numOfExprs,
5
54liuyao 已提交
3682
                              0, NULL);
3683
        rebuildTimeWindow(pInfo, pWins, pBlock->info.groupId, pOperator->exprSupp.numOfExprs, pOperator);
3684 3685
      }
      taosArrayDestroy(pWins);
5
54liuyao 已提交
3686
      continue;
5
54liuyao 已提交
3687 3688 3689
    } else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT) {
      SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
      // gap must be 0
3690
      doDeleteTimeWindows(&pInfo->streamAggSup, pBlock, 0, pWins);
5
54liuyao 已提交
3691 3692 3693 3694 3695
      if (IS_FINAL_OP(pInfo)) {
        int32_t                        childIndex = getChildIndex(pBlock);
        SOperatorInfo*                 pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
        SStreamSessionAggOperatorInfo* pChildInfo = pChildOp->info;
        // gap must be 0
3696
        doDeleteTimeWindows(&pChildInfo->streamAggSup, pBlock, 0, NULL);
5
54liuyao 已提交
3697 3698 3699 3700 3701
        rebuildTimeWindow(pInfo, pWins, pBlock->info.groupId, pOperator->exprSupp.numOfExprs, pOperator);
      }
      copyDeleteWindowInfo(pWins, pInfo->pStDeleted);
      taosArrayDestroy(pWins);
      continue;
3702
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
3703
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getResWinForSession);
5
54liuyao 已提交
3704
      continue;
5
54liuyao 已提交
3705
    }
5
54liuyao 已提交
3706

5
54liuyao 已提交
3707 3708 3709 3710
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
3711
    // the pDataBlock are always the same one, no need to call this again
3712
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
3713 3714 3715 3716 3717 3718
    doStreamSessionAggImpl(pOperator, pBlock, pStUpdated, pInfo->pStDeleted, IS_FINAL_OP(pInfo));
    if (IS_FINAL_OP(pInfo)) {
      int32_t chIndex = getChildIndex(pBlock);
      int32_t size = taosArrayGetSize(pInfo->pChildren);
      // if chIndex + 1 - size > 0, add new child
      for (int32_t i = 0; i < chIndex + 1 - size; i++) {
3719 3720
        SOperatorInfo* pChildOp =
            createStreamFinalSessionAggOperatorInfo(NULL, pInfo->pPhyNode, pOperator->pTaskInfo, 0);
5
54liuyao 已提交
3721 3722 3723 3724 3725
        if (!pChildOp) {
          longjmp(pOperator->pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
        }
        taosArrayPush(pInfo->pChildren, &pChildOp);
      }
3726
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
5
54liuyao 已提交
3727 3728
      setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
      doStreamSessionAggImpl(pChildOp, pBlock, NULL, NULL, true);
3729
    }
5
54liuyao 已提交
3730
    maxTs = TMAX(maxTs, pBlock->info.window.ekey);
5
54liuyao 已提交
3731
  }
5
54liuyao 已提交
3732 3733

  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
5
54liuyao 已提交
3734 3735
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
H
Haojun Liao 已提交
3736

3737 3738
  closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated, getResWinForSession,
                     pInfo->ignoreExpiredData);
5
54liuyao 已提交
3739
  closeChildSessionWindow(pInfo->pChildren, pInfo->twAggSup.maxTs, pInfo->ignoreExpiredData);
5
54liuyao 已提交
3740 3741 3742
  copyUpdateResult(pStUpdated, pUpdated);
  taosHashCleanup(pStUpdated);

3743
  finalizeUpdatedResult(pSup->numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
3744 3745 3746 3747
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
  doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
  if (pInfo->pDelRes->info.rows > 0) {
L
Liu Jicong 已提交
3748
    printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3749 3750 3751
    return pInfo->pDelRes;
  }
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
L
Liu Jicong 已提交
3752
  printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3753 3754 3755 3756
  return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
}

static void clearStreamSessionOperator(SStreamSessionAggOperatorInfo* pInfo) {
3757
  void** pIte = NULL;
5
54liuyao 已提交
3758
  while ((pIte = taosHashIterate(pInfo->streamAggSup.pResultRows, pIte)) != NULL) {
3759
    SArray* pWins = (SArray*)(*pIte);
5
54liuyao 已提交
3760 3761 3762 3763 3764 3765 3766 3767 3768 3769 3770 3771 3772 3773 3774 3775 3776 3777 3778 3779 3780 3781 3782 3783 3784
    int32_t size = taosArrayGetSize(pWins);
    for (int32_t i = 0; i < size; i++) {
      SResultWindowInfo* pWin = (SResultWindowInfo*)taosArrayGet(pWins, i);
      pWin->pos.pageId = -1;
      pWin->pos.offset = -1;
    }
  }
  clearDiskbasedBuf(pInfo->streamAggSup.pResultBuf);
  cleanupResultRowInfo(&pInfo->binfo.resultRowInfo);
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
}

static void removeSessionResults(SHashObj* pHashMap, SArray* pWins) {
  int32_t size = taosArrayGetSize(pWins);
  for (int32_t i = 0; i < size; i++) {
    SResultWindowInfo* pWin = taosArrayGet(pWins, i);
    taosHashRemove(pHashMap, &pWin->pos, sizeof(SResultRowPosition));
  }
}

static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
  SOptrBasicInfo*                pBInfo = &pInfo->binfo;
  TSKEY                          maxTs = INT64_MIN;
  SExprSupp*                     pSup = &pOperator->exprSupp;
3785

5
54liuyao 已提交
3786 3787 3788
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
5
54liuyao 已提交
3789 3790 3791 3792 3793 3794 3795 3796 3797
    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
    if (pBInfo->pRes->info.rows > 0) {
      printDataBlock(pBInfo->pRes, "Semi  Session");
      return pBInfo->pRes;
    }

    // doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
    if (pInfo->pDelRes->info.rows > 0 && !pInfo->returnDelete) {
      pInfo->returnDelete = true;
5
54liuyao 已提交
3798
      printDataBlock(pInfo->pDelRes, "Semi  Session");
5
54liuyao 已提交
3799 3800
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
3801 3802

    if (pInfo->pUpdateRes->info.rows > 0) {
5
54liuyao 已提交
3803 3804
      // process the rest of the data
      pOperator->status = OP_OPENED;
5
54liuyao 已提交
3805
      printDataBlock(pInfo->pUpdateRes, "Semi  Session");
5
54liuyao 已提交
3806 3807
      return pInfo->pUpdateRes;
    }
5
54liuyao 已提交
3808 3809 3810 3811
    // semi interval operator clear disk buffer
    clearStreamSessionOperator(pInfo);
    pOperator->status = OP_EXEC_DONE;
    return NULL;
5
54liuyao 已提交
3812 3813 3814 3815 3816 3817 3818 3819 3820
  }

  _hash_fn_t     hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  SHashObj*      pStUpdated = taosHashInit(64, hashFn, true, HASH_NO_LOCK);
  SOperatorInfo* downstream = pOperator->pDownstream[0];
  SArray*        pUpdated = taosArrayInit(16, POINTER_BYTES);
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
5
54liuyao 已提交
3821
      clearSpecialDataBlock(pInfo->pUpdateRes);
5
54liuyao 已提交
3822 3823 3824
      break;
    }

5
54liuyao 已提交
3825
    if (pBlock->info.type == STREAM_CLEAR) {
5
54liuyao 已提交
3826
      SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
5
54liuyao 已提交
3827
      doClearSessionWindows(&pInfo->streamAggSup, pSup, pBlock, 0, pSup->numOfExprs, 0, pWins);
5
54liuyao 已提交
3828 3829 3830 3831
      removeSessionResults(pStUpdated, pWins);
      taosArrayDestroy(pWins);
      copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
      break;
5
54liuyao 已提交
3832 3833
    } else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT) {
      // gap must be 0
3834
      doDeleteTimeWindows(&pInfo->streamAggSup, pBlock, 0, NULL);
5
54liuyao 已提交
3835 3836 3837
      copyDataBlock(pInfo->pDelRes, pBlock);
      pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;
      break;
5
54liuyao 已提交
3838 3839 3840 3841 3842
    } else if (pBlock->info.type == STREAM_GET_ALL) {
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getResWinForSession);
      continue;
    }

5
54liuyao 已提交
3843 3844 3845 3846
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
5
54liuyao 已提交
3847 3848 3849 3850 3851 3852 3853 3854 3855 3856 3857 3858
    // the pDataBlock are always the same one, no need to call this again
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
    doStreamSessionAggImpl(pOperator, pBlock, pStUpdated, pInfo->pStDeleted, false);
    maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
  }

  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
  // semi operator
  // closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated,
  //                    getResWinForSession);
5
54liuyao 已提交
3859
  copyUpdateResult(pStUpdated, pUpdated);
5
54liuyao 已提交
3860
  taosHashCleanup(pStUpdated);
5
54liuyao 已提交
3861

3862 3863
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated,
                        pSup->rowEntryInfoOffset);
5
54liuyao 已提交
3864
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
5
54liuyao 已提交
3865
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
5
54liuyao 已提交
3866 3867 3868 3869 3870 3871 3872 3873 3874 3875

  doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
  if (pBInfo->pRes->info.rows > 0) {
    printDataBlock(pBInfo->pRes, "Semi  Session");
    return pBInfo->pRes;
  }

  // doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
  if (pInfo->pDelRes->info.rows > 0 && !pInfo->returnDelete) {
    pInfo->returnDelete = true;
5
54liuyao 已提交
3876
    printDataBlock(pInfo->pDelRes, "Semi  Session");
5
54liuyao 已提交
3877 3878
    return pInfo->pDelRes;
  }
5
54liuyao 已提交
3879 3880

  if (pInfo->pUpdateRes->info.rows > 0) {
5
54liuyao 已提交
3881 3882
    // process the rest of the data
    pOperator->status = OP_OPENED;
5
54liuyao 已提交
3883
    printDataBlock(pInfo->pUpdateRes, "Semi  Session");
5
54liuyao 已提交
3884 3885
    return pInfo->pUpdateRes;
  }
5
54liuyao 已提交
3886 3887 3888

  pOperator->status = OP_EXEC_DONE;
  return NULL;
5
54liuyao 已提交
3889
}
3890

3891 3892
SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                       SExecTaskInfo* pTaskInfo, int32_t numOfChild) {
3893 3894
  int32_t        code = TSDB_CODE_OUT_OF_MEMORY;
  SOperatorInfo* pOperator = createStreamSessionAggOperatorInfo(downstream, pPhyNode, pTaskInfo);
3895 3896 3897
  if (pOperator == NULL) {
    goto _error;
  }
3898
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
5
54liuyao 已提交
3899 3900 3901 3902 3903 3904 3905

  if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
    pInfo->isFinal = true;
    pOperator->name = "StreamSessionFinalAggOperator";
  } else {
    pInfo->isFinal = false;
    pInfo->pUpdateRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
5
54liuyao 已提交
3906
    pInfo->pUpdateRes->info.type = STREAM_CLEAR;
5
54liuyao 已提交
3907 3908 3909
    blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
    pOperator->name = "StreamSessionSemiAggOperator";
    pOperator->fpSet =
3910 3911
        createOperatorFpSet(operatorDummyOpenFn, doStreamSessionSemiAgg, NULL, NULL,
                            destroyStreamSessionAggOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
3912 3913 3914 3915 3916
  }
  pOperator->operatorType = pPhyNode->type;
  if (numOfChild > 0) {
    pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
    for (int32_t i = 0; i < numOfChild; i++) {
3917
      SOperatorInfo* pChild = createStreamFinalSessionAggOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
5
54liuyao 已提交
3918 3919 3920 3921
      if (pChild == NULL) {
        goto _error;
      }
      taosArrayPush(pInfo->pChildren, &pChild);
3922 3923 3924 3925 3926 3927
    }
  }
  return pOperator;

_error:
  if (pInfo != NULL) {
3928
    destroyStreamSessionAggOperatorInfo(pInfo, pOperator->exprSupp.numOfExprs);
3929 3930 3931 3932 3933 3934 3935
  }

  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
5
54liuyao 已提交
3936 3937

void destroyStreamStateOperatorInfo(void* param, int32_t numOfOutput) {
X
Xiaoyu Wang 已提交
3938
  SStreamStateAggOperatorInfo* pInfo = (SStreamStateAggOperatorInfo*)param;
3939
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
3940 3941 3942 3943 3944
  destroyStreamAggSupporter(&pInfo->streamAggSup);
  cleanupGroupResInfo(&pInfo->groupResInfo);
  if (pInfo->pChildren != NULL) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
X
Xiaoyu Wang 已提交
3945
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, i);
5
54liuyao 已提交
3946 3947 3948 3949 3950 3951
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
      destroyStreamSessionAggOperatorInfo(pChInfo, numOfOutput);
      taosMemoryFreeClear(pChild);
      taosMemoryFreeClear(pChInfo);
    }
  }
D
dapan1121 已提交
3952 3953

  taosMemoryFreeClear(param);
5
54liuyao 已提交
3954 3955 3956 3957 3958 3959 3960
}

int64_t getStateWinTsKey(void* data, int32_t index) {
  SStateWindowInfo* pStateWin = taosArrayGet(data, index);
  return pStateWin->winInfo.win.ekey;
}

X
Xiaoyu Wang 已提交
3961 3962 3963 3964 3965 3966 3967 3968 3969 3970 3971 3972
SStateWindowInfo* addNewStateWindow(SArray* pWinInfos, TSKEY ts, char* pKeyData, SColumn* pCol) {
  SStateWindowInfo win = {
      .stateKey.bytes = pCol->bytes,
      .stateKey.type = pCol->type,
      .stateKey.pData = taosMemoryCalloc(1, pCol->bytes),
      .winInfo.pos.offset = -1,
      .winInfo.pos.pageId = -1,
      .winInfo.win.skey = ts,
      .winInfo.win.ekey = ts,
      .winInfo.isOutput = false,
      .winInfo.isClosed = false,
  };
5
54liuyao 已提交
3973 3974 3975 3976 3977 3978 3979 3980
  if (IS_VAR_DATA_TYPE(win.stateKey.type)) {
    varDataCopy(win.stateKey.pData, pKeyData);
  } else {
    memcpy(win.stateKey.pData, pKeyData, win.stateKey.bytes);
  }
  return taosArrayPush(pWinInfos, &win);
}

X
Xiaoyu Wang 已提交
3981 3982 3983 3984 3985 3986 3987 3988 3989 3990 3991 3992
SStateWindowInfo* insertNewStateWindow(SArray* pWinInfos, TSKEY ts, char* pKeyData, int32_t index, SColumn* pCol) {
  SStateWindowInfo win = {
      .stateKey.bytes = pCol->bytes,
      .stateKey.type = pCol->type,
      .stateKey.pData = taosMemoryCalloc(1, pCol->bytes),
      .winInfo.pos.offset = -1,
      .winInfo.pos.pageId = -1,
      .winInfo.win.skey = ts,
      .winInfo.win.ekey = ts,
      .winInfo.isOutput = false,
      .winInfo.isClosed = false,
  };
5
54liuyao 已提交
3993 3994 3995 3996 3997 3998 3999 4000 4001 4002 4003 4004 4005 4006 4007 4008 4009 4010 4011
  if (IS_VAR_DATA_TYPE(win.stateKey.type)) {
    varDataCopy(win.stateKey.pData, pKeyData);
  } else {
    memcpy(win.stateKey.pData, pKeyData, win.stateKey.bytes);
  }
  return taosArrayInsert(pWinInfos, index, &win);
}

bool isTsInWindow(SStateWindowInfo* pWin, TSKEY ts) {
  if (pWin->winInfo.win.skey <= ts && ts <= pWin->winInfo.win.ekey) {
    return true;
  }
  return false;
}

bool isEqualStateKey(SStateWindowInfo* pWin, char* pKeyData) {
  return pKeyData && compareVal(pKeyData, &pWin->stateKey);
}

4012 4013 4014
SStateWindowInfo* getStateWindowByTs(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int32_t* pIndex) {
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;
X
Xiaoyu Wang 已提交
4015 4016
  int32_t           size = taosArrayGetSize(pWinInfos);
  int32_t           index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
5
54liuyao 已提交
4017 4018 4019 4020 4021 4022 4023 4024 4025 4026 4027 4028 4029 4030 4031 4032 4033 4034 4035 4036
  SStateWindowInfo* pWin = NULL;
  if (index >= 0) {
    pWin = taosArrayGet(pWinInfos, index);
    if (isTsInWindow(pWin, ts)) {
      *pIndex = index;
      return pWin;
    }
  }

  if (index + 1 < size) {
    pWin = taosArrayGet(pWinInfos, index + 1);
    if (isTsInWindow(pWin, ts)) {
      *pIndex = index + 1;
      return pWin;
    }
  }
  *pIndex = 0;
  return NULL;
}

4037 4038
SStateWindowInfo* getStateWindow(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, char* pKeyData,
                                 SColumn* pCol, int32_t* pIndex) {
4039 4040
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;
5
54liuyao 已提交
4041 4042 4043 4044 4045
  int32_t size = taosArrayGetSize(pWinInfos);
  if (size == 0) {
    *pIndex = 0;
    return addNewStateWindow(pWinInfos, ts, pKeyData, pCol);
  }
X
Xiaoyu Wang 已提交
4046
  int32_t           index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
5
54liuyao 已提交
4047 4048 4049 4050 4051 4052 4053 4054 4055 4056 4057 4058 4059 4060 4061 4062 4063 4064 4065 4066 4067 4068 4069 4070 4071 4072 4073 4074 4075 4076 4077 4078 4079
  SStateWindowInfo* pWin = NULL;
  if (index >= 0) {
    pWin = taosArrayGet(pWinInfos, index);
    if (isTsInWindow(pWin, ts)) {
      *pIndex = index;
      return pWin;
    }
  }

  if (index + 1 < size) {
    pWin = taosArrayGet(pWinInfos, index + 1);
    if (isTsInWindow(pWin, ts) || isEqualStateKey(pWin, pKeyData)) {
      *pIndex = index + 1;
      return pWin;
    }
  }

  if (index >= 0) {
    pWin = taosArrayGet(pWinInfos, index);
    if (isEqualStateKey(pWin, pKeyData)) {
      *pIndex = index;
      return pWin;
    }
  }

  if (index == size - 1) {
    *pIndex = taosArrayGetSize(pWinInfos);
    return addNewStateWindow(pWinInfos, ts, pKeyData, pCol);
  }
  *pIndex = index + 1;
  return insertNewStateWindow(pWinInfos, ts, pKeyData, index + 1, pCol);
}

X
Xiaoyu Wang 已提交
4080 4081
int32_t updateStateWindowInfo(SArray* pWinInfos, int32_t winIndex, TSKEY* pTs, SColumnInfoData* pKeyCol, int32_t rows,
                              int32_t start, bool* allEqual, SHashObj* pSeDelete) {
5
54liuyao 已提交
4082 4083 4084 4085 4086
  *allEqual = true;
  SStateWindowInfo* pWinInfo = taosArrayGet(pWinInfos, winIndex);
  for (int32_t i = start; i < rows; ++i) {
    char* pKeyData = colDataGetData(pKeyCol, i);
    if (!isTsInWindow(pWinInfo, pTs[i])) {
X
Xiaoyu Wang 已提交
4087
      if (isEqualStateKey(pWinInfo, pKeyData)) {
5
54liuyao 已提交
4088 4089 4090 4091 4092 4093 4094 4095 4096 4097 4098 4099 4100 4101
        int32_t size = taosArrayGetSize(pWinInfos);
        if (winIndex + 1 < size) {
          SStateWindowInfo* pNextWin = taosArrayGet(pWinInfos, winIndex + 1);
          // ts belongs to the next window
          if (pTs[i] >= pNextWin->winInfo.win.skey) {
            return i - start;
          }
        }
      } else {
        return i - start;
      }
    }
    if (pWinInfo->winInfo.win.skey > pTs[i]) {
      if (pSeDelete && pWinInfo->winInfo.isOutput) {
X
Xiaoyu Wang 已提交
4102 4103
        taosHashPut(pSeDelete, &pWinInfo->winInfo.pos, sizeof(SResultRowPosition), &pWinInfo->winInfo.win.skey,
                    sizeof(TSKEY));
5
54liuyao 已提交
4104 4105 4106 4107 4108 4109 4110 4111 4112 4113 4114 4115
        pWinInfo->winInfo.isOutput = false;
      }
      pWinInfo->winInfo.win.skey = pTs[i];
    }
    pWinInfo->winInfo.win.ekey = TMAX(pWinInfo->winInfo.win.ekey, pTs[i]);
    if (!isEqualStateKey(pWinInfo, pKeyData)) {
      *allEqual = false;
    }
  }
  return rows - start;
}

X
Xiaoyu Wang 已提交
4116 4117
static void doClearStateWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, int32_t tsIndex, SColumn* pCol,
                                int32_t keyIndex, SHashObj* pSeUpdated, SHashObj* pSeDeleted) {
5
54liuyao 已提交
4118 4119
  SColumnInfoData* pTsColInfo = taosArrayGet(pBlock->pDataBlock, tsIndex);
  SColumnInfoData* pKeyColInfo = taosArrayGet(pBlock->pDataBlock, keyIndex);
X
Xiaoyu Wang 已提交
4120 4121 4122
  TSKEY*           tsCol = (TSKEY*)pTsColInfo->pData;
  bool             allEqual = false;
  int32_t          step = 1;
5
54liuyao 已提交
4123
  for (int32_t i = 0; i < pBlock->info.rows; i += step) {
X
Xiaoyu Wang 已提交
4124 4125
    char*             pKeyData = colDataGetData(pKeyColInfo, i);
    int32_t           winIndex = 0;
4126
    SStateWindowInfo* pCurWin = getStateWindowByTs(pAggSup, tsCol[i], pBlock->info.groupId, &winIndex);
5
54liuyao 已提交
4127 4128 4129
    if (!pCurWin) {
      continue;
    }
4130
    step = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual,
X
Xiaoyu Wang 已提交
4131
                                 pSeDeleted);
5
54liuyao 已提交
4132 4133 4134
    ASSERT(isTsInWindow(pCurWin, tsCol[i]) || isEqualStateKey(pCurWin, pKeyData));
    taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
    taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
4135
    deleteWindow(pAggSup->pCurWins, winIndex);
5
54liuyao 已提交
4136 4137 4138
  }
}

X
Xiaoyu Wang 已提交
4139 4140 4141
static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SHashObj* pSeUpdated,
                                 SHashObj* pStDeleted) {
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
4142
  SStreamStateAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
4143
  bool                         masterScan = true;
4144
  int32_t                      numOfOutput = pOperator->exprSupp.numOfExprs;
X
Xiaoyu Wang 已提交
4145 4146 4147 4148 4149 4150 4151
  int64_t                      groupId = pSDataBlock->info.groupId;
  int64_t                      code = TSDB_CODE_SUCCESS;
  int32_t                      step = 1;
  bool                         ascScan = true;
  TSKEY*                       tsCols = NULL;
  SResultRow*                  pResult = NULL;
  int32_t                      winRows = 0;
5
54liuyao 已提交
4152
  if (pSDataBlock->pDataBlock != NULL) {
X
Xiaoyu Wang 已提交
4153 4154
    SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
    tsCols = (int64_t*)pColDataInfo->pData;
5
54liuyao 已提交
4155
  } else {
X
Xiaoyu Wang 已提交
4156
    return;
5
54liuyao 已提交
4157
  }
X
Xiaoyu Wang 已提交
4158

5
54liuyao 已提交
4159
  SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
X
Xiaoyu Wang 已提交
4160 4161
  SColumnInfoData*     pKeyColInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->stateCol.slotId);
  for (int32_t i = 0; i < pSDataBlock->info.rows; i += winRows) {
5
54liuyao 已提交
4162
    if (pInfo->ignoreExpiredData && isOverdue(tsCols[i], &pInfo->twAggSup)) {
5
54liuyao 已提交
4163 4164 4165
      i++;
      continue;
    }
X
Xiaoyu Wang 已提交
4166 4167 4168
    char*             pKeyData = colDataGetData(pKeyColInfo, i);
    int32_t           winIndex = 0;
    bool              allEqual = true;
4169 4170 4171 4172
    SStateWindowInfo* pCurWin =
        getStateWindow(pAggSup, tsCols[i], pSDataBlock->info.groupId, pKeyData, &pInfo->stateCol, &winIndex);
    winRows = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCols, pKeyColInfo, pSDataBlock->info.rows, i,
                                    &allEqual, pInfo->pSeDeleted);
5
54liuyao 已提交
4173 4174 4175
    if (!allEqual) {
      taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
      taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
4176
      deleteWindow(pAggSup->pCurWins, winIndex);
5
54liuyao 已提交
4177 4178
      continue;
    }
4179
    code = doOneStateWindowAgg(pInfo, pSDataBlock, &pCurWin->winInfo, &pResult, i, winRows, numOfOutput, pOperator);
5
54liuyao 已提交
4180 4181 4182 4183 4184
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
    pCurWin->winInfo.isClosed = false;
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
5
54liuyao 已提交
4185
      SWinRes value = {.ts = pCurWin->winInfo.win.skey, .groupId = groupId};
4186
      code = taosHashPut(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition), &value, sizeof(SWinRes));
5
54liuyao 已提交
4187 4188 4189 4190 4191 4192 4193 4194 4195 4196 4197 4198 4199
      if (code != TSDB_CODE_SUCCESS) {
        longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
      }
      pCurWin->winInfo.isOutput = true;
    }
  }
}

static SSDataBlock* doStreamStateAgg(SOperatorInfo* pOperator) {
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

4200
  SExprSupp*                   pSup = &pOperator->exprSupp;
5
54liuyao 已提交
4201
  SStreamStateAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
4202
  SOptrBasicInfo*              pBInfo = &pInfo->binfo;
5
54liuyao 已提交
4203 4204 4205
  if (pOperator->status == OP_RES_TO_RETURN) {
    doBuildDeleteDataBlock(pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
    if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
4206
      printDataBlock(pInfo->pDelRes, "single state");
5
54liuyao 已提交
4207 4208
      return pInfo->pDelRes;
    }
X
Xiaoyu Wang 已提交
4209
    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
4210
    if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) {
5
54liuyao 已提交
4211 4212
      doSetOperatorCompleted(pOperator);
    }
5
54liuyao 已提交
4213
    printDataBlock(pBInfo->pRes, "single state");
5
54liuyao 已提交
4214 4215 4216
    return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
  }

X
Xiaoyu Wang 已提交
4217 4218
  _hash_fn_t     hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  SHashObj*      pSeUpdated = taosHashInit(64, hashFn, true, HASH_NO_LOCK);
5
54liuyao 已提交
4219
  SOperatorInfo* downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
4220
  SArray*        pUpdated = taosArrayInit(16, POINTER_BYTES);
5
54liuyao 已提交
4221 4222 4223 4224 4225
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
      break;
    }
5
54liuyao 已提交
4226
    printDataBlock(pBlock, "single state recv");
4227

5
54liuyao 已提交
4228
    if (pBlock->info.type == STREAM_CLEAR) {
X
Xiaoyu Wang 已提交
4229 4230
      doClearStateWindows(&pInfo->streamAggSup, pBlock, pInfo->primaryTsIndex, &pInfo->stateCol, pInfo->stateCol.slotId,
                          pSeUpdated, pInfo->pSeDeleted);
5
54liuyao 已提交
4231
      continue;
4232 4233 4234 4235 4236 4237
    } else if (pBlock->info.type == STREAM_DELETE_DATA) {
      SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
      doDeleteTimeWindows(&pInfo->streamAggSup, pBlock, 0, pWins);
      copyDeleteWindowInfo(pWins, pInfo->pSeDeleted);
      taosArrayDestroy(pWins);
      continue;
4238
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
4239
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getResWinForState);
5
54liuyao 已提交
4240
      continue;
5
54liuyao 已提交
4241
    }
4242

5
54liuyao 已提交
4243 4244 4245 4246
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
4247
    // the pDataBlock are always the same one, no need to call this again
4248
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
4249 4250 4251 4252 4253
    doStreamStateAggImpl(pOperator, pBlock, pSeUpdated, pInfo->pSeDeleted);
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
  }
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
X
Xiaoyu Wang 已提交
4254

4255 4256
  closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated, getResWinForState,
                     pInfo->ignoreExpiredData);
5
54liuyao 已提交
4257
  closeChildSessionWindow(pInfo->pChildren, pInfo->twAggSup.maxTs, pInfo->ignoreExpiredData);
5
54liuyao 已提交
4258
  copyUpdateResult(pSeUpdated, pUpdated);
5
54liuyao 已提交
4259 4260
  taosHashCleanup(pSeUpdated);

4261 4262
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated,
                        pSup->rowEntryInfoOffset);
5
54liuyao 已提交
4263 4264 4265 4266
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
  doBuildDeleteDataBlock(pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
  if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
4267
    printDataBlock(pInfo->pDelRes, "single state");
5
54liuyao 已提交
4268 4269
    return pInfo->pDelRes;
  }
X
Xiaoyu Wang 已提交
4270
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
5
54liuyao 已提交
4271
  printDataBlock(pBInfo->pRes, "single state");
5
54liuyao 已提交
4272 4273 4274
  return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
}

4275 4276 4277 4278
int32_t initStateAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SStateWindowInfo));
}

X
Xiaoyu Wang 已提交
4279 4280 4281 4282 4283 4284
SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                SExecTaskInfo* pTaskInfo) {
  SStreamStateWinodwPhysiNode* pStateNode = (SStreamStateWinodwPhysiNode*)pPhyNode;
  SSDataBlock*                 pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
  int32_t                      tsSlotId = ((SColumnNode*)pStateNode->window.pTspk)->slotId;
  SColumnNode*                 pColNode = (SColumnNode*)((STargetNode*)pStateNode->pStateKey)->pExpr;
4285
  int32_t                      code = TSDB_CODE_OUT_OF_MEMORY;
5
54liuyao 已提交
4286

X
Xiaoyu Wang 已提交
4287 4288
  SStreamStateAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamStateAggOperatorInfo));
  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
4289 4290 4291 4292
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

4293 4294
  SExprSupp* pSup = &pOperator->exprSupp;

X
Xiaoyu Wang 已提交
4295
  int32_t    numOfCols = 0;
5
54liuyao 已提交
4296 4297 4298 4299
  SExprInfo* pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &numOfCols);

  pInfo->stateCol = extractColumnFromColumnNode(pColNode);
  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
4300 4301 4302 4303 4304 4305 4306 4307 4308
  if (pStateNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pStateNode->window.pExprs, NULL, &numOfScalar);
    int32_t    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }

4309
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
X
Xiaoyu Wang 已提交
4310 4311
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pStateNode->window.watermark,
5
54liuyao 已提交
4312 4313
      .calTrigger = pStateNode->window.triggerType,
      .maxTs = INT64_MIN,
X
Xiaoyu Wang 已提交
4314
  };
5
54liuyao 已提交
4315
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
4316

4317
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
5
54liuyao 已提交
4318 4319 4320 4321
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

4322
  code = initStateAggSupporter(&pInfo->streamAggSup, "StreamStateAggOperatorInfo", pSup->pCtx, numOfCols);
5
54liuyao 已提交
4323 4324 4325 4326 4327 4328 4329 4330 4331
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  pInfo->pDummyCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfCols, sizeof(SqlFunctionCtx));
  if (pInfo->pDummyCtx == NULL) {
    goto _error;
  }

4332
  initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols);
5
54liuyao 已提交
4333 4334 4335 4336 4337
  pInfo->primaryTsIndex = tsSlotId;
  pInfo->order = TSDB_ORDER_ASC;
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  pInfo->pSeDeleted = taosHashInit(64, hashFn, true, HASH_NO_LOCK);
  pInfo->pDelIterator = NULL;
5
54liuyao 已提交
4338
  // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete
4339 4340
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
5
54liuyao 已提交
4341
  pInfo->pChildren = NULL;
5
54liuyao 已提交
4342
  pInfo->ignoreExpiredData = pStateNode->window.igExpired;
5
54liuyao 已提交
4343 4344

  pOperator->name = "StreamStateAggOperator";
4345
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
5
54liuyao 已提交
4346 4347
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
4348 4349
  pOperator->exprSupp.numOfExprs = numOfCols;
  pOperator->exprSupp.pExprInfo = pExprInfo;
5
54liuyao 已提交
4350 4351
  pOperator->pTaskInfo = pTaskInfo;
  pOperator->info = pInfo;
X
Xiaoyu Wang 已提交
4352 4353 4354
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, NULL,
                                         destroyStreamStateOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
  initDownStream(downstream, &pInfo->streamAggSup, 0, pInfo->twAggSup.waterMark, pOperator->operatorType);
5
54liuyao 已提交
4355 4356 4357 4358 4359 4360 4361 4362 4363 4364 4365 4366 4367
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
  return pOperator;

_error:
  destroyStreamStateOperatorInfo(pInfo, numOfCols);
  taosMemoryFreeClear(pInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
4368

4369 4370
void destroyMergeAlignedIntervalOperatorInfo(void* param, int32_t numOfOutput) {
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = (SMergeAlignedIntervalAggOperatorInfo*)param;
D
dapan1121 已提交
4371 4372
  destroyIntervalOperatorInfo(miaInfo->intervalAggOperatorInfo, numOfOutput);

D
dapan1121 已提交
4373
  taosMemoryFreeClear(param);
4374 4375
}

4376 4377
static int32_t outputMergeAlignedIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId,
                                                SSDataBlock* pResultBlock, TSKEY wstartTs) {
4378
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
D
dapan1121 已提交
4379
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4380
  SExecTaskInfo*                        pTaskInfo = pOperatorInfo->pTaskInfo;
4381 4382

  SExprSupp* pSup = &pOperatorInfo->exprSupp;
4383
  bool       ascScan = (iaInfo->order == TSDB_ORDER_ASC);
4384

4385 4386 4387 4388
  SET_RES_WINDOW_KEY(iaInfo->aggSup.keyBuf, &wstartTs, TSDB_KEYSIZE, tableGroupId);
  SResultRowPosition* p1 = (SResultRowPosition*)taosHashGet(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf,
                                                            GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
  ASSERT(p1 != NULL);
4389

4390 4391
  finalizeResultRowIntoResultDataBlock(iaInfo->aggSup.pResultBuf, p1, pSup->pCtx, pSup->pExprInfo, pSup->numOfExprs,
                                       pSup->rowEntryInfoOffset, pResultBlock, pTaskInfo);
4392
  taosHashRemove(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf, GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
4393

4394
  return 0;
4395 4396
}

4397 4398
static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo,
                                          SSDataBlock* pBlock, int32_t scanFlag, SSDataBlock* pResultBlock) {
4399
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
D
dapan1121 已提交
4400
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4401 4402

  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
4403
  SExprSupp*     pSup = &pOperatorInfo->exprSupp;
4404 4405

  int32_t     startPos = 0;
4406
  int32_t     numOfOutput = pSup->numOfExprs;
4407
  int64_t*    tsCols = extractTsCol(pBlock, iaInfo);
4408 4409 4410 4411
  uint64_t    tableGroupId = pBlock->info.groupId;
  TSKEY       blockStartTs = getStartTsKey(&pBlock->info.window, tsCols);
  SResultRow* pResult = NULL;

4412 4413
  STimeWindow win;
  win.skey = blockStartTs;
4414 4415
  win.ekey =
      taosTimeAdd(win.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1;
4416

4417
  // TODO: remove the hash table (groupid + winkey => result row position)
4418 4419
  int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
4420 4421 4422 4423
  if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
    longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

4424 4425
  TSKEY       currTs = blockStartTs;
  TSKEY       currPos = startPos;
4426
  STimeWindow currWin = win;
4427
  while (1) {
4428 4429
    ++currPos;
    if (currPos >= pBlock->info.rows) {
4430 4431
      break;
    }
4432 4433 4434 4435
    if (tsCols[currPos] == currTs) {
      continue;
    } else {
      updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true);
4436 4437
      doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos,
                       tsCols, pBlock->info.rows, numOfOutput, iaInfo->order);
4438

4439
      outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs);
4440 4441 4442

      currTs = tsCols[currPos];
      currWin.skey = currTs;
4443
      currWin.ekey = taosTimeAdd(currWin.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit,
4444 4445
                                 iaInfo->interval.precision) -
                     1;
4446
      startPos = currPos;
4447 4448
      ret = setTimeWindowOutputBuf(pResultRowInfo, &currWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
                                   pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
4449 4450 4451
      if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
        longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
      }
4452 4453
    }
  }
4454
  updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true);
4455 4456
  doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos,
                   tsCols, pBlock->info.rows, numOfOutput, iaInfo->order);
4457

4458
  outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs);
4459 4460
}

4461
static SSDataBlock* doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
S
shenglian zhou 已提交
4462
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
4463

4464
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperator->info;
D
dapan1121 已提交
4465
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4466 4467 4468 4469
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

4470
  SExprSupp*   pSup = &pOperator->exprSupp;
4471
  SSDataBlock* pRes = iaInfo->binfo.pRes;
4472
  blockDataCleanup(pRes);
4473
  blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
4474

4475 4476 4477 4478 4479 4480 4481 4482 4483 4484
  if (!miaInfo->inputBlocksFinished) {
    SOperatorInfo* downstream = pOperator->pDownstream[0];
    int32_t        scanFlag = MAIN_SCAN;
    while (1) {
      SSDataBlock* pBlock = NULL;
      if (miaInfo->prefetchedBlock == NULL) {
        pBlock = downstream->fpSet.getNextFn(downstream);
      } else {
        pBlock = miaInfo->prefetchedBlock;
        miaInfo->groupId = pBlock->info.groupId;
4485
        miaInfo->prefetchedBlock = NULL;
4486
      }
4487

4488 4489 4490 4491
      if (pBlock == NULL) {
        miaInfo->inputBlocksFinished = true;
        break;
      }
4492

4493 4494 4495 4496 4497 4498 4499
      if (!miaInfo->hasGroupId) {
        miaInfo->hasGroupId = true;
        miaInfo->groupId = pBlock->info.groupId;
      } else if (miaInfo->groupId != pBlock->info.groupId) {
        miaInfo->prefetchedBlock = pBlock;
        break;
      }
4500

4501
      getTableScanInfo(pOperator, &iaInfo->order, &scanFlag);
4502
      setInputDataBlock(pOperator, pSup->pCtx, pBlock, iaInfo->order, scanFlag, true);
4503
      doMergeAlignedIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);
4504 4505
      doFilter(miaInfo->pCondition, pRes);
      if (pRes->info.rows > 0) {
4506 4507 4508 4509 4510
        break;
      }
    }

    pRes->info.groupId = miaInfo->groupId;
4511 4512
  }

4513
  if (miaInfo->inputBlocksFinished) {
4514 4515 4516 4517 4518 4519 4520 4521
    doSetOperatorCompleted(pOperator);
  }

  size_t rows = pRes->info.rows;
  pOperator->resultInfo.totalRows += rows;
  return (rows == 0) ? NULL : pRes;
}

4522 4523
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo,
                                                      int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval,
4524 4525
                                                      int32_t primaryTsSlotId, SNode* pCondition,
                                                      SExecTaskInfo* pTaskInfo) {
4526
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = taosMemoryCalloc(1, sizeof(SMergeAlignedIntervalAggOperatorInfo));
4527
  SOperatorInfo*                        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
4528 4529 4530 4531
  if (miaInfo == NULL || pOperator == NULL) {
    goto _error;
  }

D
dapan1121 已提交
4532 4533 4534 4535 4536 4537
  miaInfo->intervalAggOperatorInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
  if (miaInfo->intervalAggOperatorInfo == NULL) {
    goto _error;
  }

  SIntervalAggOperatorInfo* iaInfo = miaInfo->intervalAggOperatorInfo;
4538
  SExprSupp*                pSup = &pOperator->exprSupp;
4539

4540
  miaInfo->pCondition = pCondition;
4541 4542 4543
  iaInfo->win = pTaskInfo->window;
  iaInfo->order = TSDB_ORDER_ASC;
  iaInfo->interval = *pInterval;
4544 4545
  iaInfo->execModel = pTaskInfo->execModel;
  iaInfo->primaryTsIndex = primaryTsSlotId;
4546 4547 4548 4549

  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  initResultSizeInfo(pOperator, 4096);

4550 4551
  int32_t code =
      initAggInfo(&pOperator->exprSupp, &iaInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
4552
  initBasicInfo(&iaInfo->binfo, pResBlock);
4553

4554
  initExecTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &iaInfo->win);
4555

4556
  iaInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, numOfCols, iaInfo);
4557 4558
  if (iaInfo->timeWindowInterpo) {
    iaInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition));
4559 4560
  }

4561
  if (code != TSDB_CODE_SUCCESS) {
4562 4563 4564
    goto _error;
  }

4565
  initResultRowInfo(&iaInfo->binfo.resultRowInfo);
4566

4567 4568
  pOperator->name = "TimeMergeAlignedIntervalAggOperator";
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL;
4569 4570
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
4571
  pOperator->exprSupp.pExprInfo = pExprInfo;
4572
  pOperator->pTaskInfo = pTaskInfo;
4573
  pOperator->exprSupp.numOfExprs = numOfCols;
4574
  pOperator->info = miaInfo;
4575

4576 4577
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doMergeAlignedIntervalAgg, NULL, NULL,
                                         destroyMergeAlignedIntervalOperatorInfo, NULL, NULL, NULL);
4578 4579 4580 4581 4582 4583 4584 4585 4586

  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

_error:
4587
  destroyMergeAlignedIntervalOperatorInfo(miaInfo, numOfCols);
4588
  taosMemoryFreeClear(miaInfo);
4589 4590 4591 4592
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
4593 4594 4595 4596 4597

//=====================================================================================================================
// merge interval operator
typedef struct SMergeIntervalAggOperatorInfo {
  SIntervalAggOperatorInfo intervalAggOperatorInfo;
L
Liu Jicong 已提交
4598 4599 4600 4601 4602 4603
  SList*                   groupIntervals;
  SListIter                groupIntervalsIter;
  bool                     hasGroupId;
  uint64_t                 groupId;
  SSDataBlock*             prefetchedBlock;
  bool                     inputBlocksFinished;
4604 4605
} SMergeIntervalAggOperatorInfo;

S
slzhou 已提交
4606
typedef struct SGroupTimeWindow {
L
Liu Jicong 已提交
4607
  uint64_t    groupId;
S
slzhou 已提交
4608 4609 4610
  STimeWindow window;
} SGroupTimeWindow;

4611 4612
void destroyMergeIntervalOperatorInfo(void* param, int32_t numOfOutput) {
  SMergeIntervalAggOperatorInfo* miaInfo = (SMergeIntervalAggOperatorInfo*)param;
S
slzhou 已提交
4613
  tdListFree(miaInfo->groupIntervals);
4614
  destroyIntervalOperatorInfo(&miaInfo->intervalAggOperatorInfo, numOfOutput);
4615

D
dapan1121 已提交
4616
  taosMemoryFreeClear(param);
4617 4618
}

L
Liu Jicong 已提交
4619 4620
static int32_t finalizeWindowResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, STimeWindow* win,
                                    SSDataBlock* pResultBlock) {
4621 4622 4623 4624 4625 4626 4627 4628 4629 4630 4631 4632 4633 4634 4635 4636
  SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
  SIntervalAggOperatorInfo*      iaInfo = &miaInfo->intervalAggOperatorInfo;
  SExecTaskInfo*                 pTaskInfo = pOperatorInfo->pTaskInfo;
  bool                           ascScan = (iaInfo->order == TSDB_ORDER_ASC);
  SExprSupp*                     pExprSup = &pOperatorInfo->exprSupp;

  SET_RES_WINDOW_KEY(iaInfo->aggSup.keyBuf, &win->skey, TSDB_KEYSIZE, tableGroupId);
  SResultRowPosition* p1 = (SResultRowPosition*)taosHashGet(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf,
                                                            GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
  ASSERT(p1 != NULL);
  finalizeResultRowIntoResultDataBlock(iaInfo->aggSup.pResultBuf, p1, pExprSup->pCtx, pExprSup->pExprInfo,
                                       pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset, pResultBlock, pTaskInfo);
  taosHashRemove(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf, GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
  return TSDB_CODE_SUCCESS;
}

4637 4638 4639 4640 4641 4642 4643 4644
static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, SSDataBlock* pResultBlock,
                                        STimeWindow* newWin) {
  SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
  SIntervalAggOperatorInfo*      iaInfo = &miaInfo->intervalAggOperatorInfo;
  SExecTaskInfo*                 pTaskInfo = pOperatorInfo->pTaskInfo;
  bool                           ascScan = (iaInfo->order == TSDB_ORDER_ASC);
  SExprSupp*                     pExprSup = &pOperatorInfo->exprSupp;

S
slzhou 已提交
4645 4646
  SGroupTimeWindow groupTimeWindow = {.groupId = tableGroupId, .window = *newWin};
  tdListAppend(miaInfo->groupIntervals, &groupTimeWindow);
4647

S
slzhou 已提交
4648 4649 4650 4651 4652
  SListIter iter = {0};
  tdListInitIter(miaInfo->groupIntervals, &iter, TD_LIST_FORWARD);
  SListNode* listNode = NULL;
  while ((listNode = tdListNext(&iter)) != NULL) {
    SGroupTimeWindow* prevGrpWin = (SGroupTimeWindow*)listNode->data;
L
Liu Jicong 已提交
4653
    if (prevGrpWin->groupId != tableGroupId) {
S
slzhou 已提交
4654 4655 4656
      continue;
    }
    STimeWindow* prevWin = &prevGrpWin->window;
H
Haojun Liao 已提交
4657
    if ((ascScan && newWin->skey > prevWin->ekey) || ((!ascScan) && newWin->skey < prevWin->ekey)) {
S
slzhou 已提交
4658 4659 4660
      finalizeWindowResult(pOperatorInfo, tableGroupId, prevWin, pResultBlock);
      tdListPopNode(miaInfo->groupIntervals, listNode);
    }
4661 4662 4663 4664 4665 4666 4667 4668 4669 4670 4671 4672 4673 4674 4675 4676 4677 4678 4679 4680 4681
  }

  return 0;
}

static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pBlock,
                                   int32_t scanFlag, SSDataBlock* pResultBlock) {
  SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
  SIntervalAggOperatorInfo*      iaInfo = &miaInfo->intervalAggOperatorInfo;

  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
  SExprSupp*     pExprSup = &pOperatorInfo->exprSupp;

  int32_t     startPos = 0;
  int32_t     numOfOutput = pExprSup->numOfExprs;
  int64_t*    tsCols = extractTsCol(pBlock, iaInfo);
  uint64_t    tableGroupId = pBlock->info.groupId;
  bool        ascScan = (iaInfo->order == TSDB_ORDER_ASC);
  TSKEY       blockStartTs = getStartTsKey(&pBlock->info.window, tsCols);
  SResultRow* pResult = NULL;

X
Xiaoyu Wang 已提交
4682 4683
  STimeWindow win =
      getActiveTimeWindow(iaInfo->aggSup.pResultBuf, pResultRowInfo, blockStartTs, &iaInfo->interval, iaInfo->order);
4684 4685 4686 4687 4688 4689 4690 4691 4692 4693 4694 4695 4696 4697 4698 4699 4700 4701 4702 4703 4704 4705 4706 4707 4708 4709 4710 4711 4712 4713 4714 4715 4716 4717 4718 4719 4720 4721 4722 4723 4724 4725 4726 4727 4728 4729 4730 4731 4732 4733 4734 4735 4736 4737 4738 4739 4740 4741 4742 4743 4744 4745 4746 4747 4748 4749 4750 4751 4752 4753 4754 4755 4756 4757 4758 4759 4760 4761 4762 4763 4764 4765 4766 4767 4768 4769 4770 4771 4772 4773 4774 4775 4776 4777 4778 4779 4780 4781 4782 4783 4784 4785 4786

  int32_t ret =
      setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pExprSup->pCtx,
                             numOfOutput, pExprSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
  if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
    longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

  TSKEY   ekey = ascScan ? win.ekey : win.skey;
  int32_t forwardRows =
      getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, iaInfo->order);
  ASSERT(forwardRows > 0);

  // prev time window not interpolation yet.
  if (iaInfo->timeWindowInterpo) {
    SResultRowPosition pos = addToOpenWindowList(pResultRowInfo, pResult);
    doInterpUnclosedTimeWindow(pOperatorInfo, numOfOutput, pResultRowInfo, pBlock, scanFlag, tsCols, &pos);

    // restore current time window
    ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pExprSup->pCtx,
                                 numOfOutput, pExprSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
    if (ret != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }

    // window start key interpolation
    doWindowBorderInterpolation(iaInfo, pBlock, pResult, &win, startPos, forwardRows, pExprSup);
  }

  updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &win, true);
  doApplyFunctions(pTaskInfo, pExprSup->pCtx, &win, &iaInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols,
                   pBlock->info.rows, numOfOutput, iaInfo->order);
  doCloseWindow(pResultRowInfo, iaInfo, pResult);

  // output previous interval results after this interval (&win) is closed
  outputPrevIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, &win);

  STimeWindow nextWin = win;
  while (1) {
    int32_t prevEndPos = forwardRows - 1 + startPos;
    startPos = getNextQualifiedWindow(&iaInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, iaInfo->order);
    if (startPos < 0) {
      break;
    }

    // null data, failed to allocate more memory buffer
    int32_t code =
        setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
                               pExprSup->pCtx, numOfOutput, pExprSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }

    ekey = ascScan ? nextWin.ekey : nextWin.skey;
    forwardRows =
        getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, iaInfo->order);

    // window start(end) key interpolation
    doWindowBorderInterpolation(iaInfo, pBlock, pResult, &nextWin, startPos, forwardRows, pExprSup);

    updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &nextWin, true);
    doApplyFunctions(pTaskInfo, pExprSup->pCtx, &nextWin, &iaInfo->twAggSup.timeWindowData, startPos, forwardRows,
                     tsCols, pBlock->info.rows, numOfOutput, iaInfo->order);
    doCloseWindow(pResultRowInfo, iaInfo, pResult);

    // output previous interval results after this interval (&nextWin) is closed
    outputPrevIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, &nextWin);
  }

  if (iaInfo->timeWindowInterpo) {
    saveDataBlockLastRow(iaInfo->pPrevValues, pBlock, iaInfo->pInterpCols);
  }
}

static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

  SMergeIntervalAggOperatorInfo* miaInfo = pOperator->info;
  SIntervalAggOperatorInfo*      iaInfo = &miaInfo->intervalAggOperatorInfo;
  SExprSupp*                     pExpSupp = &pOperator->exprSupp;

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

  SSDataBlock* pRes = iaInfo->binfo.pRes;
  blockDataCleanup(pRes);
  blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);

  if (!miaInfo->inputBlocksFinished) {
    SOperatorInfo* downstream = pOperator->pDownstream[0];
    int32_t        scanFlag = MAIN_SCAN;
    while (1) {
      SSDataBlock* pBlock = NULL;
      if (miaInfo->prefetchedBlock == NULL) {
        pBlock = downstream->fpSet.getNextFn(downstream);
      } else {
        pBlock = miaInfo->prefetchedBlock;
        miaInfo->groupId = pBlock->info.groupId;
        miaInfo->prefetchedBlock = NULL;
      }

      if (pBlock == NULL) {
S
slzhou 已提交
4787
        tdListInitIter(miaInfo->groupIntervals, &miaInfo->groupIntervalsIter, TD_LIST_FORWARD);
4788 4789 4790 4791 4792 4793 4794 4795 4796 4797 4798 4799 4800 4801 4802 4803 4804 4805 4806 4807 4808 4809
        miaInfo->inputBlocksFinished = true;
        break;
      }

      if (!miaInfo->hasGroupId) {
        miaInfo->hasGroupId = true;
        miaInfo->groupId = pBlock->info.groupId;
      } else if (miaInfo->groupId != pBlock->info.groupId) {
        miaInfo->prefetchedBlock = pBlock;
        break;
      }

      getTableScanInfo(pOperator, &iaInfo->order, &scanFlag);
      setInputDataBlock(pOperator, pExpSupp->pCtx, pBlock, iaInfo->order, scanFlag, true);
      doMergeIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);

      if (pRes->info.rows >= pOperator->resultInfo.threshold) {
        break;
      }
    }

    pRes->info.groupId = miaInfo->groupId;
4810 4811 4812
  }

  if (miaInfo->inputBlocksFinished) {
S
slzhou 已提交
4813
    SListNode* listNode = tdListNext(&miaInfo->groupIntervalsIter);
4814

S
slzhou 已提交
4815 4816 4817 4818
    if (listNode != NULL) {
      SGroupTimeWindow* grpWin = (SGroupTimeWindow*)(listNode->data);
      finalizeWindowResult(pOperator, grpWin->groupId, &grpWin->window, pRes);
      pRes->info.groupId = grpWin->groupId;
4819 4820 4821 4822 4823 4824 4825 4826 4827 4828 4829 4830 4831 4832 4833 4834 4835 4836 4837 4838 4839
    }
  }

  if (pRes->info.rows == 0) {
    doSetOperatorCompleted(pOperator);
  }

  size_t rows = pRes->info.rows;
  pOperator->resultInfo.totalRows += rows;
  return (rows == 0) ? NULL : pRes;
}

SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                               SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
                                               SExecTaskInfo* pTaskInfo) {
  SMergeIntervalAggOperatorInfo* miaInfo = taosMemoryCalloc(1, sizeof(SMergeIntervalAggOperatorInfo));
  SOperatorInfo*                 pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (miaInfo == NULL || pOperator == NULL) {
    goto _error;
  }

S
slzhou 已提交
4840
  miaInfo->groupIntervals = tdListNew(sizeof(SGroupTimeWindow));
4841

4842 4843 4844 4845 4846 4847 4848 4849 4850 4851 4852 4853 4854 4855 4856 4857 4858 4859 4860 4861 4862 4863 4864 4865 4866 4867 4868 4869 4870 4871 4872 4873 4874 4875 4876 4877 4878 4879 4880 4881 4882 4883 4884 4885 4886 4887 4888 4889 4890 4891 4892 4893 4894 4895
  SIntervalAggOperatorInfo* iaInfo = &miaInfo->intervalAggOperatorInfo;

  iaInfo->win = pTaskInfo->window;
  iaInfo->order = TSDB_ORDER_ASC;
  iaInfo->interval = *pInterval;
  iaInfo->execModel = pTaskInfo->execModel;

  iaInfo->primaryTsIndex = primaryTsSlotId;

  SExprSupp* pExprSupp = &pOperator->exprSupp;

  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  initResultSizeInfo(pOperator, 4096);

  int32_t code = initAggInfo(pExprSupp, &iaInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
  initBasicInfo(&iaInfo->binfo, pResBlock);

  initExecTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &iaInfo->win);

  iaInfo->timeWindowInterpo = timeWindowinterpNeeded(pExprSupp->pCtx, numOfCols, iaInfo);
  if (iaInfo->timeWindowInterpo) {
    iaInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition));
    if (iaInfo->binfo.resultRowInfo.openWindow == NULL) {
      goto _error;
    }
  }

  initResultRowInfo(&iaInfo->binfo.resultRowInfo);

  pOperator->name = "TimeMergeIntervalAggOperator";
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL;
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->exprSupp.pExprInfo = pExprInfo;
  pOperator->pTaskInfo = pTaskInfo;
  pOperator->exprSupp.numOfExprs = numOfCols;
  pOperator->info = miaInfo;

  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doMergeIntervalAgg, NULL, NULL,
                                         destroyMergeIntervalOperatorInfo, NULL, NULL, NULL);

  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

_error:
  destroyMergeIntervalOperatorInfo(miaInfo, numOfCols);
  taosMemoryFreeClear(miaInfo);
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
L
Liu Jicong 已提交
4896
}