timewindowoperator.c 189.5 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
bool inSlidingWindow(SInterval* pInterval, STimeWindow* pWin, SDataBlockInfo* pBlockInfo) {
423 424
  if (pInterval->interval != pInterval->sliding &&
      (pWin->ekey < pBlockInfo->calWin.skey || pWin->skey > pBlockInfo->calWin.ekey)) {
5
54liuyao 已提交
425 426 427 428 429
    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;
816 817
  if ((!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) &&
      inSlidingWindow(&pInfo->interval, &win, &pBlock->info)) {
L
Liu Jicong 已提交
818 819
    ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx,
                                 numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
5
54liuyao 已提交
820 821 822
    if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
823

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

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

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

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

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

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

  doCloseWindow(pResultRowInfo, pInfo, pResult);
858 859 860

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

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

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

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

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

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

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

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 已提交
912
  SListNode*         pn = tdListGetTail(pResultRowInfo->openWindow);
913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937
  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) {
      blockDataUpdateTsWindow(pBlock, pInfo->primaryTsIndex);
    }
  }

  return tsCols;
938 939 940 941 942 943 944
}

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

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

949 950
  int32_t scanFlag = MAIN_SCAN;

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

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

960 961
    getTableScanInfo(pOperator, &pInfo->order, &scanFlag);

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

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

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

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

980 981 982 983 984 985 986 987 988 989 990 991
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;
  }
}

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

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

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

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

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

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

    char* val = colDataGetData(pStateColInfoData, j);

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

1026 1027 1028 1029
      pInfo->hasKey = true;

      doKeepNewWindowStartInfo(pRowSup, tsList, j);
      doKeepTuple(pRowSup, tsList[j]);
1030
    } else if (compareVal(val, &pInfo->stateKey)) {
1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041
      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;
1042 1043
      int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx,
                                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1044 1045 1046 1047 1048
      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);
1049
      doApplyFunctions(pTaskInfo, pSup->pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1050 1051 1052 1053 1054
                       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]);
1055 1056 1057 1058 1059 1060 1061

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

  SResultRow* pResult = NULL;
  pRowSup->win.ekey = tsList[pBlock->info.rows - 1];
1067 1068
  int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1069 1070 1071 1072 1073
  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);
1074
  doApplyFunctions(pTaskInfo, pSup->pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1075 1076 1077
                   pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
}

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

  SStateWindowOperatorInfo* pInfo = pOperator->info;
1084

1085 1086
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
  SExprSupp*     pSup = &pOperator->exprSupp;
1087

1088
  SOptrBasicInfo* pBInfo = &pInfo->binfo;
1089 1090

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

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

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

1109
  int32_t order = TSDB_ORDER_ASC;
1110
  int64_t st = taosGetTimestampUs();
1111 1112 1113

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

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

1122 1123 1124
    doStateWindowAggImpl(pOperator, pInfo, pBlock);
  }

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

1127 1128 1129
  pOperator->status = OP_RES_TO_RETURN;
  closeAllResultRows(&pBInfo->resultRowInfo);

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

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

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

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

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

  SSDataBlock* pBlock = pInfo->binfo.pRes;

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

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

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

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

1184 1185 1186
    size_t rows = pBlock->info.rows;
    pOperator->resultInfo.totalRows += rows;

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

// todo merged with the build group result.
1192
static void finalizeUpdatedResult(int32_t numOfOutput, SDiskbasedBuf* pBuf, SArray* pUpdateList,
1193
                                  int32_t* rowEntryInfoOffset) {
1194 1195 1196 1197 1198 1199 1200
  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);
1201

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

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

1222
void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SExprSupp* pSup, int32_t numOfOutput) {
X
Xiaoyu Wang 已提交
1223
  SResultRow*     pResult = getResultRowByPos(pResultBuf, p1);
1224
  SqlFunctionCtx* pCtx = pSup->pCtx;
5
54liuyao 已提交
1225
  for (int32_t i = 0; i < numOfOutput; ++i) {
1226
    pCtx[i].resultInfo = getResultEntryInfo(pResult, i, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
1227 1228 1229 1230 1231 1232 1233 1234 1235
    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 已提交
1236 1237 1238
  SFilePage* bufPage = getBufPage(pResultBuf, p1->pageId);
  setBufPageDirty(bufPage, true);
  releaseBufPage(pResultBuf, bufPage);
5
54liuyao 已提交
1239 1240
}

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

1254 1255 1256 1257 1258 1259 1260 1261 1262
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 已提交
1263
  // SFilePage* bufPage = getBufPage(pAggSup->pResultBuf, p1->pageId);
1264 1265 1266 1267 1268 1269 1270
  // 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);
1271
  TSKEY*           tsStarts = (TSKEY*)pStartCol->pData;
1272
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
1273
  uint64_t*        groupIds = (uint64_t*)pGroupCol->pData;
1274 1275 1276
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    SResultRowInfo dumyInfo;
    dumyInfo.cur.pageId = -1;
1277
    STimeWindow win = getActiveTimeWindow(NULL, &dumyInfo, tsStarts[i], pInterval, TSDB_ORDER_ASC);
1278 1279 1280 1281 1282 1283 1284 1285
    doDeleteIntervalWindow(pAggSup, win.skey, groupIds[i]);
    if (pUpWins) {
      SWinRes winRes = {.ts = win.skey, .groupId = groupIds[i]};
      taosArrayPush(pUpWins, &winRes);
    }
  }
}

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

5
54liuyao 已提交
1317 1318 1319 1320 1321 1322 1323
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)));
1324
    TSKEY               ts = *(int64_t*)((char*)key + sizeof(uint64_t));
5
54liuyao 已提交
1325
    SResultRowPosition* pPos = (SResultRowPosition*)pIte;
1326
    int32_t             code = saveResult(ts, pPos->pageId, pPos->offset, groupId, resWins);
5
54liuyao 已提交
1327 1328 1329 1330 1331 1332 1333
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  }
  return TSDB_CODE_SUCCESS;
}

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

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

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 已提交
1400
    // SFilePage* bufPage = getBufPage(pDiskBuf, pageId);
1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422
    // 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 已提交
1423 1424 1425
  }
}

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

  pInfo->order = TSDB_ORDER_ASC;
1431
  SExprSupp* pSup = &pOperator->exprSupp;
1432 1433 1434 1435 1436 1437

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

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

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

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

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

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

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

1480 1481 1482
    // 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
1483
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
1484
    if (pInfo->invertible) {
1485
      setInverFunction(pSup->pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.type);
1486 1487
    }

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

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

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

D
dapan1121 已提交
1513
  taosMemoryFreeClear(param);
1514 1515 1516
}

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

D
dapan1121 已提交
1522
  taosMemoryFreeClear(param);
1523 1524
}

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

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

D
dapan1121 已提交
1546
  taosMemoryFreeClear(param);
5
54liuyao 已提交
1547 1548
}

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

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

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

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

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

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

      SColumn c = *pParam->pCol;
1587
      taosArrayPush(pInfo->pInterpCols, &c);
1588 1589 1590
      needed = true;

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

  return needed;
}

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

1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618 1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629
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;
}

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

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

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

1657 1658
  pInfo->primaryTsIndex = primaryTsSlotId;

1659 1660
  SExprSupp* pSup = &pOperator->exprSupp;

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

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

1667 1668
  if (isStream) {
    ASSERT(numOfCols > 0);
1669
    increaseTs(pSup->pCtx);
1670
  }
1671

1672
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);
1673

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

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

1691
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
1692

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

  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 已提交
1712
_error:
1713 1714 1715 1716 1717 1718 1719 1720 1721
  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 已提交
1722
                                                STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo) {
1723
  SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
L
Liu Jicong 已提交
1724
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1725 1726 1727 1728 1729 1730 1731 1732 1733 1734 1735 1736 1737 1738 1739
  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);
1740 1741
  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
  initBasicInfo(&pInfo->binfo, pResBlock);
1742 1743
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);

wmmhello's avatar
wmmhello 已提交
1744
  if (code != TSDB_CODE_SUCCESS) {
1745 1746 1747
    goto _error;
  }

1748
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
1749 1750

  pOperator->name = "StreamTimeIntervalAggOperator";
X
Xiaoyu Wang 已提交
1751
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL;
1752
  pOperator->blocking = true;
1753
  pOperator->status = OP_NOT_OPENED;
1754
  pOperator->exprSupp.pExprInfo = pExprInfo;
1755
  pOperator->pTaskInfo = pTaskInfo;
1756
  pOperator->exprSupp.numOfExprs = numOfCols;
1757 1758 1759 1760 1761 1762 1763 1764 1765 1766 1767 1768
  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 已提交
1769
_error:
1770 1771 1772 1773 1774 1775 1776 1777 1778 1779
  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;
1780
  SExprSupp*     pSup = &pOperator->exprSupp;
1781

1782
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
1783 1784

  bool    masterScan = true;
1785
  int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
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 1814 1815 1816
  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;
1817 1818
      int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx,
                                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1819 1820 1821 1822 1823 1824
      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);
1825
      doApplyFunctions(pTaskInfo, pSup->pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1826 1827 1828 1829 1830 1831 1832 1833 1834 1835
                       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];
1836 1837
  int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1838 1839 1840 1841 1842
  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);
1843
  doApplyFunctions(pTaskInfo, pSup->pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1844 1845 1846
                   pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
}

1847
static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
1848 1849 1850 1851 1852 1853
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SSessionAggOperatorInfo* pInfo = pOperator->info;
  SOptrBasicInfo*          pBInfo = &pInfo->binfo;
1854
  SExprSupp*               pSup = &pOperator->exprSupp;
1855 1856

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

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

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

1875 1876 1877
  int64_t st = taosGetTimestampUs();
  int32_t order = TSDB_ORDER_ASC;

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

  while (1) {
1881
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1882 1883 1884 1885 1886
    if (pBlock == NULL) {
      break;
    }

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

1890 1891 1892
    doSessionWindowAggImpl(pOperator, pInfo, pBlock);
  }

1893 1894
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;

1895 1896 1897 1898
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
  closeAllResultRows(&pBInfo->resultRowInfo);

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

1905 1906 1907 1908 1909
    bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
    if (!hasRemain) {
      doSetOperatorCompleted(pOperator);
      break;
    }
1910

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

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

    // 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;
1929
      char* val = colDataGetData(pColInfoData, rowIndex);
H
Haojun Liao 已提交
1930 1931 1932 1933 1934
      memcpy(pkey->pData, val, pkey->bytes);
    }
  }
}

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 2032 2033 2034
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;
  }

2035
  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049
  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 已提交
2050
static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
2051 2052 2053 2054
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

2055 2056
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

2057
  STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
2058 2059
  SSDataBlock*            pResBlock = pSliceInfo->pRes;
  SExprSupp*              pSup = &pOperator->exprSupp;
H
Haojun Liao 已提交
2060

2061 2062
  blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);

2063 2064 2065 2066 2067 2068 2069 2070
  //  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;
  //  }
2071

2072 2073
  int32_t        order = TSDB_ORDER_ASC;
  SInterval*     pInterval = &pSliceInfo->interval;
2074 2075
  SOperatorInfo* downstream = pOperator->pDownstream[0];

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

2083 2084 2085 2086 2087
    int32_t code = initPrevRowsKeeper(pSliceInfo, pBlock);
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
    }

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

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

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

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

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

2108
        pResBlock->info.rows += 1;
2109
        doKeepPrevRows(pSliceInfo, pBlock, i);
H
Haojun Liao 已提交
2110

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

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

            if (pSliceInfo->current > pSliceInfo->win.ekey) {
              doSetOperatorCompleted(pOperator);
              break;
H
Haojun Liao 已提交
2137 2138
            }
          } else {
H
Haojun Liao 已提交
2139
            // ignore current row, and do nothing
H
Haojun Liao 已提交
2140 2141
          }
        } else {  // it is the last row of current block
2142 2143 2144 2145 2146 2147 2148
          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);
2149 2150 2151
          if (pResBlock->info.rows >= pResBlock->info.capacity) {
            break;
          }
2152 2153 2154 2155 2156
        }

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

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

H
Haojun Liao 已提交
2168 2169 2170
  return pResBlock->info.rows == 0 ? NULL : pResBlock;
}

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

2178
  SInterpFuncPhysiNode* pInterpPhyNode = (SInterpFuncPhysiNode*)pPhyNode;
2179
  SExprSupp*            pSup = &pOperator->exprSupp;
2180

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

2188
  if (pInterpPhyNode->pExprs != NULL) {
2189
    int32_t    num = 0;
2190 2191 2192 2193 2194 2195 2196 2197 2198 2199
    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);
2200

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

2207
  pOperator->name = "TimeSliceOperator";
2208
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC;
2209 2210 2211 2212
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
2213

2214 2215
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doTimeslice, NULL, NULL, destroyBasicOperatorInfo, NULL, NULL, NULL);
2216

H
Haojun Liao 已提交
2217
  code = appendDownstream(pOperator, &downstream, 1);
2218 2219
  return pOperator;

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

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

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

2245 2246 2247
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;

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

2251
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
2252

L
Liu Jicong 已提交
2253
  pInfo->twAggSup = *pTwAggSup;
2254 2255
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

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

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

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

L
Liu Jicong 已提交
2272
_error:
2273 2274 2275 2276 2277 2278
  pTaskInfo->code = TSDB_CODE_SUCCESS;
  return NULL;
}

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

D
dapan1121 已提交
2281
  taosMemoryFreeClear(param);
2282 2283 2284
}

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

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

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

2302 2303
  initBasicInfo(&pInfo->binfo, pResBlock);

2304
  pInfo->twAggSup = *pTwAggSupp;
2305
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
2306 2307
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

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

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

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

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

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

5
54liuyao 已提交
2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353 2354 2355 2356 2357
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);
      }
    }
  }
}

2358 2359 2360 2361 2362 2363 2364 2365
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;
}

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

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

L
Liu Jicong 已提交
2407 2408 2409 2410
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 已提交
2411 2412 2413 2414 2415 2416 2417 2418 2419 2420 2421 2422 2423 2424
  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 已提交
2425 2426 2427 2428 2429 2430
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 已提交
2431
static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBlock, uint64_t tableGroupId,
S
shenglian zhou 已提交
2432
                           SArray* pUpdated) {
5
54liuyao 已提交
2433
  SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)pOperatorInfo->info;
X
Xiaoyu Wang 已提交
2434 2435
  SResultRowInfo*                   pResultRowInfo = &(pInfo->binfo.resultRowInfo);
  SExecTaskInfo*                    pTaskInfo = pOperatorInfo->pTaskInfo;
2436 2437
  SExprSupp*                        pSup = &pOperatorInfo->exprSupp;
  int32_t                           numOfOutput = pSup->numOfExprs;
X
Xiaoyu Wang 已提交
2438 2439 2440 2441 2442
  int32_t                           step = 1;
  bool                              ascScan = true;
  TSKEY*                            tsCols = NULL;
  SResultRow*                       pResult = NULL;
  int32_t                           forwardRows = 0;
5
54liuyao 已提交
2443

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

X
Xiaoyu Wang 已提交
2448 2449
  int32_t     startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1);
  TSKEY       ts = getStartTsKey(&pSDataBlock->info.window, tsCols);
5
54liuyao 已提交
2450 2451 2452 2453 2454 2455
  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 已提交
2456
  while (1) {
5
54liuyao 已提交
2457
    bool isClosed = isCloseWindow(&nextWin, &pInfo->twAggSup);
5
54liuyao 已提交
2458
    if ((pInfo->ignoreExpiredData && isClosed) || !inSlidingWindow(&pInfo->interval, &nextWin, &pSDataBlock->info)) {
5
54liuyao 已提交
2459 2460 2461 2462 2463 2464 2465
      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 已提交
2466 2467 2468 2469 2470
      bool    ignore = true;
      SWinRes winRes = {
          .ts = nextWin.skey,
          .groupId = tableGroupId,
      };
5
54liuyao 已提交
2471 2472 2473 2474 2475 2476 2477 2478 2479 2480
      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 已提交
2481
          chArray = *(void**)chIds;
5
54liuyao 已提交
2482 2483 2484
          int32_t chId = getChildIndex(pSDataBlock);
          index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
        }
L
Liu Jicong 已提交
2485
        if (index != -1 && pSDataBlock->info.type == STREAM_PULL_DATA) {
5
54liuyao 已提交
2486 2487 2488 2489 2490 2491
          taosArrayRemove(chArray, index);
          if (taosArrayGetSize(chArray) == 0) {
            // pull data is over
            taosHashRemove(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
          }
        }
L
Liu Jicong 已提交
2492
        if (index == -1 || pSDataBlock->info.type == STREAM_PULL_DATA) {
5
54liuyao 已提交
2493 2494 2495 2496 2497 2498 2499 2500 2501 2502 2503
          ignore = false;
        }
      }

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

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

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

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

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

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

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

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

5
54liuyao 已提交
2567 2568 2569 2570 2571 2572 2573 2574 2575 2576 2577 2578 2579 2580
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 已提交
2581
  blockDataEnsureCapacity(pBlock, size - (*pIndex));
5
54liuyao 已提交
2582 2583
  ASSERT(3 <= taosArrayGetSize(pBlock->pDataBlock));
  for (; (*pIndex) < size; (*pIndex)++) {
L
Liu Jicong 已提交
2584
    SPullWindowInfo* pWin = taosArrayGet(array, (*pIndex));
5
54liuyao 已提交
2585
    SColumnInfoData* pStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
5
54liuyao 已提交
2586 2587
    colDataAppend(pStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false);

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

5
54liuyao 已提交
2591
    SColumnInfoData* pGroupId = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
5
54liuyao 已提交
2592 2593 2594 2595 2596 2597 2598 2599 2600 2601
    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 已提交
2602
void processPullOver(SSDataBlock* pBlock, SHashObj* pMap) {
5
54liuyao 已提交
2603
  SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
2604
  TSKEY*           tsData = (TSKEY*)pStartCol->pData;
5
54liuyao 已提交
2605
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
L
Liu Jicong 已提交
2606 2607
  uint64_t*        groupIdData = (uint64_t*)pGroupCol->pData;
  int32_t          chId = getChildIndex(pBlock);
5
54liuyao 已提交
2608 2609
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    SWinRes winRes = {.ts = tsData[i], .groupId = groupIdData[i]};
L
Liu Jicong 已提交
2610
    void*   chIds = taosHashGet(pMap, &winRes, sizeof(SWinRes));
5
54liuyao 已提交
2611
    if (chIds) {
L
Liu Jicong 已提交
2612
      SArray* chArray = *(SArray**)chIds;
5
54liuyao 已提交
2613 2614 2615 2616 2617 2618 2619 2620 2621 2622 2623
      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 已提交
2624

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

2631 2632
  SExprSupp* pSup = &pOperator->exprSupp;

L
Liu Jicong 已提交
2633 2634
  qDebug("interval status %d %s", pOperator->status, IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");

5
54liuyao 已提交
2635 2636 2637 2638
  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 已提交
2639
    if (pInfo->binfo.pRes->info.rows == 0) {
5
54liuyao 已提交
2640
      pOperator->status = OP_EXEC_DONE;
5
54liuyao 已提交
2641 2642 2643
      if (!IS_FINAL_OP(pInfo)) {
        // semi interval operator clear disk buffer
        clearStreamIntervalOperator(pInfo);
2644 2645
      } else {
        freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2646
      }
5
54liuyao 已提交
2647 2648 2649 2650 2651 2652 2653 2654 2655 2656 2657 2658 2659 2660
      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 已提交
2661 2662
      // process the rest of the data
      return pInfo->pUpdateRes;
5
54liuyao 已提交
2663
    }
5
54liuyao 已提交
2664 2665 2666 2667 2668 2669 2670
    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;
    }
2671 2672 2673 2674 2675 2676
    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 已提交
2677 2678 2679 2680 2681
  }

  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
5
54liuyao 已提交
2682
      clearSpecialDataBlock(pInfo->pUpdateRes);
2683
      removeDeleteResults(pUpdated, pInfo->pDelWins);
5
54liuyao 已提交
2684
      pOperator->status = OP_RES_TO_RETURN;
L
Liu Jicong 已提交
2685
      qDebug("%s return data", IS_FINAL_OP(pInfo) ? "interval Final" : "interval  Semi");
5
54liuyao 已提交
2686 2687
      break;
    }
5
54liuyao 已提交
2688 2689
    printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval Final recv" : "interval  Semi recv");
    maxTs = TMAX(maxTs, pBlock->info.window.ekey);
2690

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

2703 2704 2705
        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 已提交
2706 2707
        taosArrayDestroy(pUpWins);
        continue;
2708
      }
5
54liuyao 已提交
2709 2710
      removeResults(pUpWins, pUpdated);
      copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
2711
      pInfo->returnUpdate = true;
2712
      taosArrayDestroy(pUpWins);
5
54liuyao 已提交
2713
      break;
2714 2715 2716 2717 2718 2719 2720 2721 2722
    } 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,
2723
                              pOperator->exprSupp.numOfExprs, pOperator->pTaskInfo, pUpdated);
2724 2725 2726 2727
        continue;
      }
      removeResults(pInfo->pDelWins, pUpdated);
      break;
5
54liuyao 已提交
2728
    } else if (pBlock->info.type == STREAM_GET_ALL && IS_FINAL_OP(pInfo)) {
5
54liuyao 已提交
2729 2730
      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
      continue;
5
54liuyao 已提交
2731
    } else if (pBlock->info.type == STREAM_RETRIEVE && !IS_FINAL_OP(pInfo)) {
2732
      SArray* pUpWins = taosArrayInit(8, sizeof(SWinRes));
5
54liuyao 已提交
2733
      doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock, pUpWins);
5
54liuyao 已提交
2734 2735 2736 2737 2738 2739
      removeResults(pUpWins, pUpdated);
      taosArrayDestroy(pUpWins);
      if (taosArrayGetSize(pUpdated) > 0) {
        break;
      }
      continue;
L
Liu Jicong 已提交
2740 2741
    } else if (pBlock->info.type == STREAM_PULL_OVER && IS_FINAL_OP(pInfo)) {
      processPullOver(pBlock, pInfo->pPullDataMap);
5
54liuyao 已提交
2742
      continue;
5
54liuyao 已提交
2743
    }
5
54liuyao 已提交
2744

5
54liuyao 已提交
2745 2746 2747 2748
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
2749
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
5
54liuyao 已提交
2750
    doHashInterval(pOperator, pBlock, pBlock->info.groupId, pUpdated);
5
54liuyao 已提交
2751
    if (IS_FINAL_OP(pInfo)) {
S
shenglian zhou 已提交
2752
      int32_t chIndex = getChildIndex(pBlock);
5
54liuyao 已提交
2753 2754 2755 2756 2757 2758 2759
      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);
        }
2760 2761
        SStreamFinalIntervalOperatorInfo* pTmpInfo = pChildOp->info;
        pTmpInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
5
54liuyao 已提交
2762 2763
        taosArrayPush(pInfo->pChildren, &pChildOp);
      }
S
shenglian zhou 已提交
2764
      SOperatorInfo*                    pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
5
54liuyao 已提交
2765
      SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info;
2766
      setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, pChInfo->order, MAIN_SCAN, true);
5
54liuyao 已提交
2767
      doHashInterval(pChildOp, pBlock, pBlock->info.groupId, NULL);
5
54liuyao 已提交
2768 2769 2770 2771

      if (needBreak(pInfo)) {
        break;
      }
5
54liuyao 已提交
2772 2773
    }
  }
S
shenglian zhou 已提交
2774

5
54liuyao 已提交
2775
  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
5
54liuyao 已提交
2776
  if (IS_FINAL_OP(pInfo)) {
2777 2778
    closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pInfo->pPullDataMap,
                        pUpdated, pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2779
    closeChildIntervalWindow(pInfo->pChildren, pInfo->twAggSup.maxTs);
5
54liuyao 已提交
2780 2781
  }

2782
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
2783 2784 2785
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2786 2787 2788 2789 2790 2791 2792 2793 2794
  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 已提交
2795 2796 2797
    // process the rest of the data
    return pInfo->pUpdateRes;
  }
5
54liuyao 已提交
2798 2799 2800 2801 2802 2803 2804 2805

  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;
  }
2806 2807 2808 2809 2810 2811 2812

  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 已提交
2813 2814 2815 2816 2817 2818 2819 2820 2821 2822 2823 2824 2825 2826 2827 2828 2829 2830 2831 2832 2833 2834 2835 2836 2837 2838
  // 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 已提交
2839 2840
}

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

5
54liuyao 已提交
2850
  pInfo->order = TSDB_ORDER_ASC;
S
shenglian zhou 已提交
2851 2852 2853 2854 2855 2856 2857 2858
  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 已提交
2859 2860
      .calTrigger = pIntervalPhyNode->window.triggerType,
      .maxTs = INT64_MIN,
S
shenglian zhou 已提交
2861
  };
2862
  ASSERT(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY);
5
54liuyao 已提交
2863 2864 2865
  pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
2866 2867 2868 2869 2870 2871 2872 2873 2874
  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 已提交
2875 2876
  int32_t      numOfCols = 0;
  SExprInfo*   pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
5
54liuyao 已提交
2877
  SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
2878 2879 2880 2881

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

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

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

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

5
54liuyao 已提交
2919
  if (!IS_FINAL_OP(pInfo) || numOfChild == 0) {
5
54liuyao 已提交
2920 2921
    pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
  }
5
54liuyao 已提交
2922 2923 2924 2925 2926
  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 已提交
2927
  pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired;
5
54liuyao 已提交
2928
  // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete
2929 2930
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
2931 2932
  pInfo->delIndex = 0;
  pInfo->pDelWins = taosArrayInit(4, sizeof(SWinRes));
5
54liuyao 已提交
2933

5
54liuyao 已提交
2934
  pOperator->operatorType = pPhyNode->type;
5
54liuyao 已提交
2935 2936
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
2937
  pOperator->exprSupp.pExprInfo = pExprInfo;
5
54liuyao 已提交
2938
  pOperator->pTaskInfo = pTaskInfo;
2939
  pOperator->exprSupp.numOfExprs = numOfCols;
5
54liuyao 已提交
2940 2941
  pOperator->info = pInfo;

S
shenglian zhou 已提交
2942 2943 2944
  pOperator->fpSet =
      createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, NULL, destroyStreamFinalIntervalOperatorInfo,
                          aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
2945 2946 2947 2948 2949 2950 2951 2952 2953 2954 2955 2956 2957 2958

  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 已提交
2959
}
5
54liuyao 已提交
2960 2961 2962

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

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

D
dapan1121 已提交
2988
  taosMemoryFreeClear(param);
5
54liuyao 已提交
2989 2990
}

2991 2992
int32_t initBasicInfoEx(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfCols,
                        SSDataBlock* pResultBlock) {
2993 2994 2995 2996 2997
  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

2998
  initBasicInfo(pBasicInfo, pResultBlock);
2999

5
54liuyao 已提交
3000
  for (int32_t i = 0; i < numOfCols; ++i) {
3001
    pSup->pCtx[i].pBuf = NULL;
5
54liuyao 已提交
3002
  }
3003

3004
  ASSERT(numOfCols > 0);
3005
  increaseTs(pSup->pCtx);
5
54liuyao 已提交
3006 3007 3008 3009 3010 3011 3012 3013
  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 已提交
3014 3015
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int64_t gap, int64_t waterMark,
                    uint8_t type) {
5
54liuyao 已提交
3016
  ASSERT(downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
3017
  SStreamScanInfo* pScanInfo = downstream->info;
X
Xiaoyu Wang 已提交
3018
  pScanInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = pAggSup, .gap = gap, .parentType = type};
5
54liuyao 已提交
3019
  pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, waterMark);
5
54liuyao 已提交
3020 3021
}

3022 3023
int32_t initSessionAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx,
                                int32_t numOfOutput) {
3024 3025 3026
  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SResultWindowInfo));
}

3027 3028 3029 3030 3031 3032 3033
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 已提交
3034
  SStreamSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamSessionAggOperatorInfo));
3035
  SOperatorInfo*                 pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
3036 3037 3038 3039 3040
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
3041 3042 3043 3044 3045 3046 3047 3048
  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;
    }
  }
3049
  SExprSupp* pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3050

3051
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
5
54liuyao 已提交
3052 3053 3054
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
3055

3056
  code = initSessionAggSupporter(&pInfo->streamAggSup, "StreamSessionAggOperatorInfo", pSup->pCtx, numOfCols);
5
54liuyao 已提交
3057 3058 3059
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
X
Xiaoyu Wang 已提交
3060

5
54liuyao 已提交
3061 3062 3063 3064
  pInfo->pDummyCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfCols, sizeof(SqlFunctionCtx));
  if (pInfo->pDummyCtx == NULL) {
    goto _error;
  }
3065
  initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols);
5
54liuyao 已提交
3066

3067 3068
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pSessionNode->window.watermark, .calTrigger = pSessionNode->window.triggerType, .maxTs = INT64_MIN};
H
Haojun Liao 已提交
3069 3070

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

5
54liuyao 已提交
3073 3074 3075 3076
  pInfo->primaryTsIndex = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
  if (pSessionNode->window.pTsEnd) {
    pInfo->endTsIndex = ((SColumnNode*)pSessionNode->window.pTsEnd)->slotId;
  }
3077
  pInfo->gap = pSessionNode->gap;
5
54liuyao 已提交
3078 3079 3080 3081 3082
  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 已提交
3083
  // pInfo->pDelRes = createPullDataBlock();
3084 3085
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
3086
  pInfo->pChildren = NULL;
5
54liuyao 已提交
3087 3088
  pInfo->isFinal = false;
  pInfo->pPhyNode = pPhyNode;
5
54liuyao 已提交
3089
  pInfo->ignoreExpiredData = pSessionNode->window.igExpired;
5
54liuyao 已提交
3090
  pInfo->returnDelete = false;
5
54liuyao 已提交
3091 3092

  pOperator->name = "StreamSessionWindowAggOperator";
3093
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
5
54liuyao 已提交
3094 3095
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
3096 3097
  pOperator->exprSupp.pExprInfo = pExprInfo;
  pOperator->exprSupp.numOfExprs = numOfCols;
5
54liuyao 已提交
3098
  pOperator->info = pInfo;
X
Xiaoyu Wang 已提交
3099 3100 3101
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, NULL, destroyStreamSessionAggOperatorInfo,
                          aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
3102
  pOperator->pTaskInfo = pTaskInfo;
5
54liuyao 已提交
3103 3104 3105 3106
  if (downstream) {
    initDownStream(downstream, &pInfo->streamAggSup, pInfo->gap, pInfo->twAggSup.waterMark, pOperator->operatorType);
    code = appendDownstream(pOperator, &downstream, 1);
  }
5
54liuyao 已提交
3107 3108 3109 3110 3111 3112 3113 3114 3115 3116 3117 3118 3119 3120
  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 已提交
3121
  SArray*            pWinInfos = (SArray*)data;
5
54liuyao 已提交
3122 3123 3124
  SResultWindowInfo* pWin = taosArrayGet(pWinInfos, index);
  return pWin->win.ekey;
}
5
54liuyao 已提交
3125 3126

bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap) {
5
54liuyao 已提交
3127
  if (ts + gap >= pWin->skey && ts - gap <= pWin->ekey) {
5
54liuyao 已提交
3128 3129 3130 3131 3132
    return true;
  }
  return false;
}

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

X
Xiaoyu Wang 已提交
3135 3136
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 已提交
3137 3138 3139 3140
  return taosArrayInsert(pWinInfos, index, &win);
}

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

3145
SArray* getWinInfos(SStreamAggSupporter* pAggSup, uint64_t groupId) {
3146
  void**  ite = taosHashGet(pAggSup->pResultRows, &groupId, sizeof(uint64_t));
3147 3148 3149
  SArray* pWinInfos = NULL;
  if (ite == NULL) {
    pWinInfos = taosArrayInit(1024, pAggSup->valueSize);
3150
    taosHashPut(pAggSup->pResultRows, &groupId, sizeof(uint64_t), &pWinInfos, sizeof(void*));
3151 3152 3153 3154 3155 3156
  } else {
    pWinInfos = *ite;
  }
  return pWinInfos;
}

5
54liuyao 已提交
3157 3158
// don't add new window
SResultWindowInfo* getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
3159
                                       int64_t gap, int32_t* pIndex) {
5
54liuyao 已提交
3160 3161 3162 3163 3164 3165 3166 3167
  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
3168
  int32_t            index = binarySearch(pWinInfos, size, startTs, TSDB_ORDER_DESC, getSessionWindowEndkey);
5
54liuyao 已提交
3169 3170 3171 3172 3173 3174 3175 3176 3177 3178 3179 3180 3181 3182 3183
  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)) {
3184
      *pIndex = index + 1;
5
54liuyao 已提交
3185 3186 3187 3188 3189 3190 3191
      return pWin;
    }
  }

  return NULL;
}

3192 3193
SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
                                        int64_t gap, int32_t* pIndex) {
3194 3195 3196
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;

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

  if (index + 1 < size) {
    pWin = taosArrayGet(pWinInfos, index + 1);
5
54liuyao 已提交
3215
    if (isInWindow(pWin, startTs, gap)) {
5
54liuyao 已提交
3216 3217
      *pIndex = index + 1;
      return pWin;
5
54liuyao 已提交
3218 3219 3220
    } else if (endTs != INT64_MIN && isInWindow(pWin, endTs, gap)) {
      *pIndex = index;
      return pWin;
5
54liuyao 已提交
3221 3222 3223 3224 3225
    }
  }

  if (index == size - 1) {
    *pIndex = taosArrayGetSize(pWinInfos);
5
54liuyao 已提交
3226
    return addNewSessionWindow(pWinInfos, startTs);
5
54liuyao 已提交
3227
  }
5
54liuyao 已提交
3228
  *pIndex = index + 1;
5
54liuyao 已提交
3229
  return insertNewSessionWindow(pWinInfos, startTs, index + 1);
5
54liuyao 已提交
3230 3231
}

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

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

5
54liuyao 已提交
3263
  if (pWinInfo->pos.pageId == -1) {
3264
    *pResult = getNewResultRow(pAggSup->pResultBuf, groupId, pAggSup->resultRowSize);
5
54liuyao 已提交
3265 3266 3267 3268 3269 3270 3271 3272 3273 3274 3275 3276 3277 3278 3279 3280 3281 3282
    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;
3283
  setResultRowInitCtx(*pResult, pCtx, numOfOutput, rowEntryInfoOffset);
5
54liuyao 已提交
3284 3285 3286
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
3287 3288 3289
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,
3290
                                  int32_t numOutput, SOperatorInfo* pOperator) {
3291
  SExprSupp*     pSup = &pOperator->exprSupp;
3292 3293
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

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

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

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

5
54liuyao 已提交
3324 3325
int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) {
  SResultWindowInfo* pCurWin = taosArrayGet(pWinInfos, startIndex);
X
Xiaoyu Wang 已提交
3326
  int32_t            size = taosArrayGetSize(pWinInfos);
5
54liuyao 已提交
3327 3328 3329 3330 3331 3332 3333 3334 3335 3336 3337
  // 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 已提交
3338
void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, uint64_t groupId,
3339
                       int32_t numOfOutput, SHashObj* pStUpdated, SHashObj* pStDeleted, SOperatorInfo* pOperator) {
3340
  SExprSupp*     pSup = &pOperator->exprSupp;
3341 3342
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

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

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

  int32_t     step = 1;
  bool        ascScan = true;
5
54liuyao 已提交
3384 3385
  TSKEY*      startTsCols = NULL;
  TSKEY*      endTsCols = NULL;
5
54liuyao 已提交
3386
  SResultRow* pResult = NULL;
X
Xiaoyu Wang 已提交
3387
  int32_t     winRows = 0;
5
54liuyao 已提交
3388

5
54liuyao 已提交
3389 3390 3391 3392 3393 3394
  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 已提交
3395
  } else {
5
54liuyao 已提交
3396
    pEndTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
3397
  }
5
54liuyao 已提交
3398
  endTsCols = (int64_t*)pEndTsCol->pData;
X
Xiaoyu Wang 已提交
3399

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

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

5
54liuyao 已提交
3432 3433 3434 3435 3436
void deleteWindow(SArray* pWinInfos, int32_t index) {
  ASSERT(index >= 0 && index < taosArrayGetSize(pWinInfos));
  taosArrayRemove(pWinInfos, index);
}

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

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

5
54liuyao 已提交
3482
static int32_t copyUpdateResult(SHashObj* pStUpdated, SArray* pUpdated) {
X
Xiaoyu Wang 已提交
3483
  void*  pData = NULL;
5
54liuyao 已提交
3484
  size_t keyLen = 0;
X
Xiaoyu Wang 已提交
3485
  while ((pData = taosHashIterate(pStUpdated, pData)) != NULL) {
5
54liuyao 已提交
3486 3487 3488 3489 3490 3491
    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 已提交
3492
    pos->groupId = ((SWinRes*)pData)->groupId;
5
54liuyao 已提交
3493
    pos->pos = *(SResultRowPosition*)key;
5
54liuyao 已提交
3494
    *(int64_t*)pos->key = ((SWinRes*)pData)->ts;
5
54liuyao 已提交
3495 3496
    taosArrayPush(pUpdated, &pos);
  }
5
54liuyao 已提交
3497
  taosArraySort(pUpdated, resultrowComparAsc);
5
54liuyao 已提交
3498 3499 3500 3501 3502
  return TSDB_CODE_SUCCESS;
}

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

X
Xiaoyu Wang 已提交
3526
static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWinArray, int32_t groupId,
3527
                              int32_t numOfOutput, SOperatorInfo* pOperator) {
3528
  SExprSupp*     pSup = &pOperator->exprSupp;
3529 3530
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

3531 3532
  int32_t size = taosArrayGetSize(pWinArray);
  ASSERT(pInfo->pChildren);
3533

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

X
Xiaoyu Wang 已提交
3571
typedef SResultWindowInfo* (*__get_win_info_)(void*);
5
54liuyao 已提交
3572 3573
SResultWindowInfo* getResWinForSession(void* pData) { return (SResultWindowInfo*)pData; }
SResultWindowInfo* getResWinForState(void* pData) { return &((SStateWindowInfo*)pData)->winInfo; }
5
54liuyao 已提交
3574

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

5
54liuyao 已提交
3611 3612 3613 3614 3615 3616 3617 3618 3619 3620
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);
  }
}

3621
int32_t getAllSessionWindow(SHashObj* pHashMap, SArray* pClosed, __get_win_info_ fn) {
3622
  void** pIte = NULL;
3623
  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
3624
    SArray* pWins = (SArray*)(*pIte);
3625 3626 3627 3628 3629 3630 3631 3632
    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 已提交
3633 3634 3635 3636 3637
    }
  }
  return TSDB_CODE_SUCCESS;
}

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

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

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

5
54liuyao 已提交
3712 3713 3714 3715
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
3716
    // the pDataBlock are always the same one, no need to call this again
3717
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
3718 3719 3720 3721 3722 3723
    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++) {
3724 3725
        SOperatorInfo* pChildOp =
            createStreamFinalSessionAggOperatorInfo(NULL, pInfo->pPhyNode, pOperator->pTaskInfo, 0);
5
54liuyao 已提交
3726 3727 3728 3729 3730
        if (!pChildOp) {
          longjmp(pOperator->pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
        }
        taosArrayPush(pInfo->pChildren, &pChildOp);
      }
3731
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
5
54liuyao 已提交
3732 3733
      setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
      doStreamSessionAggImpl(pChildOp, pBlock, NULL, NULL, true);
3734
    }
5
54liuyao 已提交
3735
    maxTs = TMAX(maxTs, pBlock->info.window.ekey);
5
54liuyao 已提交
3736
  }
5
54liuyao 已提交
3737 3738

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

3742 3743
  closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated, getResWinForSession,
                     pInfo->ignoreExpiredData);
5
54liuyao 已提交
3744
  closeChildSessionWindow(pInfo->pChildren, pInfo->twAggSup.maxTs, pInfo->ignoreExpiredData);
5
54liuyao 已提交
3745 3746 3747
  copyUpdateResult(pStUpdated, pUpdated);
  taosHashCleanup(pStUpdated);

3748
  finalizeUpdatedResult(pSup->numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
3749 3750 3751 3752
  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 已提交
3753
    printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3754 3755 3756
    return pInfo->pDelRes;
  }
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
L
Liu Jicong 已提交
3757
  printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3758 3759 3760 3761
  return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
}

static void clearStreamSessionOperator(SStreamSessionAggOperatorInfo* pInfo) {
3762
  void** pIte = NULL;
5
54liuyao 已提交
3763
  while ((pIte = taosHashIterate(pInfo->streamAggSup.pResultRows, pIte)) != NULL) {
3764
    SArray* pWins = (SArray*)(*pIte);
5
54liuyao 已提交
3765 3766 3767 3768 3769 3770 3771 3772 3773 3774 3775 3776 3777 3778 3779 3780 3781 3782 3783 3784 3785 3786 3787 3788 3789
    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;
3790

5
54liuyao 已提交
3791 3792 3793
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
5
54liuyao 已提交
3794 3795 3796 3797 3798 3799 3800 3801 3802
    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 已提交
3803
      printDataBlock(pInfo->pDelRes, "Semi  Session");
5
54liuyao 已提交
3804 3805
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
3806 3807

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

  _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 已提交
3826
      clearSpecialDataBlock(pInfo->pUpdateRes);
5
54liuyao 已提交
3827 3828 3829
      break;
    }

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

5
54liuyao 已提交
3848 3849 3850 3851
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
5
54liuyao 已提交
3852 3853 3854 3855 3856 3857 3858 3859 3860 3861 3862 3863
    // 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 已提交
3864
  copyUpdateResult(pStUpdated, pUpdated);
5
54liuyao 已提交
3865
  taosHashCleanup(pStUpdated);
5
54liuyao 已提交
3866

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

  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 已提交
3881
    printDataBlock(pInfo->pDelRes, "Semi  Session");
5
54liuyao 已提交
3882 3883
    return pInfo->pDelRes;
  }
5
54liuyao 已提交
3884 3885

  if (pInfo->pUpdateRes->info.rows > 0) {
5
54liuyao 已提交
3886 3887
    // process the rest of the data
    pOperator->status = OP_OPENED;
5
54liuyao 已提交
3888
    printDataBlock(pInfo->pUpdateRes, "Semi  Session");
5
54liuyao 已提交
3889 3890
    return pInfo->pUpdateRes;
  }
5
54liuyao 已提交
3891 3892 3893

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

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

  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 已提交
3911
    pInfo->pUpdateRes->info.type = STREAM_CLEAR;
5
54liuyao 已提交
3912 3913 3914
    blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
    pOperator->name = "StreamSessionSemiAggOperator";
    pOperator->fpSet =
3915 3916
        createOperatorFpSet(operatorDummyOpenFn, doStreamSessionSemiAgg, NULL, NULL,
                            destroyStreamSessionAggOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
3917 3918 3919 3920 3921
  }
  pOperator->operatorType = pPhyNode->type;
  if (numOfChild > 0) {
    pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
    for (int32_t i = 0; i < numOfChild; i++) {
3922
      SOperatorInfo* pChild = createStreamFinalSessionAggOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
5
54liuyao 已提交
3923 3924 3925 3926
      if (pChild == NULL) {
        goto _error;
      }
      taosArrayPush(pInfo->pChildren, &pChild);
3927 3928 3929 3930 3931 3932
    }
  }
  return pOperator;

_error:
  if (pInfo != NULL) {
3933
    destroyStreamSessionAggOperatorInfo(pInfo, pOperator->exprSupp.numOfExprs);
3934 3935 3936 3937 3938 3939 3940
  }

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

void destroyStreamStateOperatorInfo(void* param, int32_t numOfOutput) {
X
Xiaoyu Wang 已提交
3943
  SStreamStateAggOperatorInfo* pInfo = (SStreamStateAggOperatorInfo*)param;
3944
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
3945 3946 3947 3948 3949
  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 已提交
3950
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, i);
5
54liuyao 已提交
3951 3952 3953 3954 3955 3956
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
      destroyStreamSessionAggOperatorInfo(pChInfo, numOfOutput);
      taosMemoryFreeClear(pChild);
      taosMemoryFreeClear(pChInfo);
    }
  }
D
dapan1121 已提交
3957 3958

  taosMemoryFreeClear(param);
5
54liuyao 已提交
3959 3960 3961 3962 3963 3964 3965
}

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

X
Xiaoyu Wang 已提交
3966 3967 3968 3969 3970 3971 3972 3973 3974 3975 3976 3977
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 已提交
3978 3979 3980 3981 3982 3983 3984 3985
  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 已提交
3986 3987 3988 3989 3990 3991 3992 3993 3994 3995 3996 3997
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 已提交
3998 3999 4000 4001 4002 4003 4004 4005 4006 4007 4008 4009 4010 4011 4012 4013 4014 4015 4016
  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);
}

4017 4018 4019
SStateWindowInfo* getStateWindowByTs(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int32_t* pIndex) {
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;
X
Xiaoyu Wang 已提交
4020 4021
  int32_t           size = taosArrayGetSize(pWinInfos);
  int32_t           index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
5
54liuyao 已提交
4022 4023 4024 4025 4026 4027 4028 4029 4030 4031 4032 4033 4034 4035 4036 4037 4038 4039 4040 4041
  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;
}

4042 4043
SStateWindowInfo* getStateWindow(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, char* pKeyData,
                                 SColumn* pCol, int32_t* pIndex) {
4044 4045
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;
5
54liuyao 已提交
4046 4047 4048 4049 4050
  int32_t size = taosArrayGetSize(pWinInfos);
  if (size == 0) {
    *pIndex = 0;
    return addNewStateWindow(pWinInfos, ts, pKeyData, pCol);
  }
X
Xiaoyu Wang 已提交
4051
  int32_t           index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
5
54liuyao 已提交
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 4080 4081 4082 4083 4084
  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 已提交
4085 4086
int32_t updateStateWindowInfo(SArray* pWinInfos, int32_t winIndex, TSKEY* pTs, SColumnInfoData* pKeyCol, int32_t rows,
                              int32_t start, bool* allEqual, SHashObj* pSeDelete) {
5
54liuyao 已提交
4087 4088 4089 4090 4091
  *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 已提交
4092
      if (isEqualStateKey(pWinInfo, pKeyData)) {
5
54liuyao 已提交
4093 4094 4095 4096 4097 4098 4099 4100 4101 4102 4103 4104 4105 4106
        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 已提交
4107 4108
        taosHashPut(pSeDelete, &pWinInfo->winInfo.pos, sizeof(SResultRowPosition), &pWinInfo->winInfo.win.skey,
                    sizeof(TSKEY));
5
54liuyao 已提交
4109 4110 4111 4112 4113 4114 4115 4116 4117 4118 4119 4120
        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 已提交
4121 4122
static void doClearStateWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, int32_t tsIndex, SColumn* pCol,
                                int32_t keyIndex, SHashObj* pSeUpdated, SHashObj* pSeDeleted) {
5
54liuyao 已提交
4123 4124
  SColumnInfoData* pTsColInfo = taosArrayGet(pBlock->pDataBlock, tsIndex);
  SColumnInfoData* pKeyColInfo = taosArrayGet(pBlock->pDataBlock, keyIndex);
X
Xiaoyu Wang 已提交
4125 4126 4127
  TSKEY*           tsCol = (TSKEY*)pTsColInfo->pData;
  bool             allEqual = false;
  int32_t          step = 1;
5
54liuyao 已提交
4128
  for (int32_t i = 0; i < pBlock->info.rows; i += step) {
X
Xiaoyu Wang 已提交
4129 4130
    char*             pKeyData = colDataGetData(pKeyColInfo, i);
    int32_t           winIndex = 0;
4131
    SStateWindowInfo* pCurWin = getStateWindowByTs(pAggSup, tsCol[i], pBlock->info.groupId, &winIndex);
5
54liuyao 已提交
4132 4133 4134
    if (!pCurWin) {
      continue;
    }
4135
    step = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual,
X
Xiaoyu Wang 已提交
4136
                                 pSeDeleted);
5
54liuyao 已提交
4137 4138 4139
    ASSERT(isTsInWindow(pCurWin, tsCol[i]) || isEqualStateKey(pCurWin, pKeyData));
    taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
    taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
4140
    deleteWindow(pAggSup->pCurWins, winIndex);
5
54liuyao 已提交
4141 4142 4143
  }
}

X
Xiaoyu Wang 已提交
4144 4145 4146
static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SHashObj* pSeUpdated,
                                 SHashObj* pStDeleted) {
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
4147
  SStreamStateAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
4148
  bool                         masterScan = true;
4149
  int32_t                      numOfOutput = pOperator->exprSupp.numOfExprs;
X
Xiaoyu Wang 已提交
4150 4151 4152 4153 4154 4155 4156
  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 已提交
4157
  if (pSDataBlock->pDataBlock != NULL) {
X
Xiaoyu Wang 已提交
4158 4159
    SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
    tsCols = (int64_t*)pColDataInfo->pData;
5
54liuyao 已提交
4160
  } else {
X
Xiaoyu Wang 已提交
4161
    return;
5
54liuyao 已提交
4162
  }
X
Xiaoyu Wang 已提交
4163

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

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

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

5
54liuyao 已提交
4233
    if (pBlock->info.type == STREAM_CLEAR) {
X
Xiaoyu Wang 已提交
4234 4235
      doClearStateWindows(&pInfo->streamAggSup, pBlock, pInfo->primaryTsIndex, &pInfo->stateCol, pInfo->stateCol.slotId,
                          pSeUpdated, pInfo->pSeDeleted);
5
54liuyao 已提交
4236
      continue;
4237 4238 4239 4240 4241 4242
    } 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;
4243
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
4244
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getResWinForState);
5
54liuyao 已提交
4245
      continue;
5
54liuyao 已提交
4246
    }
4247

5
54liuyao 已提交
4248 4249 4250 4251
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
4252
    // the pDataBlock are always the same one, no need to call this again
4253
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
4254 4255 4256 4257 4258
    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 已提交
4259

4260 4261
  closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated, getResWinForState,
                     pInfo->ignoreExpiredData);
5
54liuyao 已提交
4262
  closeChildSessionWindow(pInfo->pChildren, pInfo->twAggSup.maxTs, pInfo->ignoreExpiredData);
5
54liuyao 已提交
4263
  copyUpdateResult(pSeUpdated, pUpdated);
5
54liuyao 已提交
4264 4265
  taosHashCleanup(pSeUpdated);

4266 4267
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated,
                        pSup->rowEntryInfoOffset);
5
54liuyao 已提交
4268 4269 4270 4271
  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 已提交
4272
    printDataBlock(pInfo->pDelRes, "single state");
5
54liuyao 已提交
4273 4274
    return pInfo->pDelRes;
  }
X
Xiaoyu Wang 已提交
4275
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
5
54liuyao 已提交
4276
  printDataBlock(pBInfo->pRes, "single state");
5
54liuyao 已提交
4277 4278 4279
  return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
}

4280 4281 4282 4283
int32_t initStateAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SStateWindowInfo));
}

X
Xiaoyu Wang 已提交
4284 4285 4286 4287 4288 4289
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;
4290
  int32_t                      code = TSDB_CODE_OUT_OF_MEMORY;
5
54liuyao 已提交
4291

X
Xiaoyu Wang 已提交
4292 4293
  SStreamStateAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamStateAggOperatorInfo));
  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
4294 4295 4296 4297
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

4298 4299
  SExprSupp* pSup = &pOperator->exprSupp;

X
Xiaoyu Wang 已提交
4300
  int32_t    numOfCols = 0;
5
54liuyao 已提交
4301 4302 4303 4304
  SExprInfo* pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &numOfCols);

  pInfo->stateCol = extractColumnFromColumnNode(pColNode);
  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
4305 4306 4307 4308 4309 4310 4311 4312 4313
  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;
    }
  }

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

4322
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
5
54liuyao 已提交
4323 4324 4325 4326
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

4327
  code = initStateAggSupporter(&pInfo->streamAggSup, "StreamStateAggOperatorInfo", pSup->pCtx, numOfCols);
5
54liuyao 已提交
4328 4329 4330 4331 4332 4333 4334 4335 4336
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

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

4337
  initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols);
5
54liuyao 已提交
4338 4339 4340 4341 4342
  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 已提交
4343
  // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete
4344 4345
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
5
54liuyao 已提交
4346
  pInfo->pChildren = NULL;
5
54liuyao 已提交
4347
  pInfo->ignoreExpiredData = pStateNode->window.igExpired;
5
54liuyao 已提交
4348 4349

  pOperator->name = "StreamStateAggOperator";
4350
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
5
54liuyao 已提交
4351 4352
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
4353 4354
  pOperator->exprSupp.numOfExprs = numOfCols;
  pOperator->exprSupp.pExprInfo = pExprInfo;
5
54liuyao 已提交
4355 4356
  pOperator->pTaskInfo = pTaskInfo;
  pOperator->info = pInfo;
X
Xiaoyu Wang 已提交
4357 4358 4359
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, NULL,
                                         destroyStreamStateOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
  initDownStream(downstream, &pInfo->streamAggSup, 0, pInfo->twAggSup.waterMark, pOperator->operatorType);
5
54liuyao 已提交
4360 4361 4362 4363 4364 4365 4366 4367 4368 4369 4370 4371 4372
  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;
}
4373

4374 4375
void destroyMergeAlignedIntervalOperatorInfo(void* param, int32_t numOfOutput) {
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = (SMergeAlignedIntervalAggOperatorInfo*)param;
D
dapan1121 已提交
4376 4377
  destroyIntervalOperatorInfo(miaInfo->intervalAggOperatorInfo, numOfOutput);

D
dapan1121 已提交
4378
  taosMemoryFreeClear(param);
4379 4380
}

4381 4382
static int32_t outputMergeAlignedIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId,
                                                SSDataBlock* pResultBlock, TSKEY wstartTs) {
4383
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
D
dapan1121 已提交
4384
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4385
  SExecTaskInfo*                        pTaskInfo = pOperatorInfo->pTaskInfo;
4386 4387

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

4390 4391 4392 4393
  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);
4394

4395 4396
  finalizeResultRowIntoResultDataBlock(iaInfo->aggSup.pResultBuf, p1, pSup->pCtx, pSup->pExprInfo, pSup->numOfExprs,
                                       pSup->rowEntryInfoOffset, pResultBlock, pTaskInfo);
4397
  taosHashRemove(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf, GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
4398

4399
  return 0;
4400 4401
}

4402 4403
static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo,
                                          SSDataBlock* pBlock, int32_t scanFlag, SSDataBlock* pResultBlock) {
4404
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
D
dapan1121 已提交
4405
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4406 4407

  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
4408
  SExprSupp*     pSup = &pOperatorInfo->exprSupp;
4409 4410

  int32_t     startPos = 0;
4411
  int32_t     numOfOutput = pSup->numOfExprs;
4412
  int64_t*    tsCols = extractTsCol(pBlock, iaInfo);
4413 4414 4415 4416
  uint64_t    tableGroupId = pBlock->info.groupId;
  TSKEY       blockStartTs = getStartTsKey(&pBlock->info.window, tsCols);
  SResultRow* pResult = NULL;

4417 4418
  STimeWindow win;
  win.skey = blockStartTs;
4419 4420
  win.ekey =
      taosTimeAdd(win.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1;
4421

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

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

4444
      outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs);
4445 4446 4447

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

4463
  outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs);
4464 4465
}

4466
static SSDataBlock* doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
S
shenglian zhou 已提交
4467
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
4468

4469
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperator->info;
D
dapan1121 已提交
4470
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4471 4472 4473 4474
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

4475
  SExprSupp*   pSup = &pOperator->exprSupp;
4476
  SSDataBlock* pRes = iaInfo->binfo.pRes;
4477
  blockDataCleanup(pRes);
4478
  blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
4479

4480 4481 4482 4483 4484 4485 4486 4487 4488 4489
  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;
4490
        miaInfo->prefetchedBlock = NULL;
4491
      }
4492

4493 4494 4495 4496
      if (pBlock == NULL) {
        miaInfo->inputBlocksFinished = true;
        break;
      }
4497

4498 4499 4500 4501 4502 4503 4504
      if (!miaInfo->hasGroupId) {
        miaInfo->hasGroupId = true;
        miaInfo->groupId = pBlock->info.groupId;
      } else if (miaInfo->groupId != pBlock->info.groupId) {
        miaInfo->prefetchedBlock = pBlock;
        break;
      }
4505

4506
      getTableScanInfo(pOperator, &iaInfo->order, &scanFlag);
4507
      setInputDataBlock(pOperator, pSup->pCtx, pBlock, iaInfo->order, scanFlag, true);
4508
      doMergeAlignedIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);
4509 4510
      doFilter(miaInfo->pCondition, pRes);
      if (pRes->info.rows > 0) {
4511 4512 4513 4514 4515
        break;
      }
    }

    pRes->info.groupId = miaInfo->groupId;
4516 4517
  }

4518
  if (miaInfo->inputBlocksFinished) {
4519 4520 4521 4522 4523 4524 4525 4526
    doSetOperatorCompleted(pOperator);
  }

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

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

D
dapan1121 已提交
4537 4538 4539 4540 4541 4542
  miaInfo->intervalAggOperatorInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
  if (miaInfo->intervalAggOperatorInfo == NULL) {
    goto _error;
  }

  SIntervalAggOperatorInfo* iaInfo = miaInfo->intervalAggOperatorInfo;
4543
  SExprSupp*                pSup = &pOperator->exprSupp;
4544

4545
  miaInfo->pCondition = pCondition;
4546 4547 4548
  iaInfo->win = pTaskInfo->window;
  iaInfo->order = TSDB_ORDER_ASC;
  iaInfo->interval = *pInterval;
4549 4550
  iaInfo->execModel = pTaskInfo->execModel;
  iaInfo->primaryTsIndex = primaryTsSlotId;
4551 4552 4553 4554

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

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

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

4561
  iaInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, numOfCols, iaInfo);
4562 4563
  if (iaInfo->timeWindowInterpo) {
    iaInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition));
4564 4565
  }

4566
  if (code != TSDB_CODE_SUCCESS) {
4567 4568 4569
    goto _error;
  }

4570
  initResultRowInfo(&iaInfo->binfo.resultRowInfo);
4571

4572 4573
  pOperator->name = "TimeMergeAlignedIntervalAggOperator";
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL;
4574 4575
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
4576
  pOperator->exprSupp.pExprInfo = pExprInfo;
4577
  pOperator->pTaskInfo = pTaskInfo;
4578
  pOperator->exprSupp.numOfExprs = numOfCols;
4579
  pOperator->info = miaInfo;
4580

4581 4582
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doMergeAlignedIntervalAgg, NULL, NULL,
                                         destroyMergeAlignedIntervalOperatorInfo, NULL, NULL, NULL);
4583 4584 4585 4586 4587 4588 4589 4590 4591

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

  return pOperator;

_error:
4592
  destroyMergeAlignedIntervalOperatorInfo(miaInfo, numOfCols);
4593
  taosMemoryFreeClear(miaInfo);
4594 4595 4596 4597
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
4598 4599 4600 4601 4602

//=====================================================================================================================
// merge interval operator
typedef struct SMergeIntervalAggOperatorInfo {
  SIntervalAggOperatorInfo intervalAggOperatorInfo;
L
Liu Jicong 已提交
4603 4604 4605 4606 4607 4608
  SList*                   groupIntervals;
  SListIter                groupIntervalsIter;
  bool                     hasGroupId;
  uint64_t                 groupId;
  SSDataBlock*             prefetchedBlock;
  bool                     inputBlocksFinished;
4609 4610
} SMergeIntervalAggOperatorInfo;

S
slzhou 已提交
4611
typedef struct SGroupTimeWindow {
L
Liu Jicong 已提交
4612
  uint64_t    groupId;
S
slzhou 已提交
4613 4614 4615
  STimeWindow window;
} SGroupTimeWindow;

4616 4617
void destroyMergeIntervalOperatorInfo(void* param, int32_t numOfOutput) {
  SMergeIntervalAggOperatorInfo* miaInfo = (SMergeIntervalAggOperatorInfo*)param;
S
slzhou 已提交
4618
  tdListFree(miaInfo->groupIntervals);
4619
  destroyIntervalOperatorInfo(&miaInfo->intervalAggOperatorInfo, numOfOutput);
4620

D
dapan1121 已提交
4621
  taosMemoryFreeClear(param);
4622 4623
}

L
Liu Jicong 已提交
4624 4625
static int32_t finalizeWindowResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, STimeWindow* win,
                                    SSDataBlock* pResultBlock) {
4626 4627 4628 4629 4630 4631 4632 4633 4634 4635 4636 4637 4638 4639 4640 4641
  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;
}

4642 4643 4644 4645 4646 4647 4648 4649
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 已提交
4650 4651
  SGroupTimeWindow groupTimeWindow = {.groupId = tableGroupId, .window = *newWin};
  tdListAppend(miaInfo->groupIntervals, &groupTimeWindow);
4652

S
slzhou 已提交
4653 4654 4655 4656 4657
  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 已提交
4658
    if (prevGrpWin->groupId != tableGroupId) {
S
slzhou 已提交
4659 4660 4661
      continue;
    }
    STimeWindow* prevWin = &prevGrpWin->window;
H
Haojun Liao 已提交
4662
    if ((ascScan && newWin->skey > prevWin->ekey) || ((!ascScan) && newWin->skey < prevWin->ekey)) {
S
slzhou 已提交
4663 4664 4665
      finalizeWindowResult(pOperatorInfo, tableGroupId, prevWin, pResultBlock);
      tdListPopNode(miaInfo->groupIntervals, listNode);
    }
4666 4667 4668 4669 4670 4671 4672 4673 4674 4675 4676 4677 4678 4679 4680 4681 4682 4683 4684 4685 4686
  }

  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 已提交
4687 4688
  STimeWindow win =
      getActiveTimeWindow(iaInfo->aggSup.pResultBuf, pResultRowInfo, blockStartTs, &iaInfo->interval, iaInfo->order);
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 4787 4788 4789 4790 4791

  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 已提交
4792
        tdListInitIter(miaInfo->groupIntervals, &miaInfo->groupIntervalsIter, TD_LIST_FORWARD);
4793 4794 4795 4796 4797 4798 4799 4800 4801 4802 4803 4804 4805 4806 4807 4808 4809 4810 4811 4812 4813 4814
        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;
4815 4816 4817
  }

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

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

  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 已提交
4845
  miaInfo->groupIntervals = tdListNew(sizeof(SGroupTimeWindow));
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 4896 4897 4898 4899 4900
  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 已提交
4901
}