timewindowoperator.c 190.7 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
                                      SExecTaskInfo* pTaskInfo) {
  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);
76

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

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
  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) {
107
  int32_t forwardRows = 0;
108 109 110 111

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

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

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

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

5
54liuyao 已提交
141
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157
  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) {
158 159 160 161 162 163 164 165 166 167 168
      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;
        }
      }
169 170 171 172 173 174

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

      if (key < keyList[midPos]) {
        firstPos = midPos + 1;
175 176
      } else if (key > keyList[midPos]) {
        lastPos = midPos - 1;
177 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
      } 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 已提交
212 213
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
                                 __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order) {
214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
  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) {
235
        item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
236 237
      }
    } else {
238
      num = pDataBlockInfo->rows - startPos;
239
      if (item != NULL) {
240
        item->lastKey = pDataBlockInfo->window.ekey + step;
241 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
      }
    }
  }

  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 已提交
272
  tw->skey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, precision);
273 274 275 276

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

  tw->ekey -= 1;
}

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

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

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

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

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

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

308
#if 0
309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327
    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) {
328 329
#endif

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

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

X
Xiaoyu Wang 已提交
336 337 338 339 340 341
    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;
342
    }
X
Xiaoyu Wang 已提交
343 344 345

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

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

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

367
  TSKEY curTs = tsCols[pos];
368 369

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

  // 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) {
376
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP);
377 378 379
    return true;
  }

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

  return true;
}

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

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

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

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

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

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

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

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

  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 已提交
442 443 444 445
  if (!inSlidingWindow(pInterval, pNext, pDataBlockInfo) && order == TSDB_ORDER_ASC) {
    return -1;
  }

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

  // tumbling time window query, a special case of sliding time window query
  if (pInterval->sliding == pInterval->interval && prevPosition != -1) {
451
    startPos = prevPosition + 1;
452
  } else {
453
    if ((skey <= pDataBlockInfo->window.skey && ascQuery) || (skey >= pDataBlockInfo->window.ekey && !ascQuery)) {
454 455
      startPos = 0;
    } else {
456
      startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, skey, order);
457 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
    }
  }

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

500 501
static bool isResultRowInterpolated(SResultRow* pResult, SResultTsInterpType type) {
  ASSERT(pResult != NULL && (type == RESULT_ROW_START_INTERP || type == RESULT_ROW_END_INTERP));
502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517
  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;
  }
}

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

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

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

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

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

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

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

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

568 569 570 571 572 573 574
  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 已提交
575
    for (int32_t i = pBlock->info.rows - 1; i >= 0; --i) {
576 577 578 579 580 581 582 583 584 585 586 587 588 589
      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;
    }
590 591 592
  }
}

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

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

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

604
  SResultRow* pResult = NULL;
605

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

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

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

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

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

    ASSERT(!isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP));

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

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

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

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

5
54liuyao 已提交
653
void printDataBlock(SSDataBlock* pBlock, const char* flag) {
5
54liuyao 已提交
654 655
  if (!pBlock || pBlock->info.rows == 0) {
    qDebug("======printDataBlock: Block is Null or Empty");
5
54liuyao 已提交
656 657
    return;
  }
L
Liu Jicong 已提交
658
  char* pBuf = NULL;
5
54liuyao 已提交
659 660
  qDebug("%s", dumpBlockData(pBlock, flag, &pBuf));
  taosMemoryFree(pBuf);
5
54liuyao 已提交
661 662
}

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

X
Xiaoyu Wang 已提交
665 666 667
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 已提交
668 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

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

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

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

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

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

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

5
54liuyao 已提交
756 757 758 759 760 761 762 763 764 765 766
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++) {
767 768
    SWinRes* pW = taosArrayGet(pWins, i);
    removeResult(pUpdated, pW->ts);
5
54liuyao 已提交
769 770 771
  }
}

772
int64_t getWinReskey(void* data, int32_t index) {
773
  SArray*  res = (SArray*)data;
774 775 776 777 778 779 780 781 782
  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);
783 784
    int64_t     key = *(int64_t*)pResKey->key;
    int32_t     index = binarySearch(pDelWins, delSize, key, TSDB_ORDER_DESC, getWinReskey);
785 786 787 788 789 790
    if (index >= 0 && key == getWinReskey(pDelWins, index)) {
      taosArrayRemove(pDelWins, index);
    }
  }
}

5
54liuyao 已提交
791 792 793 794 795
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 已提交
796
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup) { return isOverdue(pWin->ekey, pSup); }
5
54liuyao 已提交
797

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

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

X
Xiaoyu Wang 已提交
805
  int32_t     startPos = 0;
806
  int32_t     numOfOutput = pSup->numOfExprs;
X
Xiaoyu Wang 已提交
807 808 809 810 811
  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;
812

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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 已提交
911
  SListNode*         pn = tdListGetTail(pResultRowInfo->openWindow);
912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936
  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;
937 938 939 940 941 942 943
}

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

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

948 949
  int32_t scanFlag = MAIN_SCAN;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    char* val = colDataGetData(pStateColInfoData, j);

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

1025 1026 1027 1028
      pInfo->hasKey = true;

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

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

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

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

  SStateWindowOperatorInfo* pInfo = pOperator->info;
1083

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

  SSDataBlock* pBlock = pInfo->binfo.pRes;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

H
Haojun Liao 已提交
1515 1516 1517 1518 1519
static void freeItem(void* param) {
  SGroupKeys *pKey = (SGroupKeys*) param;
  taosMemoryFree(pKey->pData);
}

1520
void destroyIntervalOperatorInfo(void* param, int32_t numOfOutput) {
1521
  SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)param;
1522
  cleanupBasicInfo(&pInfo->binfo);
1523
  cleanupAggSup(&pInfo->aggSup);
H
Haojun Liao 已提交
1524 1525 1526 1527 1528 1529 1530
  pInfo->pRecycledPages = taosArrayDestroy(pInfo->pRecycledPages);
  pInfo->pInterpCols = taosArrayDestroy(pInfo->pInterpCols);
  taosArrayDestroyEx(pInfo->pPrevValues, freeItem);

  pInfo->pPrevValues = NULL;
  pInfo->pDelWins = taosArrayDestroy(pInfo->pDelWins);
  pInfo->pDelRes = blockDataDestroy(pInfo->pDelRes);
1531

H
Haojun Liao 已提交
1532 1533
  cleanupGroupResInfo(&pInfo->groupResInfo);
  colDataDestroy(&pInfo->twAggSup.timeWindowData);
D
dapan1121 已提交
1534
  taosMemoryFreeClear(param);
1535 1536
}

5
54liuyao 已提交
1537
void destroyStreamFinalIntervalOperatorInfo(void* param, int32_t numOfOutput) {
X
Xiaoyu Wang 已提交
1538
  SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)param;
1539
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
1540
  cleanupAggSup(&pInfo->aggSup);
L
Liu Jicong 已提交
1541
  // it should be empty.
5
54liuyao 已提交
1542 1543 1544
  taosHashCleanup(pInfo->pPullDataMap);
  taosArrayDestroy(pInfo->pPullWins);
  blockDataDestroy(pInfo->pPullDataRes);
1545
  taosArrayDestroy(pInfo->pRecycledPages);
5
54liuyao 已提交
1546

1547 1548 1549 1550
  if (pInfo->pChildren) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, i);
1551
      destroyStreamFinalIntervalOperatorInfo(pChildOp->info, numOfOutput);
1552 1553 1554
      taosMemoryFreeClear(pChildOp);
    }
  }
1555
  nodesDestroyNode((SNode*)pInfo->pPhyNode);
1556

D
dapan1121 已提交
1557
  taosMemoryFreeClear(param);
5
54liuyao 已提交
1558 1559
}

1560
static bool allInvertible(SqlFunctionCtx* pFCtx, int32_t numOfCols) {
5
54liuyao 已提交
1561 1562 1563 1564 1565 1566 1567 1568
  for (int32_t i = 0; i < numOfCols; i++) {
    if (!fmIsInvertible(pFCtx[i].functionId)) {
      return false;
    }
  }
  return true;
}

1569
static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SIntervalAggOperatorInfo* pInfo) {
1570 1571
  // the primary timestamp column
  bool needed = false;
1572 1573
  pInfo->pInterpCols = taosArrayInit(4, sizeof(SColumn));
  pInfo->pPrevValues = taosArrayInit(4, sizeof(SGroupKeys));
1574

X
Xiaoyu Wang 已提交
1575
  {  // ts column
1576 1577
    SColumn c = {0};
    c.colId = 1;
1578
    c.slotId = pInfo->primaryTsIndex;
1579 1580
    c.type = TSDB_DATA_TYPE_TIMESTAMP;
    c.bytes = sizeof(int64_t);
1581
    taosArrayPush(pInfo->pInterpCols, &c);
1582 1583

    SGroupKeys key = {0};
X
Xiaoyu Wang 已提交
1584 1585 1586 1587
    key.bytes = c.bytes;
    key.type = c.type;
    key.isNull = true;  // to denote no value is assigned yet
    key.pData = taosMemoryCalloc(1, c.bytes);
1588
    taosArrayPush(pInfo->pPrevValues, &key);
1589 1590
  }

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

H
Haojun Liao 已提交
1594
    if (fmIsIntervalInterpoFunc(pCtx[i].functionId)) {
1595 1596 1597
      SFunctParam* pParam = &pExpr->base.pParam[0];

      SColumn c = *pParam->pCol;
1598
      taosArrayPush(pInfo->pInterpCols, &c);
1599 1600 1601
      needed = true;

      SGroupKeys key = {0};
X
Xiaoyu Wang 已提交
1602 1603
      key.bytes = c.bytes;
      key.type = c.type;
1604
      key.isNull = false;
X
Xiaoyu Wang 已提交
1605
      key.pData = taosMemoryCalloc(1, c.bytes);
1606
      taosArrayPush(pInfo->pPrevValues, &key);
1607 1608 1609 1610 1611 1612
    }
  }

  return needed;
}

1613
void increaseTs(SqlFunctionCtx* pCtx) {
1614
  if (pCtx[0].pExpr->pExpr->_function.pFunctNode->funcType == FUNCTION_TYPE_WSTART) {
1615 1616 1617 1618
    pCtx[0].increase = true;
  }
}

1619 1620 1621 1622 1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640
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;
}

5
54liuyao 已提交
1641 1642 1643 1644 1645 1646
void initIntervalDownStream(SOperatorInfo* downstream, uint8_t type) {
  ASSERT(downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
  SStreamScanInfo* pScanInfo = downstream->info;
  pScanInfo->sessionSup.parentType = type;
}

1647 1648
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
                                          SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
L
Liu Jicong 已提交
1649 1650
                                          STimeWindowAggSupp* pTwAggSupp, SIntervalPhysiNode* pPhyNode,
                                          SExecTaskInfo* pTaskInfo, bool isStream) {
1651
  SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
L
Liu Jicong 已提交
1652
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1653 1654 1655 1656
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

L
Liu Jicong 已提交
1657 1658 1659
  pInfo->win = pTaskInfo->window;
  pInfo->order = TSDB_ORDER_ASC;
  pInfo->interval = *pInterval;
L
Liu Jicong 已提交
1660
  pInfo->execModel = pTaskInfo->execModel;
L
Liu Jicong 已提交
1661
  pInfo->twAggSup = *pTwAggSupp;
5
54liuyao 已提交
1662
  pInfo->ignoreExpiredData = pPhyNode->window.igExpired;
1663
  pInfo->pCondition = pPhyNode->window.node.pConditions;
1664 1665 1666 1667

  if (pPhyNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pPhyNode->window.pExprs, NULL, &numOfScalar);
L
Liu Jicong 已提交
1668
    int32_t    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
1669 1670 1671 1672
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }
1673

1674 1675
  pInfo->primaryTsIndex = primaryTsSlotId;

1676 1677
  SExprSupp* pSup = &pOperator->exprSupp;

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

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

1684 1685
  if (isStream) {
    ASSERT(numOfCols > 0);
1686
    increaseTs(pSup->pCtx);
1687
  }
1688

1689
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);
1690

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

1694
  pInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, numOfCols, pInfo);
1695 1696
  if (pInfo->timeWindowInterpo) {
    pInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition));
H
Haojun Liao 已提交
1697 1698 1699
    if (pInfo->binfo.resultRowInfo.openWindow == NULL) {
      goto _error;
    }
1700
  }
1701 1702 1703
  pInfo->pRecycledPages = taosArrayInit(4, sizeof(int32_t));
  pInfo->pDelWins = taosArrayInit(4, sizeof(SWinRes));
  pInfo->delIndex = 0;
5
54liuyao 已提交
1704
  // pInfo->pDelRes = createPullDataBlock(); todo(liuyao) for delete
1705 1706
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
1707

1708
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
1709

X
Xiaoyu Wang 已提交
1710 1711 1712 1713
  pOperator->name = "TimeIntervalAggOperator";
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL;
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
1714
  pOperator->exprSupp.pExprInfo = pExprInfo;
X
Xiaoyu Wang 已提交
1715
  pOperator->pTaskInfo = pTaskInfo;
1716
  pOperator->exprSupp.numOfExprs = numOfCols;
X
Xiaoyu Wang 已提交
1717
  pOperator->info = pInfo;
1718 1719 1720 1721

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

5
54liuyao 已提交
1722 1723 1724 1725
  if (nodeType(pPhyNode) == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL) {
    initIntervalDownStream(downstream, QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL);
  }

1726 1727 1728 1729 1730 1731 1732
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

L
Liu Jicong 已提交
1733
_error:
1734 1735 1736 1737 1738 1739 1740 1741 1742
  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 已提交
1743
                                                STimeWindowAggSupp* pTwAggSupp, SExecTaskInfo* pTaskInfo) {
1744
  SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
L
Liu Jicong 已提交
1745
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1746 1747 1748 1749 1750 1751 1752 1753 1754 1755 1756 1757 1758 1759 1760
  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);
1761 1762
  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
  initBasicInfo(&pInfo->binfo, pResBlock);
1763 1764
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);

wmmhello's avatar
wmmhello 已提交
1765
  if (code != TSDB_CODE_SUCCESS) {
1766 1767 1768
    goto _error;
  }

1769
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
1770 1771

  pOperator->name = "StreamTimeIntervalAggOperator";
X
Xiaoyu Wang 已提交
1772
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL;
1773
  pOperator->blocking = true;
1774
  pOperator->status = OP_NOT_OPENED;
1775
  pOperator->exprSupp.pExprInfo = pExprInfo;
1776
  pOperator->pTaskInfo = pTaskInfo;
1777
  pOperator->exprSupp.numOfExprs = numOfCols;
1778 1779 1780 1781 1782 1783 1784 1785 1786 1787 1788 1789
  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 已提交
1790
_error:
1791 1792 1793 1794 1795 1796 1797 1798 1799 1800
  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;
1801
  SExprSupp*     pSup = &pOperator->exprSupp;
1802

1803
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
1804 1805

  bool    masterScan = true;
1806
  int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
1807 1808 1809 1810 1811 1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825 1826 1827 1828 1829 1830 1831 1832 1833 1834 1835 1836 1837
  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;
1838 1839
      int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx,
                                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1840 1841 1842 1843 1844 1845
      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);
1846
      doApplyFunctions(pTaskInfo, pSup->pCtx, &window, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1847 1848 1849 1850 1851 1852 1853 1854 1855 1856
                       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];
1857 1858
  int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1859 1860 1861 1862 1863
  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);
1864
  doApplyFunctions(pTaskInfo, pSup->pCtx, &pRowSup->win, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
1865 1866 1867
                   pRowSup->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
}

1868
static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
1869 1870 1871 1872 1873 1874
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SSessionAggOperatorInfo* pInfo = pOperator->info;
  SOptrBasicInfo*          pBInfo = &pInfo->binfo;
1875
  SExprSupp*               pSup = &pOperator->exprSupp;
1876 1877

  if (pOperator->status == OP_RES_TO_RETURN) {
1878
    while (1) {
1879 1880 1881 1882 1883 1884 1885 1886
      doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
      doFilter(pInfo->pCondition, pBInfo->pRes);

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

1888 1889 1890 1891 1892
      if (pBInfo->pRes->info.rows > 0) {
        break;
      }
    }
    pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1893
    return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1894 1895
  }

1896 1897 1898
  int64_t st = taosGetTimestampUs();
  int32_t order = TSDB_ORDER_ASC;

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

  while (1) {
1902
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1903 1904 1905 1906 1907
    if (pBlock == NULL) {
      break;
    }

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

1911 1912 1913
    doSessionWindowAggImpl(pOperator, pInfo, pBlock);
  }

1914 1915
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;

1916 1917 1918 1919
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
  closeAllResultRows(&pBInfo->resultRowInfo);

1920
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
1921
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
1922
  while (1) {
1923 1924
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
    doFilter(pInfo->pCondition, pBInfo->pRes);
1925

1926 1927 1928 1929 1930
    bool hasRemain = hasDataInGroupInfo(&pInfo->groupResInfo);
    if (!hasRemain) {
      doSetOperatorCompleted(pOperator);
      break;
    }
1931

1932 1933 1934 1935 1936
    if (pBInfo->pRes->info.rows > 0) {
      break;
    }
  }
  pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1937
  return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1938 1939
}

1940
static void doKeepPrevRows(STimeSliceOperatorInfo* pSliceInfo, const SSDataBlock* pBlock, int32_t rowIndex) {
H
Haojun Liao 已提交
1941
  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
1942 1943
  for (int32_t i = 0; i < numOfCols; ++i) {
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
H
Haojun Liao 已提交
1944 1945 1946 1947 1948 1949

    // 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;
1950
      char* val = colDataGetData(pColInfoData, rowIndex);
H
Haojun Liao 已提交
1951 1952 1953 1954 1955
      memcpy(pkey->pData, val, pkey->bytes);
    }
  }
}

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 2035 2036 2037 2038 2039 2040 2041 2042 2043 2044 2045 2046 2047 2048 2049 2050 2051 2052 2053 2054 2055
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;
  }

2056
  int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
2057 2058 2059 2060 2061 2062 2063 2064 2065 2066 2067 2068 2069 2070
  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 已提交
2071
static SSDataBlock* doTimeslice(SOperatorInfo* pOperator) {
2072 2073 2074 2075
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

2076 2077
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

2078
  STimeSliceOperatorInfo* pSliceInfo = pOperator->info;
2079 2080
  SSDataBlock*            pResBlock = pSliceInfo->pRes;
  SExprSupp*              pSup = &pOperator->exprSupp;
H
Haojun Liao 已提交
2081

2082 2083
  blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);

2084 2085 2086 2087 2088 2089 2090 2091
  //  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;
  //  }
2092

2093 2094
  int32_t        order = TSDB_ORDER_ASC;
  SInterval*     pInterval = &pSliceInfo->interval;
2095 2096
  SOperatorInfo* downstream = pOperator->pDownstream[0];

H
Haojun Liao 已提交
2097
  int32_t numOfRows = 0;
2098
  while (1) {
2099
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
2100 2101 2102 2103
    if (pBlock == NULL) {
      break;
    }

2104 2105 2106 2107 2108
    int32_t code = initPrevRowsKeeper(pSliceInfo, pBlock);
    if (code != TSDB_CODE_SUCCESS) {
      longjmp(pTaskInfo->env, code);
    }

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

2112
    SColumnInfoData* pTsCol = taosArrayGet(pBlock->pDataBlock, pSliceInfo->tsCol.slotId);
2113 2114
    for (int32_t i = 0; i < pBlock->info.rows; ++i) {
      int64_t ts = *(int64_t*)colDataGetData(pTsCol, i);
H
Haojun Liao 已提交
2115 2116

      if (ts == pSliceInfo->current) {
2117
        for (int32_t j = 0; j < pOperator->exprSupp.numOfExprs; ++j) {
2118
          SExprInfo* pExprInfo = &pOperator->exprSupp.pExprInfo[j];
2119 2120
          int32_t    dstSlot = pExprInfo->base.resSchema.slotId;
          int32_t    srcSlot = pExprInfo->base.pParam[0].pCol->slotId;
H
Haojun Liao 已提交
2121 2122

          SColumnInfoData* pSrc = taosArrayGet(pBlock->pDataBlock, srcSlot);
2123
          SColumnInfoData* pDst = taosArrayGet(pResBlock->pDataBlock, dstSlot);
H
Haojun Liao 已提交
2124 2125 2126 2127 2128

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

2129
        pResBlock->info.rows += 1;
2130
        doKeepPrevRows(pSliceInfo, pBlock, i);
H
Haojun Liao 已提交
2131

2132 2133
        pSliceInfo->current =
            taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
H
Haojun Liao 已提交
2134 2135 2136 2137
        if (pSliceInfo->current > pSliceInfo->win.ekey) {
          doSetOperatorCompleted(pOperator);
          break;
        }
2138 2139 2140 2141

        if (pResBlock->info.rows >= pResBlock->info.capacity) {
          break;
        }
H
Haojun Liao 已提交
2142
      } else if (ts < pSliceInfo->current) {
2143
        if (i < pBlock->info.rows - 1) {
2144
          int64_t nextTs = *(int64_t*)colDataGetData(pTsCol, i + 1);
H
Haojun Liao 已提交
2145
          if (nextTs > pSliceInfo->current) {
2146 2147 2148
            while (pSliceInfo->current < nextTs && pSliceInfo->current <= pSliceInfo->win.ekey) {
              genInterpolationResult(pSliceInfo, &pOperator->exprSupp, pBlock, i, pResBlock);
              pSliceInfo->current =
H
Haojun Liao 已提交
2149
                  taosTimeAdd(pSliceInfo->current, pInterval->interval, pInterval->intervalUnit, pInterval->precision);
2150
              if (pResBlock->info.rows >= pResBlock->info.capacity) {
H
Haojun Liao 已提交
2151 2152
                break;
              }
H
Haojun Liao 已提交
2153
            }
2154 2155 2156 2157

            if (pSliceInfo->current > pSliceInfo->win.ekey) {
              doSetOperatorCompleted(pOperator);
              break;
H
Haojun Liao 已提交
2158 2159
            }
          } else {
H
Haojun Liao 已提交
2160
            // ignore current row, and do nothing
H
Haojun Liao 已提交
2161 2162
          }
        } else {  // it is the last row of current block
2163 2164 2165 2166 2167 2168 2169
          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);
2170 2171 2172
          if (pResBlock->info.rows >= pResBlock->info.capacity) {
            break;
          }
2173 2174 2175 2176 2177
        }

        if (pSliceInfo->current > pSliceInfo->win.ekey) {
          doSetOperatorCompleted(pOperator);
          break;
H
Haojun Liao 已提交
2178 2179 2180
        }
      }
    }
2181 2182 2183 2184
  }

  // restore the value
  setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED);
H
Haojun Liao 已提交
2185
  if (pResBlock->info.rows == 0) {
2186 2187 2188
    pOperator->status = OP_EXEC_DONE;
  }

H
Haojun Liao 已提交
2189 2190 2191
  return pResBlock->info.rows == 0 ? NULL : pResBlock;
}

2192
SOperatorInfo* createTimeSliceOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo) {
2193 2194 2195 2196 2197 2198
  STimeSliceOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(STimeSliceOperatorInfo));
  SOperatorInfo*          pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pOperator == NULL || pInfo == NULL) {
    goto _error;
  }

2199
  SInterpFuncPhysiNode* pInterpPhyNode = (SInterpFuncPhysiNode*)pPhyNode;
2200
  SExprSupp*            pSup = &pOperator->exprSupp;
2201

2202
  int32_t    numOfExprs = 0;
2203
  SExprInfo* pExprInfo = createExprInfo(pInterpPhyNode->pFuncs, NULL, &numOfExprs);
2204
  int32_t    code = initExprSupp(pSup, pExprInfo, numOfExprs);
H
Haojun Liao 已提交
2205 2206 2207 2208
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

2209
  if (pInterpPhyNode->pExprs != NULL) {
2210
    int32_t    num = 0;
2211 2212 2213 2214 2215 2216 2217 2218 2219 2220
    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);
2221

2222 2223 2224
  pInfo->pFillColInfo = createFillColInfo(pExprInfo, numOfExprs, (SNodeListNode*)pInterpPhyNode->pFillValues);
  pInfo->pRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
  pInfo->win = pInterpPhyNode->timeRange;
2225
  pInfo->interval.interval = pInterpPhyNode->interval;
2226
  pInfo->current = pInfo->win.skey;
H
Haojun Liao 已提交
2227

2228
  pOperator->name = "TimeSliceOperator";
2229
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_INTERP_FUNC;
2230 2231 2232 2233
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;
  pOperator->pTaskInfo = pTaskInfo;
2234

2235 2236
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doTimeslice, NULL, NULL, destroyBasicOperatorInfo, NULL, NULL, NULL);
2237

H
Haojun Liao 已提交
2238
  code = appendDownstream(pOperator, &downstream, 1);
2239 2240
  return pOperator;

L
Liu Jicong 已提交
2241
_error:
2242 2243 2244 2245 2246 2247 2248
  taosMemoryFree(pInfo);
  taosMemoryFree(pOperator);
  pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
  return NULL;
}

SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols,
2249
                                             SSDataBlock* pResBlock, STimeWindowAggSupp* pTwAggSup, int32_t tsSlotId,
2250
                                             SColumn* pStateKeyCol, SNode* pCondition, SExecTaskInfo* pTaskInfo) {
2251 2252 2253 2254 2255 2256
  SStateWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStateWindowOperatorInfo));
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

2257 2258 2259 2260
  pInfo->stateCol = *pStateKeyCol;
  pInfo->stateKey.type = pInfo->stateCol.type;
  pInfo->stateKey.bytes = pInfo->stateCol.bytes;
  pInfo->stateKey.pData = taosMemoryCalloc(1, pInfo->stateCol.bytes);
2261
  pInfo->pCondition = pCondition;
2262 2263 2264 2265
  if (pInfo->stateKey.pData == NULL) {
    goto _error;
  }

2266 2267 2268
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;

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

2272
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
2273

L
Liu Jicong 已提交
2274
  pInfo->twAggSup = *pTwAggSup;
2275 2276
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

X
Xiaoyu Wang 已提交
2277 2278
  pInfo->tsSlotId = tsSlotId;
  pOperator->name = "StateWindowOperator";
2279
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE;
X
Xiaoyu Wang 已提交
2280 2281
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
2282 2283
  pOperator->exprSupp.pExprInfo = pExpr;
  pOperator->exprSupp.numOfExprs = numOfCols;
X
Xiaoyu Wang 已提交
2284 2285
  pOperator->pTaskInfo = pTaskInfo;
  pOperator->info = pInfo;
2286 2287 2288 2289 2290 2291 2292

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

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

L
Liu Jicong 已提交
2293
_error:
2294 2295 2296 2297 2298 2299
  pTaskInfo->code = TSDB_CODE_SUCCESS;
  return NULL;
}

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

D
dapan1121 已提交
2302
  taosMemoryFreeClear(param);
2303 2304 2305
}

SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
L
Liu Jicong 已提交
2306
                                            SSDataBlock* pResBlock, int64_t gap, int32_t tsSlotId,
2307 2308
                                            STimeWindowAggSupp* pTwAggSupp, SNode* pCondition,
                                            SExecTaskInfo* pTaskInfo) {
2309 2310 2311 2312 2313 2314
  SSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSessionAggOperatorInfo));
  SOperatorInfo*           pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

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

2318
  int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
2319 2320 2321 2322
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

2323 2324
  initBasicInfo(&pInfo->binfo, pResBlock);

2325
  pInfo->twAggSup = *pTwAggSupp;
2326
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
2327 2328
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

L
Liu Jicong 已提交
2329 2330 2331 2332 2333
  pInfo->tsSlotId = tsSlotId;
  pInfo->gap = gap;
  pInfo->binfo.pRes = pResBlock;
  pInfo->winSup.prevTs = INT64_MIN;
  pInfo->reptScan = false;
2334
  pInfo->pCondition = pCondition;
L
Liu Jicong 已提交
2335
  pOperator->name = "SessionWindowAggOperator";
2336
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION;
2337
  pOperator->blocking = true;
L
Liu Jicong 已提交
2338
  pOperator->status = OP_NOT_OPENED;
2339 2340
  pOperator->exprSupp.pExprInfo = pExprInfo;
  pOperator->exprSupp.numOfExprs = numOfCols;
L
Liu Jicong 已提交
2341
  pOperator->info = pInfo;
2342 2343 2344 2345 2346 2347 2348 2349

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

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

L
Liu Jicong 已提交
2350
_error:
2351 2352 2353 2354 2355 2356 2357 2358
  if (pInfo != NULL) {
    destroySWindowOperatorInfo(pInfo, numOfCols);
  }

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

5
54liuyao 已提交
2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377 2378
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);
      }
    }
  }
}

2379 2380 2381 2382 2383 2384 2385 2386
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;
}

2387 2388
static void rebuildIntervalWindow(SStreamFinalIntervalOperatorInfo* pInfo, SExprSupp* pSup, SArray* pWinArray,
                                  int32_t groupId, int32_t numOfOutput, SExecTaskInfo* pTaskInfo, SArray* pUpdated) {
5
54liuyao 已提交
2389
  int32_t size = taosArrayGetSize(pWinArray);
5
54liuyao 已提交
2390 2391 2392
  if (!pInfo->pChildren) {
    return;
  }
5
54liuyao 已提交
2393
  for (int32_t i = 0; i < size; i++) {
2394 2395 2396 2397 2398
    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 已提交
2399
    int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
2400
    bool    find = true;
5
54liuyao 已提交
2401 2402 2403
    for (int32_t j = 0; j < numOfChildren; j++) {
      SOperatorInfo*            pChildOp = taosArrayGetP(pInfo->pChildren, j);
      SIntervalAggOperatorInfo* pChInfo = pChildOp->info;
2404
      SExprSupp*                pChildSup = &pChildOp->exprSupp;
2405 2406 2407 2408
      if (!hasIntervalWindow(&pChInfo->aggSup, pWinRes->ts, pWinRes->groupId)) {
        continue;
      }
      find = true;
2409
      SResultRow* pChResult = NULL;
2410 2411 2412
      setTimeWindowOutputBuf(&pChInfo->binfo.resultRowInfo, &ParentWin, true, &pChResult, pWinRes->groupId,
                             pChildSup->pCtx, pChildSup->numOfExprs, pChildSup->rowEntryInfoOffset, &pChInfo->aggSup,
                             pTaskInfo);
2413
      compactFunctions(pSup->pCtx, pChildSup->pCtx, numOfOutput, pTaskInfo);
5
54liuyao 已提交
2414
    }
2415 2416 2417
    if (find && pUpdated) {
      saveResultRow(pCurResult, pWinRes->groupId, pUpdated);
    }
5
54liuyao 已提交
2418 2419 2420 2421 2422
  }
}

bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup) {
  SET_RES_WINDOW_KEY(pSup->keyBuf, &pWin->skey, sizeof(int64_t), groupId);
2423 2424
  SResultRowPosition* p1 = (SResultRowPosition*)taosHashGet(pSup->pResultRowHashTable, pSup->keyBuf,
                                                            GET_RES_WINDOW_KEY_LEN(sizeof(int64_t)));
5
54liuyao 已提交
2425 2426 2427
  return p1 == NULL;
}

L
Liu Jicong 已提交
2428 2429 2430 2431
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 已提交
2432 2433 2434 2435 2436 2437 2438 2439 2440 2441 2442 2443 2444 2445
  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 已提交
2446 2447 2448 2449 2450 2451
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 已提交
2452
static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBlock, uint64_t tableGroupId,
S
shenglian zhou 已提交
2453
                           SArray* pUpdated) {
5
54liuyao 已提交
2454
  SStreamFinalIntervalOperatorInfo* pInfo = (SStreamFinalIntervalOperatorInfo*)pOperatorInfo->info;
X
Xiaoyu Wang 已提交
2455 2456
  SResultRowInfo*                   pResultRowInfo = &(pInfo->binfo.resultRowInfo);
  SExecTaskInfo*                    pTaskInfo = pOperatorInfo->pTaskInfo;
2457 2458
  SExprSupp*                        pSup = &pOperatorInfo->exprSupp;
  int32_t                           numOfOutput = pSup->numOfExprs;
X
Xiaoyu Wang 已提交
2459 2460 2461 2462 2463
  int32_t                           step = 1;
  bool                              ascScan = true;
  TSKEY*                            tsCols = NULL;
  SResultRow*                       pResult = NULL;
  int32_t                           forwardRows = 0;
5
54liuyao 已提交
2464

5
54liuyao 已提交
2465 2466 2467
  ASSERT(pSDataBlock->pDataBlock != NULL);
  SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
  tsCols = (int64_t*)pColDataInfo->pData;
5
54liuyao 已提交
2468

X
Xiaoyu Wang 已提交
2469 2470
  int32_t     startPos = ascScan ? 0 : (pSDataBlock->info.rows - 1);
  TSKEY       ts = getStartTsKey(&pSDataBlock->info.window, tsCols);
5
54liuyao 已提交
2471 2472 2473 2474 2475 2476
  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 已提交
2477
  while (1) {
5
54liuyao 已提交
2478
    bool isClosed = isCloseWindow(&nextWin, &pInfo->twAggSup);
5
54liuyao 已提交
2479
    if ((pInfo->ignoreExpiredData && isClosed) || !inSlidingWindow(&pInfo->interval, &nextWin, &pSDataBlock->info)) {
5
54liuyao 已提交
2480 2481 2482 2483 2484 2485 2486
      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 已提交
2487 2488 2489 2490 2491
      bool    ignore = true;
      SWinRes winRes = {
          .ts = nextWin.skey,
          .groupId = tableGroupId,
      };
5
54liuyao 已提交
2492 2493 2494 2495 2496
      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);
5
54liuyao 已提交
2497 2498 2499
        int32_t size = taosArrayGetSize(pInfo->pChildren);
        addPullWindow(pInfo->pPullDataMap, &winRes, size);
        qDebug("===stream===prepare retrive %" PRId64 ", size:%d", winRes.ts, size);
5
54liuyao 已提交
2500 2501 2502
      } else {
        int32_t index = -1;
        SArray* chArray = NULL;
5
54liuyao 已提交
2503
        int32_t chId = 0;
5
54liuyao 已提交
2504
        if (chIds) {
L
Liu Jicong 已提交
2505
          chArray = *(void**)chIds;
5
54liuyao 已提交
2506
          chId = getChildIndex(pSDataBlock);
5
54liuyao 已提交
2507 2508
          index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
        }
5
54liuyao 已提交
2509 2510 2511 2512 2513 2514 2515 2516
        // if (index != -1 && pSDataBlock->info.type == STREAM_PULL_DATA) {
        //   qDebug("===stream===delete child id %d", chId);
        //   taosArrayRemove(chArray, index);
        //   if (taosArrayGetSize(chArray) == 0) {
        //     // pull data is over
        //     taosHashRemove(pInfo->pPullDataMap, &winRes, sizeof(SWinRes));
        //   }
        // }
L
Liu Jicong 已提交
2517
        if (index == -1 || pSDataBlock->info.type == STREAM_PULL_DATA) {
5
54liuyao 已提交
2518 2519 2520 2521 2522 2523 2524 2525 2526 2527 2528
          ignore = false;
        }
      }

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

2531 2532
    int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, true, &pResult, tableGroupId, pSup->pCtx,
                                          numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
5
54liuyao 已提交
2533 2534 2535
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
5
54liuyao 已提交
2536

5
54liuyao 已提交
2537 2538 2539
    if (IS_FINAL_OP(pInfo)) {
      forwardRows = 1;
    } else {
H
Hongze Cheng 已提交
2540 2541
      forwardRows = getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, nextWin.ekey, binarySearchForKey,
                                             NULL, TSDB_ORDER_ASC);
5
54liuyao 已提交
2542
    }
5
54liuyao 已提交
2543
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdated) {
5
54liuyao 已提交
2544
      saveResultRow(pResult, tableGroupId, pUpdated);
5
54liuyao 已提交
2545
    }
5
54liuyao 已提交
2546
    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true);
2547 2548
    doApplyFunctions(pTaskInfo, pSup->pCtx, &nextWin, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, tsCols,
                     pSDataBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
2549
    int32_t prevEndPos = (forwardRows - 1) * step + startPos;
2550
    ASSERT(pSDataBlock->info.window.skey > 0 && pSDataBlock->info.window.ekey > 0);
5
54liuyao 已提交
2551 2552 2553 2554 2555 2556 2557
    startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, pInfo->order);
    if (startPos < 0) {
      break;
    }
  }
}

5
54liuyao 已提交
2558 2559 2560 2561
static void clearStreamIntervalOperator(SStreamFinalIntervalOperatorInfo* pInfo) {
  taosHashClear(pInfo->aggSup.pResultRowHashTable);
  clearDiskbasedBuf(pInfo->aggSup.pResultBuf);
  cleanupResultRowInfo(&pInfo->binfo.resultRowInfo);
2562
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
5
54liuyao 已提交
2563 2564
}

5
54liuyao 已提交
2565 2566 2567 2568
static void clearSpecialDataBlock(SSDataBlock* pBlock) {
  if (pBlock->info.rows <= 0) {
    return;
  }
5
54liuyao 已提交
2569 2570 2571
  blockDataCleanup(pBlock);
}

2572
void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsColIndex) {
5
54liuyao 已提交
2573 2574
  // ASSERT(pDest->info.capacity >= pSource->info.rows);
  blockDataEnsureCapacity(pDest, pSource->info.rows);
5
54liuyao 已提交
2575
  clearSpecialDataBlock(pDest);
5
54liuyao 已提交
2576 2577
  SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, 0);
  SColumnInfoData* pSourceCol = taosArrayGet(pSource->pDataBlock, tsColIndex);
2578

5
54liuyao 已提交
2579
  // copy timestamp column
2580 2581
  colDataAssign(pDestCol, pSourceCol, pSource->info.rows, &pDest->info);
  for (int32_t i = 1; i < taosArrayGetSize(pDest->pDataBlock); i++) {
5
54liuyao 已提交
2582 2583 2584
    SColumnInfoData* pCol = taosArrayGet(pDest->pDataBlock, i);
    colDataAppendNNULL(pCol, 0, pSource->info.rows);
  }
2585

5
54liuyao 已提交
2586
  pDest->info.rows = pSource->info.rows;
2587 2588
  pDest->info.groupId = pSource->info.groupId;
  pDest->info.type = pSource->info.type;
5
54liuyao 已提交
2589 2590 2591
  blockDataUpdateTsWindow(pDest, 0);
}

5
54liuyao 已提交
2592 2593 2594 2595 2596 2597 2598 2599 2600 2601 2602 2603 2604 2605
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 已提交
2606
  blockDataEnsureCapacity(pBlock, size - (*pIndex));
5
54liuyao 已提交
2607 2608
  ASSERT(3 <= taosArrayGetSize(pBlock->pDataBlock));
  for (; (*pIndex) < size; (*pIndex)++) {
L
Liu Jicong 已提交
2609
    SPullWindowInfo* pWin = taosArrayGet(array, (*pIndex));
5
54liuyao 已提交
2610
    SColumnInfoData* pStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
5
54liuyao 已提交
2611 2612
    colDataAppend(pStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false);

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

5
54liuyao 已提交
2616
    SColumnInfoData* pGroupId = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
5
54liuyao 已提交
2617 2618 2619 2620 2621 2622 2623 2624 2625 2626
    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 已提交
2627
void processPullOver(SSDataBlock* pBlock, SHashObj* pMap) {
5
54liuyao 已提交
2628
  SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
2629
  TSKEY*           tsData = (TSKEY*)pStartCol->pData;
5
54liuyao 已提交
2630
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
L
Liu Jicong 已提交
2631 2632
  uint64_t*        groupIdData = (uint64_t*)pGroupCol->pData;
  int32_t          chId = getChildIndex(pBlock);
5
54liuyao 已提交
2633 2634
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    SWinRes winRes = {.ts = tsData[i], .groupId = groupIdData[i]};
L
Liu Jicong 已提交
2635
    void*   chIds = taosHashGet(pMap, &winRes, sizeof(SWinRes));
5
54liuyao 已提交
2636
    if (chIds) {
L
Liu Jicong 已提交
2637
      SArray* chArray = *(SArray**)chIds;
5
54liuyao 已提交
2638 2639
      int32_t index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
      if (index != -1) {
5
54liuyao 已提交
2640
        qDebug("===stream===window %" PRId64 " delete child id %d", winRes.ts, chId);
5
54liuyao 已提交
2641 2642 2643 2644 2645 2646 2647 2648 2649
        taosArrayRemove(chArray, index);
        if (taosArrayGetSize(chArray) == 0) {
          // pull data is over
          taosHashRemove(pMap, &winRes, sizeof(SWinRes));
        }
      }
    }
  }
}
5
54liuyao 已提交
2650

5
54liuyao 已提交
2651 2652
static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
  SStreamFinalIntervalOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
2653
  SOperatorInfo*                    downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
2654
  SArray*                           pUpdated = taosArrayInit(4, POINTER_BYTES);
5
54liuyao 已提交
2655
  TSKEY                             maxTs = INT64_MIN;
5
54liuyao 已提交
2656

2657 2658
  SExprSupp* pSup = &pOperator->exprSupp;

5
54liuyao 已提交
2659
  qDebug("interval status %d %s", pOperator->status, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
L
Liu Jicong 已提交
2660

5
54liuyao 已提交
2661 2662 2663 2664
  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 已提交
2665
    if (pInfo->binfo.pRes->info.rows == 0) {
5
54liuyao 已提交
2666
      pOperator->status = OP_EXEC_DONE;
5
54liuyao 已提交
2667 2668 2669
      if (!IS_FINAL_OP(pInfo)) {
        // semi interval operator clear disk buffer
        clearStreamIntervalOperator(pInfo);
2670 2671
      } else {
        freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2672
      }
5
54liuyao 已提交
2673 2674
      return NULL;
    }
5
54liuyao 已提交
2675
    printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2676 2677 2678 2679
    return pInfo->binfo.pRes;
  } else {
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
    if (pInfo->binfo.pRes->info.rows != 0) {
5
54liuyao 已提交
2680
      printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2681 2682 2683 2684 2685
      return pInfo->binfo.pRes;
    }
    if (pInfo->pUpdateRes->info.rows != 0 && pInfo->returnUpdate) {
      pInfo->returnUpdate = false;
      ASSERT(!IS_FINAL_OP(pInfo));
5
54liuyao 已提交
2686
      printDataBlock(pInfo->pUpdateRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2687 2688
      // process the rest of the data
      return pInfo->pUpdateRes;
5
54liuyao 已提交
2689
    }
5
54liuyao 已提交
2690 2691 2692 2693
    doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
    if (pInfo->pPullDataRes->info.rows != 0) {
      // process the rest of the data
      ASSERT(IS_FINAL_OP(pInfo));
5
54liuyao 已提交
2694
      printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2695 2696
      return pInfo->pPullDataRes;
    }
2697 2698 2699
    doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
    if (pInfo->pDelRes->info.rows != 0) {
      // process the rest of the data
5
54liuyao 已提交
2700
      printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
2701 2702
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
2703 2704 2705 2706 2707
  }

  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
5
54liuyao 已提交
2708
      clearSpecialDataBlock(pInfo->pUpdateRes);
2709
      removeDeleteResults(pUpdated, pInfo->pDelWins);
5
54liuyao 已提交
2710
      pOperator->status = OP_RES_TO_RETURN;
5
54liuyao 已提交
2711
      qDebug("%s return data", IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2712 2713
      break;
    }
5
54liuyao 已提交
2714
    printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval Final recv" : "interval Semi recv");
5
54liuyao 已提交
2715
    maxTs = TMAX(maxTs, pBlock->info.window.ekey);
2716

L
Liu Jicong 已提交
2717
    if (pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_PULL_DATA ||
L
Liu Jicong 已提交
2718
        pBlock->info.type == STREAM_INVALID) {
5
54liuyao 已提交
2719 2720
      pInfo->binfo.pRes->info.type = pBlock->info.type;
    } else if (pBlock->info.type == STREAM_CLEAR) {
2721
      SArray* pUpWins = taosArrayInit(8, sizeof(SWinRes));
2722
      doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock, pUpWins);
5
54liuyao 已提交
2723
      if (IS_FINAL_OP(pInfo)) {
L
Liu Jicong 已提交
2724 2725
        int32_t                           childIndex = getChildIndex(pBlock);
        SOperatorInfo*                    pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
2726
        SStreamFinalIntervalOperatorInfo* pChildInfo = pChildOp->info;
L
Liu Jicong 已提交
2727
        SExprSupp*                        pChildSup = &pChildOp->exprSupp;
2728

2729 2730 2731
        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 已提交
2732 2733
        taosArrayDestroy(pUpWins);
        continue;
2734
      }
5
54liuyao 已提交
2735 2736
      removeResults(pUpWins, pUpdated);
      copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
2737
      pInfo->returnUpdate = true;
2738
      taosArrayDestroy(pUpWins);
5
54liuyao 已提交
2739
      break;
2740 2741 2742 2743 2744 2745 2746 2747 2748
    } 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,
2749
                              pOperator->exprSupp.numOfExprs, pOperator->pTaskInfo, pUpdated);
2750 2751 2752 2753
        continue;
      }
      removeResults(pInfo->pDelWins, pUpdated);
      break;
5
54liuyao 已提交
2754
    } else if (pBlock->info.type == STREAM_GET_ALL && IS_FINAL_OP(pInfo)) {
5
54liuyao 已提交
2755 2756
      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdated);
      continue;
5
54liuyao 已提交
2757
    } else if (pBlock->info.type == STREAM_RETRIEVE && !IS_FINAL_OP(pInfo)) {
2758
      SArray* pUpWins = taosArrayInit(8, sizeof(SWinRes));
5
54liuyao 已提交
2759
      doClearWindows(&pInfo->aggSup, pSup, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock, pUpWins);
5
54liuyao 已提交
2760 2761 2762 2763 2764 2765
      removeResults(pUpWins, pUpdated);
      taosArrayDestroy(pUpWins);
      if (taosArrayGetSize(pUpdated) > 0) {
        break;
      }
      continue;
L
Liu Jicong 已提交
2766 2767
    } else if (pBlock->info.type == STREAM_PULL_OVER && IS_FINAL_OP(pInfo)) {
      processPullOver(pBlock, pInfo->pPullDataMap);
5
54liuyao 已提交
2768
      continue;
5
54liuyao 已提交
2769
    }
5
54liuyao 已提交
2770

5
54liuyao 已提交
2771 2772 2773 2774
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
2775
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->order, MAIN_SCAN, true);
5
54liuyao 已提交
2776
    doHashInterval(pOperator, pBlock, pBlock->info.groupId, pUpdated);
5
54liuyao 已提交
2777
    if (IS_FINAL_OP(pInfo)) {
S
shenglian zhou 已提交
2778
      int32_t chIndex = getChildIndex(pBlock);
5
54liuyao 已提交
2779 2780 2781 2782 2783 2784 2785
      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);
        }
2786 2787
        SStreamFinalIntervalOperatorInfo* pTmpInfo = pChildOp->info;
        pTmpInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
5
54liuyao 已提交
2788
        taosArrayPush(pInfo->pChildren, &pChildOp);
5
54liuyao 已提交
2789
        qDebug("===stream===add child, id:%d", chIndex);
5
54liuyao 已提交
2790
      }
S
shenglian zhou 已提交
2791
      SOperatorInfo*                    pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
5
54liuyao 已提交
2792
      SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info;
2793
      setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, pChInfo->order, MAIN_SCAN, true);
5
54liuyao 已提交
2794
      doHashInterval(pChildOp, pBlock, pBlock->info.groupId, NULL);
5
54liuyao 已提交
2795 2796 2797 2798

      if (needBreak(pInfo)) {
        break;
      }
5
54liuyao 已提交
2799 2800
    }
  }
S
shenglian zhou 已提交
2801

5
54liuyao 已提交
2802
  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
5
54liuyao 已提交
2803
  if (IS_FINAL_OP(pInfo)) {
2804 2805
    closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, pInfo->pPullDataMap,
                        pUpdated, pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2806
    closeChildIntervalWindow(pInfo->pChildren, pInfo->twAggSup.maxTs);
5
54liuyao 已提交
2807 2808
  }

2809
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
2810 2811 2812
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
5
54liuyao 已提交
2813
  if (pInfo->binfo.pRes->info.rows != 0) {
5
54liuyao 已提交
2814
    printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2815 2816 2817 2818 2819 2820
    return pInfo->binfo.pRes;
  }

  if (pInfo->pUpdateRes->info.rows != 0 && pInfo->returnUpdate) {
    pInfo->returnUpdate = false;
    ASSERT(!IS_FINAL_OP(pInfo));
5
54liuyao 已提交
2821
    printDataBlock(pInfo->pUpdateRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2822 2823 2824
    // process the rest of the data
    return pInfo->pUpdateRes;
  }
5
54liuyao 已提交
2825 2826 2827 2828 2829

  doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
  if (pInfo->pPullDataRes->info.rows != 0) {
    // process the rest of the data
    ASSERT(IS_FINAL_OP(pInfo));
5
54liuyao 已提交
2830
    printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
5
54liuyao 已提交
2831 2832
    return pInfo->pPullDataRes;
  }
2833 2834 2835 2836

  doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
  if (pInfo->pDelRes->info.rows != 0) {
    // process the rest of the data
5
54liuyao 已提交
2837
    printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
2838 2839
    return pInfo->pDelRes;
  }
5
54liuyao 已提交
2840 2841 2842 2843 2844 2845 2846 2847 2848 2849 2850 2851 2852 2853 2854 2855 2856 2857 2858 2859 2860 2861 2862 2863 2864 2865
  // 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 已提交
2866 2867
}

S
shenglian zhou 已提交
2868 2869 2870
SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                     SExecTaskInfo* pTaskInfo, int32_t numOfChild) {
  SIntervalPhysiNode*               pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
5
54liuyao 已提交
2871
  SStreamFinalIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamFinalIntervalOperatorInfo));
S
shenglian zhou 已提交
2872
  SOperatorInfo*                    pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
2873 2874 2875
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
2876

5
54liuyao 已提交
2877
  pInfo->order = TSDB_ORDER_ASC;
S
shenglian zhou 已提交
2878 2879 2880 2881 2882 2883 2884 2885
  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 已提交
2886 2887
      .calTrigger = pIntervalPhyNode->window.triggerType,
      .maxTs = INT64_MIN,
S
shenglian zhou 已提交
2888
  };
2889
  ASSERT(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY);
5
54liuyao 已提交
2890 2891 2892
  pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
2893 2894 2895 2896 2897 2898 2899 2900 2901
  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 已提交
2902 2903
  int32_t      numOfCols = 0;
  SExprInfo*   pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
5
54liuyao 已提交
2904
  SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
2905 2906 2907 2908

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

2909
  ASSERT(numOfCols > 0);
2910
  increaseTs(pOperator->exprSupp.pCtx);
5
54liuyao 已提交
2911 2912 2913 2914
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
2915
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
5
54liuyao 已提交
2916 2917
  pInfo->pChildren = NULL;
  if (numOfChild > 0) {
2918
    pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
5
54liuyao 已提交
2919 2920 2921
    for (int32_t i = 0; i < numOfChild; i++) {
      SOperatorInfo* pChildOp = createStreamFinalIntervalOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
      if (pChildOp) {
2922 2923
        SStreamFinalIntervalOperatorInfo* pChInfo = pChildOp->info;
        pChInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
5
54liuyao 已提交
2924 2925 2926 2927 2928 2929
        taosArrayPush(pInfo->pChildren, &pChildOp);
        continue;
      }
      goto _error;
    }
  }
S
shenglian zhou 已提交
2930
  pInfo->pUpdateRes = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
5
54liuyao 已提交
2931
  pInfo->pUpdateRes->info.type = STREAM_CLEAR;
5
54liuyao 已提交
2932
  blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
5
54liuyao 已提交
2933 2934
  pInfo->returnUpdate = false;

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

5
54liuyao 已提交
2937 2938 2939 2940
  if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL) {
    pInfo->isFinal = true;
    pOperator->name = "StreamFinalIntervalOperator";
  } else {
5
54liuyao 已提交
2941
    // semi interval operator does not catch result
5
54liuyao 已提交
2942 2943 2944 2945
    pInfo->isFinal = false;
    pOperator->name = "StreamSemiIntervalOperator";
  }

5
54liuyao 已提交
2946
  if (!IS_FINAL_OP(pInfo) || numOfChild == 0) {
5
54liuyao 已提交
2947 2948
    pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
  }
5
54liuyao 已提交
2949 2950 2951 2952 2953
  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 已提交
2954
  pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired;
5
54liuyao 已提交
2955
  // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete
2956 2957
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
2958 2959
  pInfo->delIndex = 0;
  pInfo->pDelWins = taosArrayInit(4, sizeof(SWinRes));
5
54liuyao 已提交
2960

5
54liuyao 已提交
2961
  pOperator->operatorType = pPhyNode->type;
5
54liuyao 已提交
2962 2963
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
2964
  pOperator->exprSupp.pExprInfo = pExprInfo;
5
54liuyao 已提交
2965
  pOperator->pTaskInfo = pTaskInfo;
2966
  pOperator->exprSupp.numOfExprs = numOfCols;
5
54liuyao 已提交
2967 2968
  pOperator->info = pInfo;

S
shenglian zhou 已提交
2969 2970 2971
  pOperator->fpSet =
      createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, NULL, destroyStreamFinalIntervalOperatorInfo,
                          aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
2972 2973 2974 2975 2976 2977 2978 2979 2980 2981 2982 2983 2984 2985

  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 已提交
2986
}
5
54liuyao 已提交
2987 2988 2989

void destroyStreamAggSupporter(SStreamAggSupporter* pSup) {
  taosMemoryFreeClear(pSup->pKeyBuf);
2990
  void** pIte = NULL;
2991
  while ((pIte = taosHashIterate(pSup->pResultRows, pIte)) != NULL) {
2992
    SArray* pWins = (SArray*)(*pIte);
2993 2994 2995
    taosArrayDestroy(pWins);
  }
  taosHashCleanup(pSup->pResultRows);
5
54liuyao 已提交
2996 2997 2998 2999 3000
  destroyDiskbasedBuf(pSup->pResultBuf);
}

void destroyStreamSessionAggOperatorInfo(void* param, int32_t numOfOutput) {
  SStreamSessionAggOperatorInfo* pInfo = (SStreamSessionAggOperatorInfo*)param;
3001
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
3002 3003
  destroyStreamAggSupporter(&pInfo->streamAggSup);
  cleanupGroupResInfo(&pInfo->groupResInfo);
3004 3005 3006
  if (pInfo->pChildren != NULL) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
X
Xiaoyu Wang 已提交
3007
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, i);
3008 3009 3010 3011 3012 3013
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
      destroyStreamSessionAggOperatorInfo(pChInfo, numOfOutput);
      taosMemoryFreeClear(pChild);
      taosMemoryFreeClear(pChInfo);
    }
  }
3014

D
dapan1121 已提交
3015
  taosMemoryFreeClear(param);
5
54liuyao 已提交
3016 3017
}

3018 3019
int32_t initBasicInfoEx(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfCols,
                        SSDataBlock* pResultBlock) {
3020 3021 3022 3023 3024
  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

3025
  initBasicInfo(pBasicInfo, pResultBlock);
3026

5
54liuyao 已提交
3027
  for (int32_t i = 0; i < numOfCols; ++i) {
3028
    pSup->pCtx[i].pBuf = NULL;
5
54liuyao 已提交
3029
  }
3030

3031
  ASSERT(numOfCols > 0);
3032
  increaseTs(pSup->pCtx);
5
54liuyao 已提交
3033 3034 3035 3036 3037 3038 3039 3040
  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;
  }
}
5
54liuyao 已提交
3041

X
Xiaoyu Wang 已提交
3042 3043
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int64_t gap, int64_t waterMark,
                    uint8_t type) {
5
54liuyao 已提交
3044
  ASSERT(downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN);
3045
  SStreamScanInfo* pScanInfo = downstream->info;
X
Xiaoyu Wang 已提交
3046
  pScanInfo->sessionSup = (SessionWindowSupporter){.pStreamAggSup = pAggSup, .gap = gap, .parentType = type};
5
54liuyao 已提交
3047
  pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, waterMark);
5
54liuyao 已提交
3048 3049
}

3050 3051
int32_t initSessionAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx,
                                int32_t numOfOutput) {
3052 3053 3054
  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SResultWindowInfo));
}

3055 3056 3057 3058 3059 3060 3061
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 已提交
3062
  SStreamSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamSessionAggOperatorInfo));
3063
  SOperatorInfo*                 pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
3064 3065 3066 3067 3068
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
3069 3070 3071 3072 3073 3074 3075 3076
  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;
    }
  }
3077
  SExprSupp* pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3078

3079
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
5
54liuyao 已提交
3080 3081 3082
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
3083

3084
  code = initSessionAggSupporter(&pInfo->streamAggSup, "StreamSessionAggOperatorInfo", pSup->pCtx, numOfCols);
5
54liuyao 已提交
3085 3086 3087
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
X
Xiaoyu Wang 已提交
3088

5
54liuyao 已提交
3089 3090 3091 3092
  pInfo->pDummyCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfCols, sizeof(SqlFunctionCtx));
  if (pInfo->pDummyCtx == NULL) {
    goto _error;
  }
3093
  initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols);
5
54liuyao 已提交
3094

3095 3096
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pSessionNode->window.watermark, .calTrigger = pSessionNode->window.triggerType, .maxTs = INT64_MIN};
H
Haojun Liao 已提交
3097 3098

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

5
54liuyao 已提交
3101 3102 3103 3104
  pInfo->primaryTsIndex = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
  if (pSessionNode->window.pTsEnd) {
    pInfo->endTsIndex = ((SColumnNode*)pSessionNode->window.pTsEnd)->slotId;
  }
3105
  pInfo->gap = pSessionNode->gap;
5
54liuyao 已提交
3106 3107 3108 3109 3110
  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 已提交
3111
  // pInfo->pDelRes = createPullDataBlock();
3112 3113
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
3114
  pInfo->pChildren = NULL;
5
54liuyao 已提交
3115 3116
  pInfo->isFinal = false;
  pInfo->pPhyNode = pPhyNode;
5
54liuyao 已提交
3117
  pInfo->ignoreExpiredData = pSessionNode->window.igExpired;
5
54liuyao 已提交
3118
  pInfo->returnDelete = false;
5
54liuyao 已提交
3119 3120

  pOperator->name = "StreamSessionWindowAggOperator";
3121
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION;
5
54liuyao 已提交
3122 3123
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
3124 3125
  pOperator->exprSupp.pExprInfo = pExprInfo;
  pOperator->exprSupp.numOfExprs = numOfCols;
5
54liuyao 已提交
3126
  pOperator->info = pInfo;
X
Xiaoyu Wang 已提交
3127 3128 3129
  pOperator->fpSet =
      createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, NULL, destroyStreamSessionAggOperatorInfo,
                          aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
3130
  pOperator->pTaskInfo = pTaskInfo;
5
54liuyao 已提交
3131 3132 3133 3134
  if (downstream) {
    initDownStream(downstream, &pInfo->streamAggSup, pInfo->gap, pInfo->twAggSup.waterMark, pOperator->operatorType);
    code = appendDownstream(pOperator, &downstream, 1);
  }
5
54liuyao 已提交
3135 3136 3137 3138 3139 3140 3141 3142 3143 3144 3145 3146 3147 3148
  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 已提交
3149
  SArray*            pWinInfos = (SArray*)data;
5
54liuyao 已提交
3150 3151 3152
  SResultWindowInfo* pWin = taosArrayGet(pWinInfos, index);
  return pWin->win.ekey;
}
5
54liuyao 已提交
3153 3154

bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap) {
5
54liuyao 已提交
3155
  if (ts + gap >= pWin->skey && ts - gap <= pWin->ekey) {
5
54liuyao 已提交
3156 3157 3158 3159 3160
    return true;
  }
  return false;
}

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

X
Xiaoyu Wang 已提交
3163 3164
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 已提交
3165 3166 3167 3168
  return taosArrayInsert(pWinInfos, index, &win);
}

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

3173
SArray* getWinInfos(SStreamAggSupporter* pAggSup, uint64_t groupId) {
3174
  void**  ite = taosHashGet(pAggSup->pResultRows, &groupId, sizeof(uint64_t));
3175 3176 3177
  SArray* pWinInfos = NULL;
  if (ite == NULL) {
    pWinInfos = taosArrayInit(1024, pAggSup->valueSize);
3178
    taosHashPut(pAggSup->pResultRows, &groupId, sizeof(uint64_t), &pWinInfos, sizeof(void*));
3179 3180 3181 3182 3183 3184
  } else {
    pWinInfos = *ite;
  }
  return pWinInfos;
}

5
54liuyao 已提交
3185 3186
// don't add new window
SResultWindowInfo* getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
3187
                                       int64_t gap, int32_t* pIndex) {
5
54liuyao 已提交
3188 3189 3190 3191 3192 3193 3194 3195
  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
3196
  int32_t            index = binarySearch(pWinInfos, size, startTs, TSDB_ORDER_DESC, getSessionWindowEndkey);
5
54liuyao 已提交
3197 3198 3199 3200 3201 3202 3203 3204 3205 3206 3207 3208 3209 3210 3211
  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)) {
3212
      *pIndex = index + 1;
5
54liuyao 已提交
3213 3214 3215 3216 3217 3218 3219
      return pWin;
    }
  }

  return NULL;
}

3220 3221
SResultWindowInfo* getSessionTimeWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
                                        int64_t gap, int32_t* pIndex) {
3222 3223 3224
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;

5
54liuyao 已提交
3225 3226
  int32_t size = taosArrayGetSize(pWinInfos);
  if (size == 0) {
5
54liuyao 已提交
3227
    *pIndex = 0;
5
54liuyao 已提交
3228
    return addNewSessionWindow(pWinInfos, startTs);
5
54liuyao 已提交
3229 3230
  }
  // find the first position which is smaller than the key
5
54liuyao 已提交
3231
  int32_t            index = binarySearch(pWinInfos, size, startTs, TSDB_ORDER_DESC, getSessionWindowEndkey);
5
54liuyao 已提交
3232 3233 3234
  SResultWindowInfo* pWin = NULL;
  if (index >= 0) {
    pWin = taosArrayGet(pWinInfos, index);
5
54liuyao 已提交
3235
    if (isInWindow(pWin, startTs, gap)) {
5
54liuyao 已提交
3236 3237 3238 3239 3240 3241 3242
      *pIndex = index;
      return pWin;
    }
  }

  if (index + 1 < size) {
    pWin = taosArrayGet(pWinInfos, index + 1);
5
54liuyao 已提交
3243
    if (isInWindow(pWin, startTs, gap)) {
5
54liuyao 已提交
3244 3245
      *pIndex = index + 1;
      return pWin;
5
54liuyao 已提交
3246 3247 3248
    } else if (endTs != INT64_MIN && isInWindow(pWin, endTs, gap)) {
      *pIndex = index;
      return pWin;
5
54liuyao 已提交
3249 3250 3251 3252 3253
    }
  }

  if (index == size - 1) {
    *pIndex = taosArrayGetSize(pWinInfos);
5
54liuyao 已提交
3254
    return addNewSessionWindow(pWinInfos, startTs);
5
54liuyao 已提交
3255
  }
5
54liuyao 已提交
3256
  *pIndex = index + 1;
5
54liuyao 已提交
3257
  return insertNewSessionWindow(pWinInfos, startTs, index + 1);
5
54liuyao 已提交
3258 3259
}

3260 3261
int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs, TSKEY* pEndTs, int32_t rows,
                                int32_t start, int64_t gap, SHashObj* pStDeleted) {
5
54liuyao 已提交
3262
  for (int32_t i = start; i < rows; ++i) {
3263
    if (!isInWindow(pWinInfo, pStartTs[i], gap) && (!pEndTs || !isInWindow(pWinInfo, pEndTs[i], gap))) {
5
54liuyao 已提交
3264 3265
      return i - start;
    }
5
54liuyao 已提交
3266
    if (pWinInfo->win.skey > pStartTs[i]) {
5
54liuyao 已提交
3267 3268 3269 3270
      if (pStDeleted && pWinInfo->isOutput) {
        taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY));
        pWinInfo->isOutput = false;
      }
5
54liuyao 已提交
3271 3272 3273 3274 3275
      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 已提交
3276 3277 3278 3279 3280
    }
  }
  return rows - start;
}

X
Xiaoyu Wang 已提交
3281
static int32_t setWindowOutputBuf(SResultWindowInfo* pWinInfo, SResultRow** pResult, SqlFunctionCtx* pCtx,
3282
                                  uint64_t groupId, int32_t numOfOutput, int32_t* rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3283
                                  SStreamAggSupporter* pAggSup, SExecTaskInfo* pTaskInfo) {
5
54liuyao 已提交
3284 3285
  assert(pWinInfo->win.skey <= pWinInfo->win.ekey);
  // too many time window in query
3286
  int32_t size = taosArrayGetSize(pAggSup->pCurWins);
5
54liuyao 已提交
3287 3288 3289
  if (size > MAX_INTERVAL_TIME_WINDOW) {
    longjmp(pTaskInfo->env, TSDB_CODE_QRY_TOO_MANY_TIMEWINDOW);
  }
X
Xiaoyu Wang 已提交
3290

5
54liuyao 已提交
3291
  if (pWinInfo->pos.pageId == -1) {
3292
    *pResult = getNewResultRow(pAggSup->pResultBuf, groupId, pAggSup->resultRowSize);
5
54liuyao 已提交
3293 3294 3295 3296 3297 3298 3299 3300 3301 3302 3303 3304 3305 3306 3307 3308 3309 3310
    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;
3311
  setResultRowInitCtx(*pResult, pCtx, numOfOutput, rowEntryInfoOffset);
5
54liuyao 已提交
3312 3313 3314
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
3315 3316 3317
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,
3318
                                  int32_t numOutput, SOperatorInfo* pOperator) {
3319
  SExprSupp*     pSup = &pOperator->exprSupp;
3320 3321
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

X
Xiaoyu Wang 已提交
3322 3323
  SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, tsColId);
  TSKEY*           tsCols = (int64_t*)pColDataInfo->pData;
3324 3325
  int32_t          code = setWindowOutputBuf(pCurWin, pResult, pSup->pCtx, pSDataBlock->info.groupId, numOutput,
                                             pSup->rowEntryInfoOffset, pAggSup, pTaskInfo);
5
54liuyao 已提交
3326 3327 3328
  if (code != TSDB_CODE_SUCCESS || (*pResult) == NULL) {
    return TSDB_CODE_QRY_OUT_OF_MEMORY;
  }
5
54liuyao 已提交
3329
  updateTimeWindowInfo(pTimeWindowData, &pCurWin->win, false);
3330
  doApplyFunctions(pTaskInfo, pSup->pCtx, &pCurWin->win, pTimeWindowData, startIndex, winRows, tsCols,
X
Xiaoyu Wang 已提交
3331
                   pSDataBlock->info.rows, numOutput, TSDB_ORDER_ASC);
5
54liuyao 已提交
3332 3333 3334
  SFilePage* bufPage = getBufPage(pAggSup->pResultBuf, pCurWin->pos.pageId);
  setBufPageDirty(bufPage, true);
  releaseBufPage(pAggSup->pResultBuf, bufPage);
5
54liuyao 已提交
3335 3336 3337
  return TSDB_CODE_SUCCESS;
}

X
Xiaoyu Wang 已提交
3338 3339
static int32_t doOneWindowAgg(SStreamSessionAggOperatorInfo* pInfo, SSDataBlock* pSDataBlock,
                              SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex, int32_t winRows,
3340
                              int32_t numOutput, SOperatorInfo* pOperator) {
X
Xiaoyu Wang 已提交
3341
  return doOneWindowAggImpl(pInfo->primaryTsIndex, &pInfo->binfo, &pInfo->streamAggSup, &pInfo->twAggSup.timeWindowData,
3342
                            pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pOperator);
5
54liuyao 已提交
3343 3344
}

X
Xiaoyu Wang 已提交
3345 3346
static int32_t doOneStateWindowAgg(SStreamStateAggOperatorInfo* pInfo, SSDataBlock* pSDataBlock,
                                   SResultWindowInfo* pCurWin, SResultRow** pResult, int32_t startIndex,
3347
                                   int32_t winRows, int32_t numOutput, SOperatorInfo* pOperator) {
X
Xiaoyu Wang 已提交
3348
  return doOneWindowAggImpl(pInfo->primaryTsIndex, &pInfo->binfo, &pInfo->streamAggSup, &pInfo->twAggSup.timeWindowData,
3349
                            pSDataBlock, pCurWin, pResult, startIndex, winRows, numOutput, pOperator);
5
54liuyao 已提交
3350 3351
}

5
54liuyao 已提交
3352 3353
int32_t getNumCompactWindow(SArray* pWinInfos, int32_t startIndex, int64_t gap) {
  SResultWindowInfo* pCurWin = taosArrayGet(pWinInfos, startIndex);
X
Xiaoyu Wang 已提交
3354
  int32_t            size = taosArrayGetSize(pWinInfos);
5
54liuyao 已提交
3355 3356 3357 3358 3359 3360 3361 3362 3363 3364 3365
  // 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 已提交
3366
void compactTimeWindow(SStreamSessionAggOperatorInfo* pInfo, int32_t startIndex, int32_t num, uint64_t groupId,
3367
                       int32_t numOfOutput, SHashObj* pStUpdated, SHashObj* pStDeleted, SOperatorInfo* pOperator) {
3368
  SExprSupp*     pSup = &pOperator->exprSupp;
3369 3370
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

3371
  SResultWindowInfo* pCurWin = taosArrayGet(pInfo->streamAggSup.pCurWins, startIndex);
X
Xiaoyu Wang 已提交
3372
  SResultRow*        pCurResult = NULL;
3373
  setWindowOutputBuf(pCurWin, &pCurResult, pSup->pCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3374
                     &pInfo->streamAggSup, pTaskInfo);
5
54liuyao 已提交
3375
  num += startIndex + 1;
3376
  ASSERT(num <= taosArrayGetSize(pInfo->streamAggSup.pCurWins));
5
54liuyao 已提交
3377 3378
  // Just look for the window behind StartIndex
  for (int32_t i = startIndex + 1; i < num; i++) {
3379
    SResultWindowInfo* pWinInfo = taosArrayGet(pInfo->streamAggSup.pCurWins, i);
X
Xiaoyu Wang 已提交
3380
    SResultRow*        pWinResult = NULL;
3381
    setWindowOutputBuf(pWinInfo, &pWinResult, pInfo->pDummyCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3382
                       &pInfo->streamAggSup, pTaskInfo);
5
54liuyao 已提交
3383
    pCurWin->win.ekey = TMAX(pCurWin->win.ekey, pWinInfo->win.ekey);
3384
    compactFunctions(pSup->pCtx, pInfo->pDummyCtx, numOfOutput, pTaskInfo);
5
54liuyao 已提交
3385 3386 3387 3388 3389
    taosHashRemove(pStUpdated, &pWinInfo->pos, sizeof(SResultRowPosition));
    if (pWinInfo->isOutput) {
      taosHashPut(pStDeleted, &pWinInfo->pos, sizeof(SResultRowPosition), &pWinInfo->win.skey, sizeof(TSKEY));
      pWinInfo->isOutput = false;
    }
3390
    taosArrayRemove(pInfo->streamAggSup.pCurWins, i);
5
54liuyao 已提交
3391 3392
    SFilePage* tmpPage = getBufPage(pInfo->streamAggSup.pResultBuf, pWinInfo->pos.pageId);
    releaseBufPage(pInfo->streamAggSup.pResultBuf, tmpPage);
5
54liuyao 已提交
3393
  }
5
54liuyao 已提交
3394 3395 3396 3397
  SFilePage* bufPage = getBufPage(pInfo->streamAggSup.pResultBuf, pCurWin->pos.pageId);
  ASSERT(num > 0);
  setBufPageDirty(bufPage, true);
  releaseBufPage(pInfo->streamAggSup.pResultBuf, bufPage);
5
54liuyao 已提交
3398 3399
}

3400 3401
static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SHashObj* pStUpdated,
                                   SHashObj* pStDeleted, bool hasEndTs) {
X
Xiaoyu Wang 已提交
3402
  SExecTaskInfo*                 pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
3403
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
3404
  bool                           masterScan = true;
3405
  int32_t                        numOfOutput = pOperator->exprSupp.numOfExprs;
5
54liuyao 已提交
3406
  uint64_t                       groupId = pSDataBlock->info.groupId;
X
Xiaoyu Wang 已提交
3407 3408 3409 3410 3411
  int64_t                        gap = pInfo->gap;
  int64_t                        code = TSDB_CODE_SUCCESS;

  int32_t     step = 1;
  bool        ascScan = true;
5
54liuyao 已提交
3412 3413
  TSKEY*      startTsCols = NULL;
  TSKEY*      endTsCols = NULL;
5
54liuyao 已提交
3414
  SResultRow* pResult = NULL;
X
Xiaoyu Wang 已提交
3415
  int32_t     winRows = 0;
5
54liuyao 已提交
3416

5
54liuyao 已提交
3417 3418 3419 3420 3421 3422
  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 已提交
3423
  } else {
5
54liuyao 已提交
3424
    pEndTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
3425
  }
5
54liuyao 已提交
3426
  endTsCols = (int64_t*)pEndTsCol->pData;
X
Xiaoyu Wang 已提交
3427

5
54liuyao 已提交
3428
  SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
X
Xiaoyu Wang 已提交
3429
  for (int32_t i = 0; i < pSDataBlock->info.rows;) {
5
54liuyao 已提交
3430
    if (pInfo->ignoreExpiredData && isOverdue(endTsCols[i], &pInfo->twAggSup)) {
5
54liuyao 已提交
3431 3432 3433
      i++;
      continue;
    }
X
Xiaoyu Wang 已提交
3434
    int32_t            winIndex = 0;
3435 3436 3437
    SResultWindowInfo* pCurWin = getSessionTimeWindow(pAggSup, startTsCols[i], endTsCols[i], groupId, gap, &winIndex);
    winRows =
        updateSessionWindowInfo(pCurWin, startTsCols, endTsCols, pSDataBlock->info.rows, i, pInfo->gap, pStDeleted);
3438
    code = doOneWindowAgg(pInfo, pSDataBlock, pCurWin, &pResult, i, winRows, numOfOutput, pOperator);
5
54liuyao 已提交
3439 3440 3441
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
      longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
    }
5
54liuyao 已提交
3442

3443
    int32_t winNum = getNumCompactWindow(pAggSup->pCurWins, winIndex, gap);
5
54liuyao 已提交
3444
    if (winNum > 0) {
3445
      compactTimeWindow(pInfo, winIndex, winNum, groupId, numOfOutput, pStUpdated, pStDeleted, pOperator);
5
54liuyao 已提交
3446
    }
5
54liuyao 已提交
3447
    pCurWin->isClosed = false;
5
54liuyao 已提交
3448
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pStUpdated) {
5
54liuyao 已提交
3449 3450
      SWinRes value = {.ts = pCurWin->win.skey, .groupId = groupId};
      code = taosHashPut(pStUpdated, &pCurWin->pos, sizeof(SResultRowPosition), &value, sizeof(SWinRes));
5
54liuyao 已提交
3451 3452 3453 3454
      if (code != TSDB_CODE_SUCCESS) {
        longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
      }
      pCurWin->isOutput = true;
5
54liuyao 已提交
3455 3456 3457 3458 3459
    }
    i += winRows;
  }
}

5
54liuyao 已提交
3460 3461 3462 3463 3464
void deleteWindow(SArray* pWinInfos, int32_t index) {
  ASSERT(index >= 0 && index < taosArrayGetSize(pWinInfos));
  taosArrayRemove(pWinInfos, index);
}

3465
static void doDeleteTimeWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, int64_t gap, SArray* result) {
5
54liuyao 已提交
3466
  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
3467
  TSKEY*           startDatas = (TSKEY*)pStartTsCol->pData;
5
54liuyao 已提交
3468
  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
3469
  TSKEY*           endDatas = (TSKEY*)pEndTsCol->pData;
5
54liuyao 已提交
3470
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
3471
  uint64_t*        gpDatas = (uint64_t*)pGroupCol->pData;
5
54liuyao 已提交
3472 3473
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
    int32_t winIndex = 0;
3474 3475
    while (1) {
      SResultWindowInfo* pCurWin = getCurSessionWindow(pAggSup, startDatas[i], endDatas[i], gpDatas[i], gap, &winIndex);
5
54liuyao 已提交
3476 3477 3478 3479 3480 3481 3482 3483 3484 3485 3486
      if (!pCurWin) {
        break;
      }
      deleteWindow(pAggSup->pCurWins, winIndex);
      if (result) {
        taosArrayPush(result, pCurWin);
      }
    }
  }
}

3487 3488
static void doClearSessionWindows(SStreamAggSupporter* pAggSup, SExprSupp* pSup, SSDataBlock* pBlock, int32_t tsIndex,
                                  int32_t numOfOutput, int64_t gap, SArray* result) {
5
54liuyao 已提交
3489
  SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, tsIndex);
X
Xiaoyu Wang 已提交
3490 3491
  TSKEY*           tsCols = (TSKEY*)pColDataInfo->pData;
  int32_t          step = 0;
5
54liuyao 已提交
3492
  for (int32_t i = 0; i < pBlock->info.rows; i += step) {
X
Xiaoyu Wang 已提交
3493
    int32_t            winIndex = 0;
3494
    SResultWindowInfo* pCurWin =
5
54liuyao 已提交
3495 3496
        getCurSessionWindow(pAggSup, tsCols[i], INT64_MIN, pBlock->info.groupId, gap, &winIndex);
    if (!pCurWin || pCurWin->pos.pageId == -1) {
5
54liuyao 已提交
3497
      // window has been closed.
5
54liuyao 已提交
3498
      step = 1;
5
54liuyao 已提交
3499 3500
      continue;
    }
5
54liuyao 已提交
3501 3502
    step = updateSessionWindowInfo(pCurWin, tsCols, NULL, pBlock->info.rows, i, gap, NULL);
    ASSERT(isInWindow(pCurWin, tsCols[i], gap));
3503
    doClearWindowImpl(&pCurWin->pos, pAggSup->pResultBuf, pSup, numOfOutput);
3504 3505 3506
    if (result) {
      taosArrayPush(result, pCurWin);
    }
5
54liuyao 已提交
3507 3508 3509
  }
}

5
54liuyao 已提交
3510
static int32_t copyUpdateResult(SHashObj* pStUpdated, SArray* pUpdated) {
X
Xiaoyu Wang 已提交
3511
  void*  pData = NULL;
5
54liuyao 已提交
3512
  size_t keyLen = 0;
X
Xiaoyu Wang 已提交
3513
  while ((pData = taosHashIterate(pStUpdated, pData)) != NULL) {
5
54liuyao 已提交
3514 3515 3516 3517 3518 3519
    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 已提交
3520
    pos->groupId = ((SWinRes*)pData)->groupId;
5
54liuyao 已提交
3521
    pos->pos = *(SResultRowPosition*)key;
5
54liuyao 已提交
3522
    *(int64_t*)pos->key = ((SWinRes*)pData)->ts;
5
54liuyao 已提交
3523 3524
    taosArrayPush(pUpdated, &pos);
  }
5
54liuyao 已提交
3525
  taosArraySort(pUpdated, resultrowComparAsc);
5
54liuyao 已提交
3526 3527 3528 3529 3530
  return TSDB_CODE_SUCCESS;
}

void doBuildDeleteDataBlock(SHashObj* pStDeleted, SSDataBlock* pBlock, void** Ite) {
  blockDataCleanup(pBlock);
3531 3532 3533 3534 3535
  int32_t size = taosHashGetSize(pStDeleted);
  if (size == 0) {
    return;
  }
  blockDataEnsureCapacity(pBlock, size);
5
54liuyao 已提交
3536
  size_t keyLen = 0;
X
Xiaoyu Wang 已提交
3537
  while (((*Ite) = taosHashIterate(pStDeleted, *Ite)) != NULL) {
5
54liuyao 已提交
3538
    SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
X
Xiaoyu Wang 已提交
3539
    colDataAppend(pColInfoData, pBlock->info.rows, *Ite, false);
3540
    for (int32_t i = 1; i < taosArrayGetSize(pBlock->pDataBlock); i++) {
5
54liuyao 已提交
3541
      pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
X
Xiaoyu Wang 已提交
3542
      colDataAppendNULL(pColInfoData, pBlock->info.rows);
5
54liuyao 已提交
3543 3544 3545 3546 3547 3548 3549 3550 3551 3552 3553
    }
    pBlock->info.rows += 1;
    if (pBlock->info.rows + 1 >= pBlock->info.capacity) {
      break;
    }
  }
  if ((*Ite) == NULL) {
    taosHashClear(pStDeleted);
  }
}

X
Xiaoyu Wang 已提交
3554
static void rebuildTimeWindow(SStreamSessionAggOperatorInfo* pInfo, SArray* pWinArray, int32_t groupId,
3555
                              int32_t numOfOutput, SOperatorInfo* pOperator) {
3556
  SExprSupp*     pSup = &pOperator->exprSupp;
3557 3558
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;

3559 3560
  int32_t size = taosArrayGetSize(pWinArray);
  ASSERT(pInfo->pChildren);
3561

3562 3563
  for (int32_t i = 0; i < size; i++) {
    SResultWindowInfo* pParentWin = taosArrayGet(pWinArray, i);
X
Xiaoyu Wang 已提交
3564
    SResultRow*        pCurResult = NULL;
3565
    setWindowOutputBuf(pParentWin, &pCurResult, pSup->pCtx, groupId, numOfOutput, pSup->rowEntryInfoOffset,
X
Xiaoyu Wang 已提交
3566
                       &pInfo->streamAggSup, pTaskInfo);
3567 3568
    int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
    for (int32_t j = 0; j < numOfChildren; j++) {
X
Xiaoyu Wang 已提交
3569
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, j);
3570
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
3571
      SArray*                        pChWins = getWinInfos(&pChInfo->streamAggSup, groupId);
X
Xiaoyu Wang 已提交
3572 3573
      int32_t                        chWinSize = taosArrayGetSize(pChWins);
      int32_t index = binarySearch(pChWins, chWinSize, pParentWin->win.skey, TSDB_ORDER_DESC, getSessionWindowEndkey);
5
54liuyao 已提交
3574
      if (index < 0) {
3575
        index = 0;
5
54liuyao 已提交
3576 3577
      }
      for (int32_t k = index; k < chWinSize; k++) {
5
54liuyao 已提交
3578 3579
        SResultWindowInfo* pChWin = taosArrayGet(pChWins, k);
        if (pParentWin->win.skey <= pChWin->win.skey && pChWin->win.ekey <= pParentWin->win.ekey) {
3580
          SResultRow* pChResult = NULL;
5
54liuyao 已提交
3581
          setWindowOutputBuf(pChWin, &pChResult, pChild->exprSupp.pCtx, groupId, numOfOutput,
3582 3583
                             pChild->exprSupp.rowEntryInfoOffset, &pChInfo->streamAggSup, pTaskInfo);
          compactFunctions(pSup->pCtx, pChild->exprSupp.pCtx, numOfOutput, pTaskInfo);
5
54liuyao 已提交
3584 3585
          SFilePage* bufPage = getBufPage(pChInfo->streamAggSup.pResultBuf, pChWin->pos.pageId);
          releaseBufPage(pChInfo->streamAggSup.pResultBuf, bufPage);
3586
          continue;
5
54liuyao 已提交
3587 3588
        } else if (!pChWin->isClosed) {
          break;
3589 3590 3591
        }
      }
    }
5
54liuyao 已提交
3592 3593 3594 3595
    SFilePage* bufPage = getBufPage(pInfo->streamAggSup.pResultBuf, pParentWin->pos.pageId);
    ASSERT(size > 0);
    setBufPageDirty(bufPage, true);
    releaseBufPage(pInfo->streamAggSup.pResultBuf, bufPage);
3596 3597 3598
  }
}

X
Xiaoyu Wang 已提交
3599
typedef SResultWindowInfo* (*__get_win_info_)(void*);
5
54liuyao 已提交
3600 3601
SResultWindowInfo* getResWinForSession(void* pData) { return (SResultWindowInfo*)pData; }
SResultWindowInfo* getResWinForState(void* pData) { return &((SStateWindowInfo*)pData)->winInfo; }
5
54liuyao 已提交
3602

3603 3604
int32_t closeSessionWindow(SHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SArray* pClosed, __get_win_info_ fn,
                           bool delete) {
5
54liuyao 已提交
3605
  // Todo(liuyao) save window to tdb
3606
  void** pIte = NULL;
5
54liuyao 已提交
3607
  size_t keyLen = 0;
3608
  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
5
54liuyao 已提交
3609
    uint64_t* pGroupId = taosHashGetKey(pIte, &keyLen);
3610 3611
    SArray*   pWins = (SArray*)(*pIte);
    int32_t   size = taosArrayGetSize(pWins);
3612 3613 3614
    for (int32_t i = 0; i < size; i++) {
      void*              pWin = taosArrayGet(pWins, i);
      SResultWindowInfo* pSeWin = fn(pWin);
5
54liuyao 已提交
3615
      if (isCloseWindow(&pSeWin->win, pTwSup)) {
3616 3617
        if (!pSeWin->isClosed) {
          pSeWin->isClosed = true;
5
54liuyao 已提交
3618
          if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE && pClosed) {
5
54liuyao 已提交
3619
            int32_t code = saveResult(pSeWin->win.skey, pSeWin->pos.pageId, pSeWin->pos.offset, *pGroupId, pClosed);
5
54liuyao 已提交
3620 3621 3622
            if (code != TSDB_CODE_SUCCESS) {
              return code;
            }
3623 3624
            pSeWin->isOutput = true;
          }
5
54liuyao 已提交
3625
          if (delete) {
5
54liuyao 已提交
3626
            deleteWindow(pWins, i);
5
54liuyao 已提交
3627 3628 3629
            i--;
            size = taosArrayGetSize(pWins);
          }
5
54liuyao 已提交
3630
        }
3631
        continue;
5
54liuyao 已提交
3632
      }
3633
      break;
5
54liuyao 已提交
3634 3635 3636 3637 3638
    }
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3639 3640 3641 3642 3643 3644 3645 3646 3647 3648
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);
  }
}

3649
int32_t getAllSessionWindow(SHashObj* pHashMap, SArray* pClosed, __get_win_info_ fn) {
3650
  void** pIte = NULL;
3651
  while ((pIte = taosHashIterate(pHashMap, pIte)) != NULL) {
3652
    SArray* pWins = (SArray*)(*pIte);
3653 3654 3655 3656 3657 3658 3659 3660
    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 已提交
3661 3662 3663 3664 3665
    }
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3666 3667 3668 3669 3670 3671 3672 3673
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 已提交
3674
static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
5
54liuyao 已提交
3675
  SExprSupp*                     pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3676
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
3677
  SOptrBasicInfo*                pBInfo = &pInfo->binfo;
5
54liuyao 已提交
3678 3679 3680 3681
  TSKEY                          maxTs = INT64_MIN;
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
5
54liuyao 已提交
3682 3683
    doBuildDeleteDataBlock(pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
    if (pInfo->pDelRes->info.rows > 0) {
L
Liu Jicong 已提交
3684
      printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3685 3686
      return pInfo->pDelRes;
    }
X
Xiaoyu Wang 已提交
3687
    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
3688
    if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) {
5
54liuyao 已提交
3689 3690
      doSetOperatorCompleted(pOperator);
    }
L
Liu Jicong 已提交
3691
    printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3692 3693 3694
    return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
  }

X
Xiaoyu Wang 已提交
3695 3696
  _hash_fn_t     hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  SHashObj*      pStUpdated = taosHashInit(64, hashFn, true, HASH_NO_LOCK);
5
54liuyao 已提交
3697
  SOperatorInfo* downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
3698
  SArray*        pUpdated = taosArrayInit(16, POINTER_BYTES);
5
54liuyao 已提交
3699 3700 3701 3702 3703
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
      break;
    }
5
54liuyao 已提交
3704
    printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "Final  Session Recv" : "Single Session Recv");
3705

5
54liuyao 已提交
3706
    if (pBlock->info.type == STREAM_CLEAR) {
X
Xiaoyu Wang 已提交
3707
      SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
3708 3709
      doClearSessionWindows(&pInfo->streamAggSup, &pOperator->exprSupp, pBlock, 0, pOperator->exprSupp.numOfExprs, 0,
                            pWins);
5
54liuyao 已提交
3710 3711
      if (IS_FINAL_OP(pInfo)) {
        int32_t                        childIndex = getChildIndex(pBlock);
X
Xiaoyu Wang 已提交
3712
        SOperatorInfo*                 pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
3713
        SStreamSessionAggOperatorInfo* pChildInfo = pChildOp->info;
3714
        doClearSessionWindows(&pChildInfo->streamAggSup, &pChildOp->exprSupp, pBlock, 0, pChildOp->exprSupp.numOfExprs,
5
54liuyao 已提交
3715
                              0, NULL);
3716
        rebuildTimeWindow(pInfo, pWins, pBlock->info.groupId, pOperator->exprSupp.numOfExprs, pOperator);
3717 3718
      }
      taosArrayDestroy(pWins);
5
54liuyao 已提交
3719
      continue;
5
54liuyao 已提交
3720 3721 3722
    } else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT) {
      SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
      // gap must be 0
3723
      doDeleteTimeWindows(&pInfo->streamAggSup, pBlock, 0, pWins);
5
54liuyao 已提交
3724 3725 3726 3727 3728
      if (IS_FINAL_OP(pInfo)) {
        int32_t                        childIndex = getChildIndex(pBlock);
        SOperatorInfo*                 pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
        SStreamSessionAggOperatorInfo* pChildInfo = pChildOp->info;
        // gap must be 0
3729
        doDeleteTimeWindows(&pChildInfo->streamAggSup, pBlock, 0, NULL);
5
54liuyao 已提交
3730 3731 3732 3733 3734
        rebuildTimeWindow(pInfo, pWins, pBlock->info.groupId, pOperator->exprSupp.numOfExprs, pOperator);
      }
      copyDeleteWindowInfo(pWins, pInfo->pStDeleted);
      taosArrayDestroy(pWins);
      continue;
3735
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
3736
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getResWinForSession);
5
54liuyao 已提交
3737
      continue;
5
54liuyao 已提交
3738
    }
5
54liuyao 已提交
3739

5
54liuyao 已提交
3740 3741 3742 3743
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
3744
    // the pDataBlock are always the same one, no need to call this again
3745
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
3746 3747 3748 3749 3750 3751
    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++) {
3752 3753
        SOperatorInfo* pChildOp =
            createStreamFinalSessionAggOperatorInfo(NULL, pInfo->pPhyNode, pOperator->pTaskInfo, 0);
5
54liuyao 已提交
3754 3755 3756 3757 3758
        if (!pChildOp) {
          longjmp(pOperator->pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
        }
        taosArrayPush(pInfo->pChildren, &pChildOp);
      }
3759
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
5
54liuyao 已提交
3760 3761
      setInputDataBlock(pChildOp, pChildOp->exprSupp.pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
      doStreamSessionAggImpl(pChildOp, pBlock, NULL, NULL, true);
3762
    }
5
54liuyao 已提交
3763
    maxTs = TMAX(maxTs, pBlock->info.window.ekey);
5
54liuyao 已提交
3764
  }
5
54liuyao 已提交
3765 3766

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

3770 3771
  closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated, getResWinForSession,
                     pInfo->ignoreExpiredData);
5
54liuyao 已提交
3772
  closeChildSessionWindow(pInfo->pChildren, pInfo->twAggSup.maxTs, pInfo->ignoreExpiredData);
5
54liuyao 已提交
3773 3774 3775
  copyUpdateResult(pStUpdated, pUpdated);
  taosHashCleanup(pStUpdated);

3776
  finalizeUpdatedResult(pSup->numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
3777 3778 3779 3780
  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 已提交
3781
    printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3782 3783 3784
    return pInfo->pDelRes;
  }
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
L
Liu Jicong 已提交
3785
  printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "Final  Session" : "Single Session");
5
54liuyao 已提交
3786 3787 3788 3789
  return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
}

static void clearStreamSessionOperator(SStreamSessionAggOperatorInfo* pInfo) {
3790
  void** pIte = NULL;
5
54liuyao 已提交
3791
  while ((pIte = taosHashIterate(pInfo->streamAggSup.pResultRows, pIte)) != NULL) {
3792
    SArray* pWins = (SArray*)(*pIte);
5
54liuyao 已提交
3793 3794 3795 3796 3797 3798 3799 3800 3801 3802 3803 3804 3805 3806 3807 3808 3809 3810 3811 3812 3813 3814 3815 3816 3817
    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;
3818

5
54liuyao 已提交
3819 3820 3821
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
5
54liuyao 已提交
3822 3823 3824 3825 3826 3827 3828 3829 3830
    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 已提交
3831
      printDataBlock(pInfo->pDelRes, "Semi  Session");
5
54liuyao 已提交
3832 3833
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
3834 3835

    if (pInfo->pUpdateRes->info.rows > 0) {
5
54liuyao 已提交
3836 3837
      // process the rest of the data
      pOperator->status = OP_OPENED;
5
54liuyao 已提交
3838
      printDataBlock(pInfo->pUpdateRes, "Semi  Session");
5
54liuyao 已提交
3839 3840
      return pInfo->pUpdateRes;
    }
5
54liuyao 已提交
3841 3842 3843 3844
    // semi interval operator clear disk buffer
    clearStreamSessionOperator(pInfo);
    pOperator->status = OP_EXEC_DONE;
    return NULL;
5
54liuyao 已提交
3845 3846 3847 3848 3849 3850 3851 3852 3853
  }

  _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 已提交
3854
      clearSpecialDataBlock(pInfo->pUpdateRes);
5
54liuyao 已提交
3855 3856 3857
      break;
    }

5
54liuyao 已提交
3858
    if (pBlock->info.type == STREAM_CLEAR) {
5
54liuyao 已提交
3859
      SArray* pWins = taosArrayInit(16, sizeof(SResultWindowInfo));
5
54liuyao 已提交
3860
      doClearSessionWindows(&pInfo->streamAggSup, pSup, pBlock, 0, pSup->numOfExprs, 0, pWins);
5
54liuyao 已提交
3861 3862 3863 3864
      removeSessionResults(pStUpdated, pWins);
      taosArrayDestroy(pWins);
      copyUpdateDataBlock(pInfo->pUpdateRes, pBlock, pInfo->primaryTsIndex);
      break;
5
54liuyao 已提交
3865 3866
    } else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT) {
      // gap must be 0
3867
      doDeleteTimeWindows(&pInfo->streamAggSup, pBlock, 0, NULL);
5
54liuyao 已提交
3868 3869 3870
      copyDataBlock(pInfo->pDelRes, pBlock);
      pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;
      break;
5
54liuyao 已提交
3871 3872 3873 3874 3875
    } else if (pBlock->info.type == STREAM_GET_ALL) {
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getResWinForSession);
      continue;
    }

5
54liuyao 已提交
3876 3877 3878 3879
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
5
54liuyao 已提交
3880 3881 3882 3883 3884 3885 3886 3887 3888 3889 3890 3891
    // 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 已提交
3892
  copyUpdateResult(pStUpdated, pUpdated);
5
54liuyao 已提交
3893
  taosHashCleanup(pStUpdated);
5
54liuyao 已提交
3894

3895 3896
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated,
                        pSup->rowEntryInfoOffset);
5
54liuyao 已提交
3897
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
5
54liuyao 已提交
3898
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
5
54liuyao 已提交
3899 3900 3901 3902 3903 3904 3905 3906 3907 3908

  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 已提交
3909
    printDataBlock(pInfo->pDelRes, "Semi  Session");
5
54liuyao 已提交
3910 3911
    return pInfo->pDelRes;
  }
5
54liuyao 已提交
3912 3913

  if (pInfo->pUpdateRes->info.rows > 0) {
5
54liuyao 已提交
3914 3915
    // process the rest of the data
    pOperator->status = OP_OPENED;
5
54liuyao 已提交
3916
    printDataBlock(pInfo->pUpdateRes, "Semi  Session");
5
54liuyao 已提交
3917 3918
    return pInfo->pUpdateRes;
  }
5
54liuyao 已提交
3919 3920 3921

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

3924 3925
SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                       SExecTaskInfo* pTaskInfo, int32_t numOfChild) {
3926 3927
  int32_t        code = TSDB_CODE_OUT_OF_MEMORY;
  SOperatorInfo* pOperator = createStreamSessionAggOperatorInfo(downstream, pPhyNode, pTaskInfo);
3928 3929 3930
  if (pOperator == NULL) {
    goto _error;
  }
3931
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
5
54liuyao 已提交
3932 3933 3934 3935 3936 3937 3938

  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 已提交
3939
    pInfo->pUpdateRes->info.type = STREAM_CLEAR;
5
54liuyao 已提交
3940 3941 3942
    blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
    pOperator->name = "StreamSessionSemiAggOperator";
    pOperator->fpSet =
3943 3944
        createOperatorFpSet(operatorDummyOpenFn, doStreamSessionSemiAgg, NULL, NULL,
                            destroyStreamSessionAggOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
5
54liuyao 已提交
3945 3946 3947 3948 3949
  }
  pOperator->operatorType = pPhyNode->type;
  if (numOfChild > 0) {
    pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
    for (int32_t i = 0; i < numOfChild; i++) {
3950
      SOperatorInfo* pChild = createStreamFinalSessionAggOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
5
54liuyao 已提交
3951 3952 3953 3954
      if (pChild == NULL) {
        goto _error;
      }
      taosArrayPush(pInfo->pChildren, &pChild);
3955 3956 3957 3958 3959 3960
    }
  }
  return pOperator;

_error:
  if (pInfo != NULL) {
3961
    destroyStreamSessionAggOperatorInfo(pInfo, pOperator->exprSupp.numOfExprs);
3962 3963 3964 3965 3966 3967 3968
  }

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

void destroyStreamStateOperatorInfo(void* param, int32_t numOfOutput) {
X
Xiaoyu Wang 已提交
3971
  SStreamStateAggOperatorInfo* pInfo = (SStreamStateAggOperatorInfo*)param;
3972
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
3973 3974 3975 3976 3977
  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 已提交
3978
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, i);
5
54liuyao 已提交
3979 3980 3981 3982 3983 3984
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
      destroyStreamSessionAggOperatorInfo(pChInfo, numOfOutput);
      taosMemoryFreeClear(pChild);
      taosMemoryFreeClear(pChInfo);
    }
  }
D
dapan1121 已提交
3985 3986

  taosMemoryFreeClear(param);
5
54liuyao 已提交
3987 3988 3989 3990 3991 3992 3993
}

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

X
Xiaoyu Wang 已提交
3994 3995 3996 3997 3998 3999 4000 4001 4002 4003 4004 4005
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 已提交
4006 4007 4008 4009 4010 4011 4012 4013
  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 已提交
4014 4015 4016 4017 4018 4019 4020 4021 4022 4023 4024 4025
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 已提交
4026 4027 4028 4029 4030 4031 4032 4033 4034 4035 4036 4037 4038 4039 4040 4041 4042 4043 4044
  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);
}

4045 4046 4047
SStateWindowInfo* getStateWindowByTs(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, int32_t* pIndex) {
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;
X
Xiaoyu Wang 已提交
4048 4049
  int32_t           size = taosArrayGetSize(pWinInfos);
  int32_t           index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
5
54liuyao 已提交
4050 4051 4052 4053 4054 4055 4056 4057 4058 4059 4060 4061 4062 4063 4064 4065 4066 4067 4068 4069
  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;
}

4070 4071
SStateWindowInfo* getStateWindow(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, char* pKeyData,
                                 SColumn* pCol, int32_t* pIndex) {
4072 4073
  SArray* pWinInfos = getWinInfos(pAggSup, groupId);
  pAggSup->pCurWins = pWinInfos;
5
54liuyao 已提交
4074 4075 4076 4077 4078
  int32_t size = taosArrayGetSize(pWinInfos);
  if (size == 0) {
    *pIndex = 0;
    return addNewStateWindow(pWinInfos, ts, pKeyData, pCol);
  }
X
Xiaoyu Wang 已提交
4079
  int32_t           index = binarySearch(pWinInfos, size, ts, TSDB_ORDER_DESC, getStateWinTsKey);
5
54liuyao 已提交
4080 4081 4082 4083 4084 4085 4086 4087 4088 4089 4090 4091 4092 4093 4094 4095 4096 4097 4098 4099 4100 4101 4102 4103 4104 4105 4106 4107 4108 4109 4110 4111 4112
  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 已提交
4113 4114
int32_t updateStateWindowInfo(SArray* pWinInfos, int32_t winIndex, TSKEY* pTs, SColumnInfoData* pKeyCol, int32_t rows,
                              int32_t start, bool* allEqual, SHashObj* pSeDelete) {
5
54liuyao 已提交
4115 4116 4117 4118 4119
  *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 已提交
4120
      if (isEqualStateKey(pWinInfo, pKeyData)) {
5
54liuyao 已提交
4121 4122 4123 4124 4125 4126 4127 4128 4129 4130 4131 4132 4133 4134
        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 已提交
4135 4136
        taosHashPut(pSeDelete, &pWinInfo->winInfo.pos, sizeof(SResultRowPosition), &pWinInfo->winInfo.win.skey,
                    sizeof(TSKEY));
5
54liuyao 已提交
4137 4138 4139 4140 4141 4142 4143 4144 4145 4146 4147 4148
        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 已提交
4149 4150
static void doClearStateWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, int32_t tsIndex, SColumn* pCol,
                                int32_t keyIndex, SHashObj* pSeUpdated, SHashObj* pSeDeleted) {
5
54liuyao 已提交
4151 4152
  SColumnInfoData* pTsColInfo = taosArrayGet(pBlock->pDataBlock, tsIndex);
  SColumnInfoData* pKeyColInfo = taosArrayGet(pBlock->pDataBlock, keyIndex);
X
Xiaoyu Wang 已提交
4153 4154 4155
  TSKEY*           tsCol = (TSKEY*)pTsColInfo->pData;
  bool             allEqual = false;
  int32_t          step = 1;
5
54liuyao 已提交
4156
  for (int32_t i = 0; i < pBlock->info.rows; i += step) {
X
Xiaoyu Wang 已提交
4157 4158
    char*             pKeyData = colDataGetData(pKeyColInfo, i);
    int32_t           winIndex = 0;
4159
    SStateWindowInfo* pCurWin = getStateWindowByTs(pAggSup, tsCol[i], pBlock->info.groupId, &winIndex);
5
54liuyao 已提交
4160 4161 4162
    if (!pCurWin) {
      continue;
    }
4163
    step = updateStateWindowInfo(pAggSup->pCurWins, winIndex, tsCol, pKeyColInfo, pBlock->info.rows, i, &allEqual,
X
Xiaoyu Wang 已提交
4164
                                 pSeDeleted);
5
54liuyao 已提交
4165 4166 4167
    ASSERT(isTsInWindow(pCurWin, tsCol[i]) || isEqualStateKey(pCurWin, pKeyData));
    taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
    taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
4168
    deleteWindow(pAggSup->pCurWins, winIndex);
5
54liuyao 已提交
4169 4170 4171
  }
}

X
Xiaoyu Wang 已提交
4172 4173 4174
static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SHashObj* pSeUpdated,
                                 SHashObj* pStDeleted) {
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
4175
  SStreamStateAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
4176
  bool                         masterScan = true;
4177
  int32_t                      numOfOutput = pOperator->exprSupp.numOfExprs;
X
Xiaoyu Wang 已提交
4178 4179 4180 4181 4182 4183 4184
  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 已提交
4185
  if (pSDataBlock->pDataBlock != NULL) {
X
Xiaoyu Wang 已提交
4186 4187
    SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
    tsCols = (int64_t*)pColDataInfo->pData;
5
54liuyao 已提交
4188
  } else {
X
Xiaoyu Wang 已提交
4189
    return;
5
54liuyao 已提交
4190
  }
X
Xiaoyu Wang 已提交
4191

5
54liuyao 已提交
4192
  SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
X
Xiaoyu Wang 已提交
4193 4194
  SColumnInfoData*     pKeyColInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->stateCol.slotId);
  for (int32_t i = 0; i < pSDataBlock->info.rows; i += winRows) {
5
54liuyao 已提交
4195
    if (pInfo->ignoreExpiredData && isOverdue(tsCols[i], &pInfo->twAggSup)) {
5
54liuyao 已提交
4196 4197 4198
      i++;
      continue;
    }
X
Xiaoyu Wang 已提交
4199 4200 4201
    char*             pKeyData = colDataGetData(pKeyColInfo, i);
    int32_t           winIndex = 0;
    bool              allEqual = true;
4202 4203 4204 4205
    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 已提交
4206 4207 4208
    if (!allEqual) {
      taosArrayPush(pAggSup->pScanWindow, &pCurWin->winInfo.win);
      taosHashRemove(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition));
4209
      deleteWindow(pAggSup->pCurWins, winIndex);
5
54liuyao 已提交
4210 4211
      continue;
    }
4212
    code = doOneStateWindowAgg(pInfo, pSDataBlock, &pCurWin->winInfo, &pResult, i, winRows, numOfOutput, pOperator);
5
54liuyao 已提交
4213 4214 4215 4216 4217
    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 已提交
4218
      SWinRes value = {.ts = pCurWin->winInfo.win.skey, .groupId = groupId};
4219
      code = taosHashPut(pSeUpdated, &pCurWin->winInfo.pos, sizeof(SResultRowPosition), &value, sizeof(SWinRes));
5
54liuyao 已提交
4220 4221 4222 4223 4224 4225 4226 4227 4228 4229 4230 4231 4232
      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;
  }

4233
  SExprSupp*                   pSup = &pOperator->exprSupp;
5
54liuyao 已提交
4234
  SStreamStateAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
4235
  SOptrBasicInfo*              pBInfo = &pInfo->binfo;
5
54liuyao 已提交
4236 4237 4238
  if (pOperator->status == OP_RES_TO_RETURN) {
    doBuildDeleteDataBlock(pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
    if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
4239
      printDataBlock(pInfo->pDelRes, "single state");
5
54liuyao 已提交
4240 4241
      return pInfo->pDelRes;
    }
X
Xiaoyu Wang 已提交
4242
    doBuildResultDatablock(pOperator, pBInfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
4243
    if (pBInfo->pRes->info.rows == 0 || !hasDataInGroupInfo(&pInfo->groupResInfo)) {
5
54liuyao 已提交
4244 4245
      doSetOperatorCompleted(pOperator);
    }
5
54liuyao 已提交
4246
    printDataBlock(pBInfo->pRes, "single state");
5
54liuyao 已提交
4247 4248 4249
    return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
  }

X
Xiaoyu Wang 已提交
4250 4251
  _hash_fn_t     hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  SHashObj*      pSeUpdated = taosHashInit(64, hashFn, true, HASH_NO_LOCK);
5
54liuyao 已提交
4252
  SOperatorInfo* downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
4253
  SArray*        pUpdated = taosArrayInit(16, POINTER_BYTES);
5
54liuyao 已提交
4254 4255 4256 4257 4258
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
      break;
    }
5
54liuyao 已提交
4259
    printDataBlock(pBlock, "single state recv");
4260

5
54liuyao 已提交
4261
    if (pBlock->info.type == STREAM_CLEAR) {
X
Xiaoyu Wang 已提交
4262 4263
      doClearStateWindows(&pInfo->streamAggSup, pBlock, pInfo->primaryTsIndex, &pInfo->stateCol, pInfo->stateCol.slotId,
                          pSeUpdated, pInfo->pSeDeleted);
5
54liuyao 已提交
4264
      continue;
4265 4266 4267 4268 4269 4270
    } 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;
4271
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
4272
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pUpdated, getResWinForState);
5
54liuyao 已提交
4273
      continue;
5
54liuyao 已提交
4274
    }
4275

5
54liuyao 已提交
4276 4277 4278 4279
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
4280
    // the pDataBlock are always the same one, no need to call this again
4281
    setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
4282 4283 4284 4285 4286
    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 已提交
4287

4288 4289
  closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pUpdated, getResWinForState,
                     pInfo->ignoreExpiredData);
5
54liuyao 已提交
4290
  closeChildSessionWindow(pInfo->pChildren, pInfo->twAggSup.maxTs, pInfo->ignoreExpiredData);
5
54liuyao 已提交
4291
  copyUpdateResult(pSeUpdated, pUpdated);
5
54liuyao 已提交
4292 4293
  taosHashCleanup(pSeUpdated);

4294 4295
  finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->streamAggSup.pResultBuf, pUpdated,
                        pSup->rowEntryInfoOffset);
5
54liuyao 已提交
4296 4297 4298 4299
  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 已提交
4300
    printDataBlock(pInfo->pDelRes, "single state");
5
54liuyao 已提交
4301 4302
    return pInfo->pDelRes;
  }
X
Xiaoyu Wang 已提交
4303
  doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->streamAggSup.pResultBuf);
5
54liuyao 已提交
4304
  printDataBlock(pBInfo->pRes, "single state");
5
54liuyao 已提交
4305 4306 4307
  return pBInfo->pRes->info.rows == 0 ? NULL : pBInfo->pRes;
}

4308 4309 4310 4311
int32_t initStateAggSupporter(SStreamAggSupporter* pSup, const char* pKey, SqlFunctionCtx* pCtx, int32_t numOfOutput) {
  return initStreamAggSupporter(pSup, pKey, pCtx, numOfOutput, sizeof(SStateWindowInfo));
}

X
Xiaoyu Wang 已提交
4312 4313 4314 4315 4316 4317
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;
4318
  int32_t                      code = TSDB_CODE_OUT_OF_MEMORY;
5
54liuyao 已提交
4319

X
Xiaoyu Wang 已提交
4320 4321
  SStreamStateAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamStateAggOperatorInfo));
  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
4322 4323 4324 4325
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

4326 4327
  SExprSupp* pSup = &pOperator->exprSupp;

X
Xiaoyu Wang 已提交
4328
  int32_t    numOfCols = 0;
5
54liuyao 已提交
4329 4330 4331 4332
  SExprInfo* pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &numOfCols);

  pInfo->stateCol = extractColumnFromColumnNode(pColNode);
  initResultSizeInfo(pOperator, 4096);
5
54liuyao 已提交
4333 4334 4335 4336 4337 4338 4339 4340 4341
  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;
    }
  }

4342
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
X
Xiaoyu Wang 已提交
4343 4344
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pStateNode->window.watermark,
5
54liuyao 已提交
4345 4346
      .calTrigger = pStateNode->window.triggerType,
      .maxTs = INT64_MIN,
X
Xiaoyu Wang 已提交
4347
  };
5
54liuyao 已提交
4348
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
4349

4350
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
5
54liuyao 已提交
4351 4352 4353 4354
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

4355
  code = initStateAggSupporter(&pInfo->streamAggSup, "StreamStateAggOperatorInfo", pSup->pCtx, numOfCols);
5
54liuyao 已提交
4356 4357 4358 4359 4360 4361 4362 4363 4364
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

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

4365
  initDummyFunction(pInfo->pDummyCtx, pSup->pCtx, numOfCols);
5
54liuyao 已提交
4366 4367 4368 4369 4370
  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 已提交
4371
  // pInfo->pDelRes = createPullDataBlock(); // todo(liuyao) for delete
4372 4373
  pInfo->pDelRes = createOneDataBlock(pInfo->binfo.pRes, false);  // todo(liuyao) for delete
  pInfo->pDelRes->info.type = STREAM_DELETE_RESULT;               // todo(liuyao) for delete
5
54liuyao 已提交
4374
  pInfo->pChildren = NULL;
5
54liuyao 已提交
4375
  pInfo->ignoreExpiredData = pStateNode->window.igExpired;
5
54liuyao 已提交
4376 4377

  pOperator->name = "StreamStateAggOperator";
4378
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE;
5
54liuyao 已提交
4379 4380
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
4381 4382
  pOperator->exprSupp.numOfExprs = numOfCols;
  pOperator->exprSupp.pExprInfo = pExprInfo;
5
54liuyao 已提交
4383 4384
  pOperator->pTaskInfo = pTaskInfo;
  pOperator->info = pInfo;
X
Xiaoyu Wang 已提交
4385 4386 4387
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, NULL,
                                         destroyStreamStateOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
  initDownStream(downstream, &pInfo->streamAggSup, 0, pInfo->twAggSup.waterMark, pOperator->operatorType);
5
54liuyao 已提交
4388 4389 4390 4391 4392 4393 4394 4395 4396 4397 4398 4399 4400
  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;
}
4401

4402 4403
void destroyMergeAlignedIntervalOperatorInfo(void* param, int32_t numOfOutput) {
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = (SMergeAlignedIntervalAggOperatorInfo*)param;
D
dapan1121 已提交
4404 4405
  destroyIntervalOperatorInfo(miaInfo->intervalAggOperatorInfo, numOfOutput);

D
dapan1121 已提交
4406
  taosMemoryFreeClear(param);
4407 4408
}

4409 4410
static int32_t outputMergeAlignedIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId,
                                                SSDataBlock* pResultBlock, TSKEY wstartTs) {
4411
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
D
dapan1121 已提交
4412
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4413
  SExecTaskInfo*                        pTaskInfo = pOperatorInfo->pTaskInfo;
4414 4415

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

4418 4419 4420 4421
  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);
4422

4423 4424
  finalizeResultRowIntoResultDataBlock(iaInfo->aggSup.pResultBuf, p1, pSup->pCtx, pSup->pExprInfo, pSup->numOfExprs,
                                       pSup->rowEntryInfoOffset, pResultBlock, pTaskInfo);
4425
  taosHashRemove(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf, GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
4426

4427
  return 0;
4428 4429
}

4430 4431
static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo,
                                          SSDataBlock* pBlock, int32_t scanFlag, SSDataBlock* pResultBlock) {
4432
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
D
dapan1121 已提交
4433
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4434 4435

  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
4436
  SExprSupp*     pSup = &pOperatorInfo->exprSupp;
4437 4438

  int32_t     startPos = 0;
4439
  int32_t     numOfOutput = pSup->numOfExprs;
4440
  int64_t*    tsCols = extractTsCol(pBlock, iaInfo);
4441 4442 4443 4444
  uint64_t    tableGroupId = pBlock->info.groupId;
  TSKEY       blockStartTs = getStartTsKey(&pBlock->info.window, tsCols);
  SResultRow* pResult = NULL;

4445 4446
  STimeWindow win;
  win.skey = blockStartTs;
4447 4448
  win.ekey =
      taosTimeAdd(win.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1;
4449

4450
  // TODO: remove the hash table (groupid + winkey => result row position)
4451 4452
  int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
4453 4454 4455 4456
  if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
    longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
  }

4457 4458
  TSKEY       currTs = blockStartTs;
  TSKEY       currPos = startPos;
4459
  STimeWindow currWin = win;
4460
  while (1) {
4461 4462
    ++currPos;
    if (currPos >= pBlock->info.rows) {
4463 4464
      break;
    }
4465 4466 4467 4468
    if (tsCols[currPos] == currTs) {
      continue;
    } else {
      updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true);
4469 4470
      doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos,
                       tsCols, pBlock->info.rows, numOfOutput, iaInfo->order);
4471

4472
      outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs);
4473 4474 4475

      currTs = tsCols[currPos];
      currWin.skey = currTs;
4476
      currWin.ekey = taosTimeAdd(currWin.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit,
4477 4478
                                 iaInfo->interval.precision) -
                     1;
4479
      startPos = currPos;
4480 4481
      ret = setTimeWindowOutputBuf(pResultRowInfo, &currWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
                                   pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
4482 4483 4484
      if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
        longjmp(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
      }
4485 4486
    }
  }
4487
  updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true);
4488 4489
  doApplyFunctions(pTaskInfo, pSup->pCtx, &currWin, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos,
                   tsCols, pBlock->info.rows, numOfOutput, iaInfo->order);
4490

4491
  outputMergeAlignedIntervalResult(pOperatorInfo, tableGroupId, pResultBlock, currTs);
4492 4493
}

4494
static SSDataBlock* doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
S
shenglian zhou 已提交
4495
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
4496

4497
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperator->info;
D
dapan1121 已提交
4498
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4499 4500 4501 4502
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

4503
  SExprSupp*   pSup = &pOperator->exprSupp;
4504
  SSDataBlock* pRes = iaInfo->binfo.pRes;
4505
  blockDataCleanup(pRes);
4506
  blockDataEnsureCapacity(pRes, pOperator->resultInfo.capacity);
4507

4508 4509 4510 4511 4512 4513 4514 4515 4516 4517
  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;
4518
        miaInfo->prefetchedBlock = NULL;
4519
      }
4520

4521 4522 4523 4524
      if (pBlock == NULL) {
        miaInfo->inputBlocksFinished = true;
        break;
      }
4525

4526 4527 4528 4529 4530 4531 4532
      if (!miaInfo->hasGroupId) {
        miaInfo->hasGroupId = true;
        miaInfo->groupId = pBlock->info.groupId;
      } else if (miaInfo->groupId != pBlock->info.groupId) {
        miaInfo->prefetchedBlock = pBlock;
        break;
      }
4533

4534
      getTableScanInfo(pOperator, &iaInfo->order, &scanFlag);
4535
      setInputDataBlock(pOperator, pSup->pCtx, pBlock, iaInfo->order, scanFlag, true);
4536
      doMergeAlignedIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);
4537
      doFilter(miaInfo->pCondition, pRes);
4538
      if (pRes->info.rows >= pOperator->resultInfo.capacity) {
4539 4540 4541 4542 4543
        break;
      }
    }

    pRes->info.groupId = miaInfo->groupId;
4544
  }
4545
  miaInfo->hasGroupId = false;
4546

4547
  if (miaInfo->inputBlocksFinished) {
4548 4549 4550 4551 4552 4553 4554 4555
    doSetOperatorCompleted(pOperator);
  }

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

4556 4557
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo,
                                                      int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval,
4558 4559
                                                      int32_t primaryTsSlotId, SNode* pCondition,
                                                      SExecTaskInfo* pTaskInfo) {
4560
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = taosMemoryCalloc(1, sizeof(SMergeAlignedIntervalAggOperatorInfo));
4561
  SOperatorInfo*                        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
4562 4563 4564 4565
  if (miaInfo == NULL || pOperator == NULL) {
    goto _error;
  }

D
dapan1121 已提交
4566 4567 4568 4569 4570 4571
  miaInfo->intervalAggOperatorInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
  if (miaInfo->intervalAggOperatorInfo == NULL) {
    goto _error;
  }

  SIntervalAggOperatorInfo* iaInfo = miaInfo->intervalAggOperatorInfo;
4572
  SExprSupp*                pSup = &pOperator->exprSupp;
4573

4574
  miaInfo->pCondition = pCondition;
4575 4576 4577
  iaInfo->win = pTaskInfo->window;
  iaInfo->order = TSDB_ORDER_ASC;
  iaInfo->interval = *pInterval;
4578 4579
  iaInfo->execModel = pTaskInfo->execModel;
  iaInfo->primaryTsIndex = primaryTsSlotId;
4580 4581 4582 4583

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

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

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

4590
  iaInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, numOfCols, iaInfo);
4591 4592
  if (iaInfo->timeWindowInterpo) {
    iaInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SResultRowPosition));
4593 4594
  }

4595
  if (code != TSDB_CODE_SUCCESS) {
4596 4597 4598
    goto _error;
  }

4599
  initResultRowInfo(&iaInfo->binfo.resultRowInfo);
4600

4601 4602
  pOperator->name = "TimeMergeAlignedIntervalAggOperator";
  pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL;
4603 4604
  pOperator->blocking = false;
  pOperator->status = OP_NOT_OPENED;
4605
  pOperator->exprSupp.pExprInfo = pExprInfo;
4606
  pOperator->pTaskInfo = pTaskInfo;
4607
  pOperator->exprSupp.numOfExprs = numOfCols;
4608
  pOperator->info = miaInfo;
4609

4610 4611
  pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doMergeAlignedIntervalAgg, NULL, NULL,
                                         destroyMergeAlignedIntervalOperatorInfo, NULL, NULL, NULL);
4612 4613 4614 4615 4616 4617 4618 4619 4620

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

  return pOperator;

_error:
4621
  destroyMergeAlignedIntervalOperatorInfo(miaInfo, numOfCols);
4622
  taosMemoryFreeClear(miaInfo);
4623 4624 4625 4626
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
4627 4628 4629 4630 4631

//=====================================================================================================================
// merge interval operator
typedef struct SMergeIntervalAggOperatorInfo {
  SIntervalAggOperatorInfo intervalAggOperatorInfo;
L
Liu Jicong 已提交
4632 4633 4634 4635 4636 4637
  SList*                   groupIntervals;
  SListIter                groupIntervalsIter;
  bool                     hasGroupId;
  uint64_t                 groupId;
  SSDataBlock*             prefetchedBlock;
  bool                     inputBlocksFinished;
4638 4639
} SMergeIntervalAggOperatorInfo;

S
slzhou 已提交
4640
typedef struct SGroupTimeWindow {
L
Liu Jicong 已提交
4641
  uint64_t    groupId;
S
slzhou 已提交
4642 4643 4644
  STimeWindow window;
} SGroupTimeWindow;

4645 4646
void destroyMergeIntervalOperatorInfo(void* param, int32_t numOfOutput) {
  SMergeIntervalAggOperatorInfo* miaInfo = (SMergeIntervalAggOperatorInfo*)param;
S
slzhou 已提交
4647
  tdListFree(miaInfo->groupIntervals);
4648
  destroyIntervalOperatorInfo(&miaInfo->intervalAggOperatorInfo, numOfOutput);
4649

D
dapan1121 已提交
4650
  taosMemoryFreeClear(param);
4651 4652
}

L
Liu Jicong 已提交
4653 4654
static int32_t finalizeWindowResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, STimeWindow* win,
                                    SSDataBlock* pResultBlock) {
4655 4656 4657 4658 4659 4660 4661 4662 4663 4664 4665 4666 4667 4668 4669 4670
  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;
}

4671 4672 4673 4674 4675 4676 4677 4678
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 已提交
4679 4680
  SGroupTimeWindow groupTimeWindow = {.groupId = tableGroupId, .window = *newWin};
  tdListAppend(miaInfo->groupIntervals, &groupTimeWindow);
4681

S
slzhou 已提交
4682 4683 4684 4685 4686
  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 已提交
4687
    if (prevGrpWin->groupId != tableGroupId) {
S
slzhou 已提交
4688 4689 4690
      continue;
    }
    STimeWindow* prevWin = &prevGrpWin->window;
H
Haojun Liao 已提交
4691
    if ((ascScan && newWin->skey > prevWin->ekey) || ((!ascScan) && newWin->skey < prevWin->ekey)) {
S
slzhou 已提交
4692 4693 4694
      finalizeWindowResult(pOperatorInfo, tableGroupId, prevWin, pResultBlock);
      tdListPopNode(miaInfo->groupIntervals, listNode);
    }
4695 4696 4697 4698 4699 4700 4701 4702 4703 4704 4705 4706 4707 4708 4709 4710 4711 4712 4713 4714 4715
  }

  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 已提交
4716 4717
  STimeWindow win =
      getActiveTimeWindow(iaInfo->aggSup.pResultBuf, pResultRowInfo, blockStartTs, &iaInfo->interval, iaInfo->order);
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 4792 4793 4794 4795 4796 4797 4798 4799 4800 4801 4802 4803 4804 4805 4806 4807 4808 4809 4810 4811 4812 4813 4814 4815 4816 4817 4818 4819 4820

  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 已提交
4821
        tdListInitIter(miaInfo->groupIntervals, &miaInfo->groupIntervalsIter, TD_LIST_FORWARD);
4822 4823 4824 4825 4826 4827 4828 4829 4830 4831 4832 4833 4834 4835 4836 4837 4838 4839 4840 4841 4842 4843
        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;
4844 4845 4846
  }

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

S
slzhou 已提交
4849 4850 4851 4852
    if (listNode != NULL) {
      SGroupTimeWindow* grpWin = (SGroupTimeWindow*)(listNode->data);
      finalizeWindowResult(pOperator, grpWin->groupId, &grpWin->window, pRes);
      pRes->info.groupId = grpWin->groupId;
4853 4854 4855 4856 4857 4858 4859 4860 4861 4862 4863 4864 4865 4866 4867 4868 4869 4870 4871 4872 4873
    }
  }

  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 已提交
4874
  miaInfo->groupIntervals = tdListNew(sizeof(SGroupTimeWindow));
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 4901 4902 4903 4904 4905 4906 4907 4908 4909 4910 4911 4912 4913 4914 4915 4916 4917 4918 4919 4920 4921 4922 4923 4924 4925 4926 4927 4928 4929
  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 已提交
4930
}