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

L
Liu Jicong 已提交
25
#define IS_FINAL_OP(op)    ((op)->isFinal)
5
54liuyao 已提交
26
#define DEAULT_DELETE_MARK (1000LL * 60LL * 60LL * 24LL * 365LL * 10LL);
H
Haojun Liao 已提交
27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51

typedef struct SSessionAggOperatorInfo {
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
  bool               reptScan;  // next round scan
  int64_t            gap;       // session window gap
  int32_t            tsSlotId;  // primary timestamp slot id
  STimeWindowAggSupp twAggSup;
} SSessionAggOperatorInfo;

typedef struct SStateWindowOperatorInfo {
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SExprSupp          scalarSup;
  SGroupResInfo      groupResInfo;
  SWindowRowsSup     winSup;
  SColumn            stateCol;  // start row index
  bool               hasKey;
  SStateKeys         stateKey;
  int32_t            tsSlotId;  // primary timestamp column slot id
  STimeWindowAggSupp twAggSup;
} SStateWindowOperatorInfo;

52 53 54 55 56
typedef enum SResultTsInterpType {
  RESULT_ROW_START_INTERP = 1,
  RESULT_ROW_END_INTERP = 2,
} SResultTsInterpType;

5
54liuyao 已提交
57 58
typedef struct SPullWindowInfo {
  STimeWindow window;
L
Liu Jicong 已提交
59
  uint64_t    groupId;
5
54liuyao 已提交
60
  STimeWindow calWin;
5
54liuyao 已提交
61 62
} SPullWindowInfo;

63 64
typedef struct SOpenWindowInfo {
  SResultRowPosition pos;
L
Liu Jicong 已提交
65
  uint64_t           groupId;
66 67
} SOpenWindowInfo;

68 69
static int64_t* extractTsCol(SSDataBlock* pBlock, const SIntervalAggOperatorInfo* pInfo);

L
Liu Jicong 已提交
70 71
static SResultRowPosition addToOpenWindowList(SResultRowInfo* pResultRowInfo, const SResultRow* pResult,
                                              uint64_t groupId);
72 73
static void doCloseWindow(SResultRowInfo* pResultRowInfo, const SIntervalAggOperatorInfo* pInfo, SResultRow* pResult);

X
Xiaoyu Wang 已提交
74
static TSKEY getStartTsKey(STimeWindow* win, const TSKEY* tsCols) { return tsCols == NULL ? win->skey : tsCols[0]; }
75 76 77

static int32_t setTimeWindowOutputBuf(SResultRowInfo* pResultRowInfo, STimeWindow* win, bool masterscan,
                                      SResultRow** pResult, int64_t tableGroupId, SqlFunctionCtx* pCtx,
78
                                      int32_t numOfOutput, int32_t* rowEntryInfoOffset, SAggSupporter* pAggSup,
79 80
                                      SExecTaskInfo* pTaskInfo) {
  SResultRow* pResultRow = doSetResultOutBufByKey(pAggSup->pResultBuf, pResultRowInfo, (char*)&win->skey, TSDB_KEYSIZE,
D
dapan1121 已提交
81
                                                  masterscan, tableGroupId, pTaskInfo, true, pAggSup, true);
82 83 84 85 86 87 88 89

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

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

91
  *pResult = pResultRow;
92
  setResultRowInitCtx(pResultRow, pCtx, numOfOutput, rowEntryInfoOffset);
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
}

107
static void doKeepTuple(SWindowRowsSup* pRowSup, int64_t ts, uint64_t groupId) {
108 109 110
  pRowSup->win.ekey = ts;
  pRowSup->prevTs = ts;
  pRowSup->numOfRows += 1;
111
  pRowSup->groupId = groupId;
112 113
}

dengyihao's avatar
dengyihao 已提交
114 115
static void doKeepNewWindowStartInfo(SWindowRowsSup* pRowSup, const int64_t* tsList, int32_t rowIndex,
                                     uint64_t groupId) {
116 117 118
  pRowSup->startRowIndex = rowIndex;
  pRowSup->numOfRows = 0;
  pRowSup->win.skey = tsList[rowIndex];
119
  pRowSup->groupId = groupId;
120 121
}

L
Liu Jicong 已提交
122 123
FORCE_INLINE int32_t getForwardStepsInBlock(int32_t numOfRows, __block_search_fn_t searchFn, TSKEY ekey, int32_t pos,
                                            int32_t order, int64_t* pData) {
124
  int32_t forwardRows = 0;
125 126 127 128

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

S
slzhou 已提交
131
      while (pData[end + pos] == ekey) {
132
        forwardRows += 1;
S
slzhou 已提交
133
        ++pos;
134 135 136
      }
    }
  } else {
137
    int32_t end = searchFn((char*)&pData[pos], numOfRows - pos, ekey, order);
138
    if (end >= 0) {
139
      forwardRows = end;
140

S
slzhou 已提交
141
      while (pData[end + pos] == ekey) {
142
        forwardRows += 1;
S
slzhou 已提交
143
        ++pos;
144 145
      }
    }
X
Xiaoyu Wang 已提交
146 147 148 149 150 151 152 153
    //    int32_t end = searchFn((char*)pData, pos + 1, ekey, order);
    //    if (end >= 0) {
    //      forwardRows = pos - end;
    //
    //      if (pData[end] == ekey) {
    //        forwardRows += 1;
    //      }
    //    }
154 155
  }

156 157
  assert(forwardRows >= 0);
  return forwardRows;
158 159
}

5
54liuyao 已提交
160
int32_t binarySearchForKey(char* pValue, int num, TSKEY key, int order) {
161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176
  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) {
177 178 179 180 181 182 183 184 185 186 187
      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;
        }
      }
188 189 190 191 192 193

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

      if (key < keyList[midPos]) {
        firstPos = midPos + 1;
194 195
      } else if (key > keyList[midPos]) {
        lastPos = midPos - 1;
196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230
      } 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 已提交
231 232
int32_t getNumOfRowsInTimeWindow(SDataBlockInfo* pDataBlockInfo, TSKEY* pPrimaryColumn, int32_t startPos, TSKEY ekey,
                                 __block_search_fn_t searchFn, STableQueryInfo* item, int32_t order) {
233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253
  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) {
254
        item->lastKey = pPrimaryColumn[startPos + (num - 1)] + step;
255 256
      }
    } else {
257
      num = pDataBlockInfo->rows - startPos;
258
      if (item != NULL) {
259
        item->lastKey = pDataBlockInfo->window.ekey + step;
260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285
      }
    }
  }

  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;
286
  taosLocalTime(&t, &tm, NULL);
287 288 289 290

  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 已提交
291
  tw->skey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, precision);
292 293 294 295

  mon = (int)(mon + interval);
  tm.tm_year = mon / 12;
  tm.tm_mon = mon % 12;
wafwerar's avatar
wafwerar 已提交
296
  tw->ekey = convertTimePrecision((int64_t)taosMktime(&tm) * 1000LL, TSDB_TIME_PRECISION_MILLI, precision);
297 298 299 300

  tw->ekey -= 1;
}

5
54liuyao 已提交
301 302 303 304
void getNextIntervalWindow(SInterval* pInterval, STimeWindow* tw, int32_t order) {
  getNextTimeWindow(pInterval, pInterval->precision, order, tw);
}

305 306
void doTimeWindowInterpolation(SArray* pPrevValues, SArray* pDataBlock, TSKEY prevTs, int32_t prevRowIndex, TSKEY curTs,
                               int32_t curRowIndex, TSKEY windowKey, int32_t type, SExprSupp* pSup) {
307
  SqlFunctionCtx* pCtx = pSup->pCtx;
308

309
  int32_t index = 1;
310
  for (int32_t k = 0; k < pSup->numOfExprs; ++k) {
H
Haojun Liao 已提交
311
    if (!fmIsIntervalInterpoFunc(pCtx[k].functionId)) {
312 313 314 315
      pCtx[k].start.key = INT64_MIN;
      continue;
    }

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

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

321
    double v1 = 0, v2 = 0, v = 0;
322
    if (prevRowIndex == -1) {
323
      SGroupKeys* p = taosArrayGet(pPrevValues, index);
324
      GET_TYPED_DATA(v1, double, pColInfo->info.type, p->pData);
325
    } else {
326
      GET_TYPED_DATA(v1, double, pColInfo->info.type, colDataGetData(pColInfo, prevRowIndex));
327 328
    }

329
    GET_TYPED_DATA(v2, double, pColInfo->info.type, colDataGetData(pColInfo, curRowIndex));
330

331
#if 0
332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350
    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) {
351 352
#endif

X
Xiaoyu Wang 已提交
353 354 355
    SPoint point1 = (SPoint){.key = prevTs, .val = &v1};
    SPoint point2 = (SPoint){.key = curTs, .val = &v2};
    SPoint point = (SPoint){.key = windowKey, .val = &v};
356

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

X
Xiaoyu Wang 已提交
359 360 361 362 363 364
    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;
365
    }
X
Xiaoyu Wang 已提交
366 367 368

    index += 1;
  }
369
#if 0
370
  }
371
#endif
372 373 374 375 376 377 378 379 380 381 382 383 384 385
}

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

386 387
static bool setTimeWindowInterpolationStartTs(SIntervalAggOperatorInfo* pInfo, int32_t pos, SSDataBlock* pBlock,
                                              const TSKEY* tsCols, STimeWindow* win, SExprSupp* pSup) {
388
  bool ascQuery = (pInfo->inputOrder == TSDB_ORDER_ASC);
389

390
  TSKEY curTs = tsCols[pos];
391 392

  SGroupKeys* pTsKey = taosArrayGet(pInfo->pPrevValues, 0);
X
Xiaoyu Wang 已提交
393
  TSKEY       lastTs = *(int64_t*)pTsKey->pData;
394 395 396 397 398

  // 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) {
399
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP);
400 401 402
    return true;
  }

403 404
  // it is the first time window, no need to do interpolation
  if (pTsKey->isNull && pos == 0) {
405
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP);
406 407
  } else {
    TSKEY prevTs = ((pos == 0) ? lastTs : tsCols[pos - 1]);
408 409
    doTimeWindowInterpolation(pInfo->pPrevValues, pBlock->pDataBlock, prevTs, pos - 1, curTs, pos, key,
                              RESULT_ROW_START_INTERP, pSup);
410 411 412 413 414
  }

  return true;
}

415 416 417
static bool setTimeWindowInterpolationEndTs(SIntervalAggOperatorInfo* pInfo, SExprSupp* pSup, int32_t endRowIndex,
                                            SArray* pDataBlock, const TSKEY* tsCols, TSKEY blockEkey,
                                            STimeWindow* win) {
418
  int32_t order = pInfo->inputOrder;
419 420

  TSKEY actualEndKey = tsCols[endRowIndex];
421
  TSKEY key = (order == TSDB_ORDER_ASC) ? win->ekey : win->skey;
422 423

  // not ended in current data block, do not invoke interpolation
424
  if ((key > blockEkey && (order == TSDB_ORDER_ASC)) || (key < blockEkey && (order == TSDB_ORDER_DESC))) {
425
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP);
426 427 428
    return false;
  }

429
  // there is actual end point of current time window, no interpolation needs
430
  if (key == actualEndKey) {
431
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP);
432 433 434
    return true;
  }

435
  int32_t nextRowIndex = endRowIndex + 1;
436 437 438
  assert(nextRowIndex >= 0);

  TSKEY nextKey = tsCols[nextRowIndex];
439 440
  doTimeWindowInterpolation(pInfo->pPrevValues, pDataBlock, actualEndKey, endRowIndex, nextKey, nextRowIndex, key,
                            RESULT_ROW_END_INTERP, pSup);
441 442 443
  return true;
}

444 445
bool inCalSlidingWindow(SInterval* pInterval, STimeWindow* pWin, TSKEY calStart, TSKEY calEnd) {
  if (pInterval->interval != pInterval->sliding && (pWin->ekey < calStart || pWin->skey > calEnd)) {
5
54liuyao 已提交
446 447 448 449 450
    return false;
  }
  return true;
}

451 452 453 454
bool inSlidingWindow(SInterval* pInterval, STimeWindow* pWin, SDataBlockInfo* pBlockInfo) {
  return inCalSlidingWindow(pInterval, pWin, pBlockInfo->calWin.skey, pBlockInfo->calWin.ekey);
}

455
static int32_t getNextQualifiedWindow(SInterval* pInterval, STimeWindow* pNext, SDataBlockInfo* pDataBlockInfo,
5
54liuyao 已提交
456
                                      TSKEY* primaryKeys, int32_t prevPosition, int32_t order) {
X
Xiaoyu Wang 已提交
457
  bool ascQuery = (order == TSDB_ORDER_ASC);
458 459 460 461 462 463 464 465 466 467

  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 已提交
468 469 470 471
  if (!inSlidingWindow(pInterval, pNext, pDataBlockInfo) && order == TSDB_ORDER_ASC) {
    return -1;
  }

472
  TSKEY   skey = ascQuery ? pNext->skey : pNext->ekey;
473 474 475 476
  int32_t startPos = 0;

  // tumbling time window query, a special case of sliding time window query
  if (pInterval->sliding == pInterval->interval && prevPosition != -1) {
477
    startPos = prevPosition + 1;
478
  } else {
479
    if ((skey <= pDataBlockInfo->window.skey && ascQuery) || (skey >= pDataBlockInfo->window.ekey && !ascQuery)) {
480 481
      startPos = 0;
    } else {
482
      startPos = binarySearchForKey((char*)primaryKeys, pDataBlockInfo->rows, skey, order);
483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525
    }
  }

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

526
static bool isResultRowInterpolated(SResultRow* pResult, SResultTsInterpType type) {
527
  ASSERT(pResult != NULL && (type == RESULT_ROW_START_INTERP || type == RESULT_ROW_END_INTERP));
528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543
  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;
  }
}

544 545
static void doWindowBorderInterpolation(SIntervalAggOperatorInfo* pInfo, SSDataBlock* pBlock, SResultRow* pResult,
                                        STimeWindow* win, int32_t startPos, int32_t forwardRows, SExprSupp* pSup) {
546
  if (!pInfo->timeWindowInterpo) {
547 548 549
    return;
  }

550
  ASSERT(pBlock != NULL);
551 552 553 554 555
  if (pBlock->pDataBlock == NULL) {
    //    tscError("pBlock->pDataBlock == NULL");
    return;
  }

556
  SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
557 558

  TSKEY* tsCols = (TSKEY*)(pColInfo->pData);
559
  bool   done = isResultRowInterpolated(pResult, RESULT_ROW_START_INTERP);
560
  if (!done) {  // it is not interpolated, now start to generated the interpolated value
561
    bool interp = setTimeWindowInterpolationStartTs(pInfo, startPos, pBlock, tsCols, win, pSup);
562 563 564 565
    if (interp) {
      setResultRowInterpo(pResult, RESULT_ROW_START_INTERP);
    }
  } else {
566
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_START_INTERP);
567 568 569 570 571 572 573 574
  }

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

  // interpolation query does not generate the time window end interpolation
575
  done = isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP);
576
  if (!done) {
577
    int32_t endRowIndex = startPos + forwardRows - 1;
578

579
    TSKEY endKey = (pInfo->inputOrder == TSDB_ORDER_ASC) ? pBlock->info.window.ekey : pBlock->info.window.skey;
580
    bool  interp = setTimeWindowInterpolationEndTs(pInfo, pSup, endRowIndex, pBlock->pDataBlock, tsCols, endKey, win);
581 582 583 584
    if (interp) {
      setResultRowInterpo(pResult, RESULT_ROW_END_INTERP);
    }
  } else {
585
    setNotInterpoWindowKey(pSup->pCtx, pSup->numOfExprs, RESULT_ROW_END_INTERP);
586 587 588
  }
}

589 590
static void saveDataBlockLastRow(SArray* pPrevKeys, const SSDataBlock* pBlock, SArray* pCols) {
  if (pBlock->pDataBlock == NULL) {
591 592 593
    return;
  }

594 595 596 597 598 599 600
  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 已提交
601
    for (int32_t i = pBlock->info.rows - 1; i >= 0; --i) {
602 603 604 605 606 607 608
      if (colDataIsNull_s(pColInfo, i)) {
        continue;
      }

      char* val = colDataGetData(pColInfo, i);
      if (IS_VAR_DATA_TYPE(pkey->type)) {
        memcpy(pkey->pData, val, varDataTLen(val));
609
        ASSERT(varDataTLen(val) <= pkey->bytes);
610 611 612 613 614 615
      } else {
        memcpy(pkey->pData, val, pkey->bytes);
      }

      break;
    }
616 617 618
  }
}

619 620 621 622
static void doInterpUnclosedTimeWindow(SOperatorInfo* pOperatorInfo, int32_t numOfExprs, SResultRowInfo* pResultRowInfo,
                                       SSDataBlock* pBlock, int32_t scanFlag, int64_t* tsCols, SResultRowPosition* p) {
  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;

623
  SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
624
  SExprSupp*                pSup = &pOperatorInfo->exprSupp;
625

L
Liu Jicong 已提交
626 627
  int32_t startPos = 0;
  int32_t numOfOutput = pSup->numOfExprs;
628

629
  SResultRow* pResult = NULL;
630

631
  while (1) {
L
Liu Jicong 已提交
632 633 634
    SListNode*          pn = tdListGetHead(pResultRowInfo->openWindow);
    SOpenWindowInfo*    pOpenWin = (SOpenWindowInfo*)pn->data;
    uint64_t            groupId = pOpenWin->groupId;
635
    SResultRowPosition* p1 = &pOpenWin->pos;
636 637 638
    if (p->pageId == p1->pageId && p->offset == p1->offset) {
      break;
    }
639

640
    SResultRow* pr = getResultRowByPos(pInfo->aggSup.pResultBuf, p1, false);
641 642 643
    if (NULL == pr) {
      T_LONG_JMP(pTaskInfo->env, terrno);
    }
644

645
    ASSERT(pr->offset == p1->offset && pr->pageId == p1->pageId);
646

647
    if (pr->closed) {
648
      ASSERT(isResultRowInterpolated(pr, RESULT_ROW_START_INTERP) &&
X
Xiaoyu Wang 已提交
649
             isResultRowInterpolated(pr, RESULT_ROW_END_INTERP));
650 651
      SListNode* pNode = tdListPopHead(pResultRowInfo->openWindow);
      taosMemoryFree(pNode);
652 653
      continue;
    }
654

655
    STimeWindow w = pr->win;
656 657
    int32_t     ret = setTimeWindowOutputBuf(pResultRowInfo, &w, (scanFlag == MAIN_SCAN), &pResult, groupId, pSup->pCtx,
                                             numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
658
    if (ret != TSDB_CODE_SUCCESS) {
S
Shengliang Guan 已提交
659
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
660 661
    }

662
    ASSERT(!isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP));
663

X
Xiaoyu Wang 已提交
664 665
    SGroupKeys* pTsKey = taosArrayGet(pInfo->pPrevValues, 0);
    int64_t     prevTs = *(int64_t*)pTsKey->pData;
H
Haojun Liao 已提交
666
    if (groupId == pBlock->info.id.groupId) {
667 668 669
      doTimeWindowInterpolation(pInfo->pPrevValues, pBlock->pDataBlock, prevTs, -1, tsCols[startPos], startPos, w.ekey,
                                RESULT_ROW_END_INTERP, pSup);
    }
670 671

    setResultRowInterpo(pResult, RESULT_ROW_END_INTERP);
672
    setNotInterpoWindowKey(pSup->pCtx, numOfExprs, RESULT_ROW_START_INTERP);
673

674
    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &w, true);
L
Liu Jicong 已提交
675 676
    applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, 0,
                                    pBlock->info.rows, numOfExprs);
677 678 679

    if (isResultRowInterpolated(pResult, RESULT_ROW_END_INTERP)) {
      closeResultRow(pr);
680 681
      SListNode* pNode = tdListPopHead(pResultRowInfo->openWindow);
      taosMemoryFree(pNode);
X
Xiaoyu Wang 已提交
682
    } else {  // the remains are can not be closed yet.
683
      break;
684
    }
685
  }
686
}
687

5
54liuyao 已提交
688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743
typedef int32_t (*__compare_fn_t)(void* pKey, void* data, int32_t index);

int32_t binarySearchCom(void* keyList, int num, void* pKey, int order, __compare_fn_t comparefn) {
  int firstPos = 0, lastPos = num - 1, midPos = -1;
  int numOfRows = 0;

  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 (comparefn(pKey, keyList, lastPos) >= 0) return lastPos;
      if (comparefn(pKey, keyList, firstPos) == 0) return firstPos;
      if (comparefn(pKey, keyList, firstPos) < 0) return firstPos - 1;

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

      if (comparefn(pKey, keyList, midPos) < 0) {
        lastPos = midPos - 1;
      } else if (comparefn(pKey, keyList, midPos) > 0) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }

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

      if (comparefn(pKey, keyList, lastPos) > 0) {
        lastPos = lastPos + 1;
        if (lastPos >= num)
          return -1;
        else
          return lastPos;
      }

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

      if (comparefn(pKey, keyList, midPos) < 0) {
        lastPos = midPos - 1;
      } else if (comparefn(pKey, keyList, midPos) > 0) {
        firstPos = midPos + 1;
      } else {
        break;
      }
    }
  }

  return midPos;
}

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

X
Xiaoyu Wang 已提交
746 747 748
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 已提交
749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794

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

5
54liuyao 已提交
797 798 799
  return midPos;
}

5
54liuyao 已提交
800
int32_t comparePullWinKey(void* pKey, void* data, int32_t index) {
L
Liu Jicong 已提交
801
  SArray*          res = (SArray*)data;
5
54liuyao 已提交
802
  SPullWindowInfo* pos = taosArrayGet(res, index);
L
Liu Jicong 已提交
803
  SPullWindowInfo* pData = (SPullWindowInfo*)pKey;
5
54liuyao 已提交
804
  if (pData->groupId > pos->groupId) {
5
54liuyao 已提交
805
    return 1;
5
54liuyao 已提交
806 807 808 809 810 811 812 813
  } else if (pData->groupId < pos->groupId) {
    return -1;
  }

  if (pData->window.skey > pos->window.ekey) {
    return 1;
  } else if (pData->window.ekey < pos->window.skey) {
    return -1;
5
54liuyao 已提交
814
  }
5
54liuyao 已提交
815
  return 0;
5
54liuyao 已提交
816 817 818 819 820 821 822 823
}

static int32_t savePullWindow(SPullWindowInfo* pPullInfo, SArray* pPullWins) {
  int32_t size = taosArrayGetSize(pPullWins);
  int32_t index = binarySearchCom(pPullWins, size, pPullInfo, TSDB_ORDER_DESC, comparePullWinKey);
  if (index == -1) {
    index = 0;
  } else {
5
54liuyao 已提交
824 825
    int32_t code = comparePullWinKey(pPullInfo, pPullWins, index);
    if (code == 0) {
L
Liu Jicong 已提交
826
      SPullWindowInfo* pos = taosArrayGet(pPullWins, index);
5
54liuyao 已提交
827 828 829 830
      pos->window.skey = TMIN(pos->window.skey, pPullInfo->window.skey);
      pos->window.ekey = TMAX(pos->window.ekey, pPullInfo->window.ekey);
      pos->calWin.skey = TMIN(pos->calWin.skey, pPullInfo->calWin.skey);
      pos->calWin.ekey = TMAX(pos->calWin.ekey, pPullInfo->calWin.ekey);
5
54liuyao 已提交
831
      return TSDB_CODE_SUCCESS;
L
Liu Jicong 已提交
832
    } else if (code > 0) {
5
54liuyao 已提交
833
      index++;
5
54liuyao 已提交
834 835 836 837 838 839 840 841
    }
  }
  if (taosArrayInsert(pPullWins, index, pPullInfo) == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
842 843 844
static int32_t saveResult(SResultWindowInfo winInfo, SSHashObj* pStUpdated) {
  winInfo.sessionWin.win.ekey = winInfo.sessionWin.win.skey;
  return tSimpleHashPut(pStUpdated, &winInfo.sessionWin, sizeof(SSessionKey), &winInfo, sizeof(SResultWindowInfo));
5
54liuyao 已提交
845 846
}

847
static int32_t saveWinResult(int64_t ts, uint64_t groupId, SSHashObj* pUpdatedMap) {
H
Haojun Liao 已提交
848
  SWinKey key = {.ts = ts, .groupId = groupId};
849
  tSimpleHashPut(pUpdatedMap, &key, sizeof(SWinKey), NULL, 0);
5
54liuyao 已提交
850
  return TSDB_CODE_SUCCESS;
5
54liuyao 已提交
851 852
}

853 854
static int32_t saveWinResultInfo(TSKEY ts, uint64_t groupId, SSHashObj* pUpdatedMap) {
  return saveWinResult(ts, groupId, pUpdatedMap);
5
54liuyao 已提交
855 856
}

857
static void removeResults(SArray* pWins, SSHashObj* pUpdatedMap) {
5
54liuyao 已提交
858 859
  int32_t size = taosArrayGetSize(pWins);
  for (int32_t i = 0; i < size; i++) {
H
Haojun Liao 已提交
860
    SWinKey* pW = taosArrayGet(pWins, i);
861
    void*    tmp = tSimpleHashGet(pUpdatedMap, pW, sizeof(SWinKey));
5
54liuyao 已提交
862 863 864
    if (tmp) {
      void* value = *(void**)tmp;
      taosMemoryFree(value);
865
      tSimpleHashRemove(pUpdatedMap, pW, sizeof(SWinKey));
5
54liuyao 已提交
866
    }
5
54liuyao 已提交
867 868 869
  }
}

870 871 872 873 874 875
int32_t compareWinKey(void* pKey, void* data, int32_t index) {
  SArray*  res = (SArray*)data;
  SWinKey* pDataPos = taosArrayGet(res, index);
  SWinKey* pWKey = (SWinKey*)pKey;

  if (pWKey->groupId > pDataPos->groupId) {
5
54liuyao 已提交
876
    return 1;
877
  } else if (pWKey->groupId < pDataPos->groupId) {
5
54liuyao 已提交
878
    return -1;
5
54liuyao 已提交
879
  }
L
Liu Jicong 已提交
880

881
  if (pWKey->ts > pDataPos->ts) {
5
54liuyao 已提交
882
    return 1;
883
  } else if (pWKey->ts < pDataPos->ts) {
5
54liuyao 已提交
884 885 886
    return -1;
  }
  return 0;
5
54liuyao 已提交
887 888
}

889
static void removeDeleteResults(SSHashObj* pUpdatedMap, SArray* pDelWins) {
5
54liuyao 已提交
890 891
  taosArraySort(pDelWins, winKeyCmprImpl);
  taosArrayRemoveDuplicate(pDelWins, winKeyCmprImpl, NULL);
L
Liu Jicong 已提交
892
  int32_t delSize = taosArrayGetSize(pDelWins);
893
  if (tSimpleHashGetSize(pUpdatedMap) == 0 || delSize == 0) {
5
54liuyao 已提交
894
    return;
dengyihao's avatar
dengyihao 已提交
895
  }
896 897 898 899 900 901
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pUpdatedMap, pIte, &iter)) != NULL) {
    SWinKey* pResKey = tSimpleHashGetKey(pIte, NULL);
    int32_t  index = binarySearchCom(pDelWins, delSize, pResKey, TSDB_ORDER_DESC, compareWinKey);
    if (index >= 0 && 0 == compareWinKey(pResKey, pDelWins, index)) {
902
      taosArrayRemove(pDelWins, index);
903
      delSize = taosArrayGetSize(pDelWins);
904 905 906 907
    }
  }
}

5
54liuyao 已提交
908
bool isOverdue(TSKEY ekey, STimeWindowAggSupp* pTwSup) {
5
54liuyao 已提交
909
  ASSERTS(pTwSup->maxTs == INT64_MIN || pTwSup->maxTs > 0, "maxts should greater than 0");
5
54liuyao 已提交
910
  return pTwSup->maxTs != INT64_MIN && ekey < pTwSup->maxTs - pTwSup->waterMark;
5
54liuyao 已提交
911 912
}

5
54liuyao 已提交
913 914 915 916 917
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pTwSup) { return isOverdue(pWin->ekey, pTwSup); }

bool needDeleteWindowBuf(STimeWindow* pWin, STimeWindowAggSupp* pTwSup) {
  return pTwSup->maxTs != INT64_MIN && pWin->ekey < pTwSup->maxTs - pTwSup->deleteMark;
}
5
54liuyao 已提交
918

5
54liuyao 已提交
919
static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pBlock,
920
                            int32_t scanFlag) {
921
  SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
922

923
  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
924
  SExprSupp*     pSup = &pOperatorInfo->exprSupp;
925

X
Xiaoyu Wang 已提交
926
  int32_t     startPos = 0;
927
  int32_t     numOfOutput = pSup->numOfExprs;
X
Xiaoyu Wang 已提交
928
  int64_t*    tsCols = extractTsCol(pBlock, pInfo);
H
Haojun Liao 已提交
929
  uint64_t    tableGroupId = pBlock->info.id.groupId;
930
  bool        ascScan = (pInfo->inputOrder == TSDB_ORDER_ASC);
X
Xiaoyu Wang 已提交
931 932
  TSKEY       ts = getStartTsKey(&pBlock->info.window, tsCols);
  SResultRow* pResult = NULL;
933

934 935
  STimeWindow win =
      getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->inputOrder);
936 937
  int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
938
  if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
S
Shengliang Guan 已提交
939
    T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
940
  }
941

X
Xiaoyu Wang 已提交
942 943
  TSKEY   ekey = ascScan ? win.ekey : win.skey;
  int32_t forwardRows =
944
      getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder);
945 946

  // prev time window not interpolation yet.
947
  if (pInfo->timeWindowInterpo) {
948
    SResultRowPosition pos = addToOpenWindowList(pResultRowInfo, pResult, tableGroupId);
949
    doInterpUnclosedTimeWindow(pOperatorInfo, numOfOutput, pResultRowInfo, pBlock, scanFlag, tsCols, &pos);
950 951

    // restore current time window
952 953
    ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx,
                                 numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
954
    if (ret != TSDB_CODE_SUCCESS) {
S
Shengliang Guan 已提交
955
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
956 957
    }

958
    // window start key interpolation
959
    doWindowBorderInterpolation(pInfo, pBlock, pResult, &win, startPos, forwardRows, pSup);
960
  }
961

962
  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &win, true);
L
Liu Jicong 已提交
963 964
  applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, forwardRows,
                                  pBlock->info.rows, numOfOutput);
965 966

  doCloseWindow(pResultRowInfo, pInfo, pResult);
967 968 969

  STimeWindow nextWin = win;
  while (1) {
970
    int32_t prevEndPos = forwardRows - 1 + startPos;
971
    startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, pInfo->inputOrder);
972 973 974 975
    if (startPos < 0) {
      break;
    }
    // null data, failed to allocate more memory buffer
X
Xiaoyu Wang 已提交
976
    int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
977
                                          pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
978
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
S
Shengliang Guan 已提交
979
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
980 981
    }

X
Xiaoyu Wang 已提交
982
    ekey = ascScan ? nextWin.ekey : nextWin.skey;
983
    forwardRows =
984
        getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder);
985
    // window start(end) key interpolation
986
    doWindowBorderInterpolation(pInfo, pBlock, pResult, &nextWin, startPos, forwardRows, pSup);
L
Liu Jicong 已提交
987
    // TODO: add to open window? how to close the open windows after input blocks exhausted?
S
shenglian zhou 已提交
988
#if 0
989 990 991 992
    if ((ascScan && ekey <= pBlock->info.window.ekey) ||
        (!ascScan && ekey >= pBlock->info.window.skey)) {
      // window start(end) key interpolation
      doWindowBorderInterpolation(pInfo, pBlock, pResult, &nextWin, startPos, forwardRows, pSup);
993
    } else if (pInfo->timeWindowInterpo) {
994 995
      addToOpenWindowList(pResultRowInfo, pResult, tableGroupId);
    }
S
shenglian zhou 已提交
996
#endif
997
    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true);
L
Liu Jicong 已提交
998 999
    applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, forwardRows,
                                    pBlock->info.rows, numOfOutput);
1000
    doCloseWindow(pResultRowInfo, pInfo, pResult);
1001 1002 1003
  }

  if (pInfo->timeWindowInterpo) {
1004
    saveDataBlockLastRow(pInfo->pPrevValues, pBlock, pInfo->pInterpCols);
1005
  }
1006 1007 1008 1009 1010 1011
}

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);
1012
    SListNode* pNode = tdListPopHead(pResultRowInfo->openWindow);
D
dapan1121 已提交
1013
    taosMemoryFree(pNode);
1014 1015 1016
  }
}

1017 1018 1019 1020 1021
SResultRowPosition addToOpenWindowList(SResultRowInfo* pResultRowInfo, const SResultRow* pResult, uint64_t groupId) {
  SOpenWindowInfo openWin = {0};
  openWin.pos.pageId = pResult->pageId;
  openWin.pos.offset = pResult->offset;
  openWin.groupId = groupId;
L
Liu Jicong 已提交
1022
  SListNode* pn = tdListGetTail(pResultRowInfo->openWindow);
1023
  if (pn == NULL) {
1024 1025
    tdListAppend(pResultRowInfo->openWindow, &openWin);
    return openWin.pos;
1026 1027
  }

L
Liu Jicong 已提交
1028
  SOpenWindowInfo* px = (SOpenWindowInfo*)pn->data;
1029 1030
  if (px->pos.pageId != openWin.pos.pageId || px->pos.offset != openWin.pos.offset || px->groupId != openWin.groupId) {
    tdListAppend(pResultRowInfo->openWindow, &openWin);
1031 1032
  }

1033
  return openWin.pos;
1034 1035 1036 1037
}

int64_t* extractTsCol(SSDataBlock* pBlock, const SIntervalAggOperatorInfo* pInfo) {
  TSKEY* tsCols = NULL;
1038

D
dapan1121 已提交
1039
  if (pBlock->pDataBlock != NULL && pBlock->info.dataLoad) {
1040 1041
    SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
    tsCols = (int64_t*)pColDataInfo->pData;
H
Haojun Liao 已提交
1042
    ASSERT(tsCols[0] != 0);
1043

1044 1045 1046 1047 1048 1049
    // no data in primary ts
    if (tsCols[0] == 0 && tsCols[pBlock->info.rows - 1] == 0) {
      return NULL;
    }

    if (tsCols[0] != 0 && (pBlock->info.window.skey == 0 && pBlock->info.window.ekey == 0)) {
1050 1051 1052 1053 1054
      blockDataUpdateTsWindow(pBlock, pInfo->primaryTsIndex);
    }
  }

  return tsCols;
1055 1056 1057 1058 1059 1060 1061
}

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

1062 1063 1064
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
  SOperatorInfo* downstream = pOperator->pDownstream[0];

1065
  SIntervalAggOperatorInfo* pInfo = pOperator->info;
1066
  SExprSupp*                pSup = &pOperator->exprSupp;
1067

1068
  int32_t scanFlag = MAIN_SCAN;
1069
  int64_t st = taosGetTimestampUs();
1070 1071

  while (1) {
1072
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1073 1074 1075 1076
    if (pBlock == NULL) {
      break;
    }

1077
    getTableScanInfo(pOperator, &pInfo->inputOrder, &scanFlag, true);
1078

1079
    if (pInfo->scalarSupp.pExprInfo != NULL) {
L
Liu Jicong 已提交
1080 1081
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
1082 1083
    }

1084
    // the pDataBlock are always the same one, no need to call this again
1085
    setInputDataBlock(pSup, pBlock, pInfo->inputOrder, scanFlag, true);
1086
    hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag);
1087 1088
  }

1089
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->resultTsOrder);
1090
  OPTR_SET_OPENED(pOperator);
1091 1092

  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
1093 1094 1095
  return TSDB_CODE_SUCCESS;
}

1096 1097 1098 1099 1100
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 {
D
dapan1121 已提交
1101
      return memcmp(varDataVal(v), varDataVal(pKey->pData), varDataLen(v)) == 0;
1102 1103 1104 1105 1106 1107
    }
  } else {
    return memcmp(pKey->pData, v, pKey->bytes) == 0;
  }
}

1108
static void doStateWindowAggImpl(SOperatorInfo* pOperator, SStateWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {
L
Liu Jicong 已提交
1109
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1110
  SExprSupp*     pSup = &pOperator->exprSupp;
1111

1112
  SColumnInfoData* pStateColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->stateCol.slotId);
H
Haojun Liao 已提交
1113
  int64_t          gid = pBlock->info.id.groupId;
1114 1115

  bool    masterScan = true;
1116
  int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
1117 1118
  int16_t bytes = pStateColInfoData->info.bytes;

1119
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
1120 1121 1122 1123 1124
  TSKEY*           tsList = (TSKEY*)pColInfoData->pData;

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

1125
  struct SColumnDataAgg* pAgg = NULL;
1126
  for (int32_t j = 0; j < pBlock->info.rows; ++j) {
X
Xiaoyu Wang 已提交
1127
    pAgg = (pBlock->pBlockAgg != NULL) ? pBlock->pBlockAgg[pInfo->stateCol.slotId] : NULL;
1128
    if (colDataIsNull(pStateColInfoData, pBlock->info.rows, j, pAgg)) {
1129 1130 1131 1132 1133
      continue;
    }

    char* val = colDataGetData(pStateColInfoData, j);

1134
    if (gid != pRowSup->groupId || !pInfo->hasKey) {
1135 1136 1137 1138 1139 1140 1141
      // todo extract method
      if (IS_VAR_DATA_TYPE(pInfo->stateKey.type)) {
        varDataCopy(pInfo->stateKey.pData, val);
      } else {
        memcpy(pInfo->stateKey.pData, val, bytes);
      }

1142 1143
      pInfo->hasKey = true;

1144 1145
      doKeepNewWindowStartInfo(pRowSup, tsList, j, gid);
      doKeepTuple(pRowSup, tsList[j], gid);
1146
    } else if (compareVal(val, &pInfo->stateKey)) {
1147
      doKeepTuple(pRowSup, tsList[j], gid);
1148 1149 1150 1151 1152 1153 1154 1155 1156 1157
      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;
1158 1159
      int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx,
                                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1160
      if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
S
Shengliang Guan 已提交
1161
        T_LONG_JMP(pTaskInfo->env, TSDB_CODE_APP_ERROR);
1162 1163 1164
      }

      updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &window, false);
H
Haojun Liao 已提交
1165
      applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
L
Liu Jicong 已提交
1166
                                      pRowSup->numOfRows, pBlock->info.rows, numOfOutput);
1167 1168

      // here we start a new session window
1169 1170
      doKeepNewWindowStartInfo(pRowSup, tsList, j, gid);
      doKeepTuple(pRowSup, tsList[j], gid);
1171 1172 1173 1174 1175 1176 1177

      // todo extract method
      if (IS_VAR_DATA_TYPE(pInfo->stateKey.type)) {
        varDataCopy(pInfo->stateKey.pData, val);
      } else {
        memcpy(pInfo->stateKey.pData, val, bytes);
      }
1178 1179 1180 1181 1182
    }
  }

  SResultRow* pResult = NULL;
  pRowSup->win.ekey = tsList[pBlock->info.rows - 1];
1183 1184
  int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1185
  if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
S
Shengliang Guan 已提交
1186
    T_LONG_JMP(pTaskInfo->env, TSDB_CODE_APP_ERROR);
1187 1188 1189
  }

  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, false);
L
Liu Jicong 已提交
1190 1191
  applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
                                  pRowSup->numOfRows, pBlock->info.rows, numOfOutput);
1192 1193
}

H
Hongze Cheng 已提交
1194
static int32_t openStateWindowAggOptr(SOperatorInfo* pOperator) {
1195 1196
  if (OPTR_IS_OPENED(pOperator)) {
    return TSDB_CODE_SUCCESS;
1197 1198 1199
  }

  SStateWindowOperatorInfo* pInfo = pOperator->info;
1200
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
1201

1202 1203 1204
  SExprSupp* pSup = &pOperator->exprSupp;
  int32_t    order = TSDB_ORDER_ASC;
  int64_t    st = taosGetTimestampUs();
1205 1206 1207

  SOperatorInfo* downstream = pOperator->pDownstream[0];
  while (1) {
1208
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1209 1210 1211 1212
    if (pBlock == NULL) {
      break;
    }

1213
    setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
1214 1215
    blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);

1216 1217 1218 1219 1220 1221 1222 1223 1224
    // there is an scalar expression that needs to be calculated right before apply the group aggregation.
    if (pInfo->scalarSup.pExprInfo != NULL) {
      pTaskInfo->code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
                                              pInfo->scalarSup.numOfExprs, NULL);
      if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
        T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
      }
    }

1225 1226 1227
    doStateWindowAggImpl(pOperator, pInfo, pBlock);
  }

X
Xiaoyu Wang 已提交
1228
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;
1229
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
1230 1231
  pOperator->status = OP_RES_TO_RETURN;

1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245
  return TSDB_CODE_SUCCESS;
}

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

  SStateWindowOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
  SOptrBasicInfo*           pBInfo = &pInfo->binfo;

  pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
  if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
H
Haojun Liao 已提交
1246
    setOperatorCompleted(pOperator);
1247 1248 1249
    return NULL;
  }

1250
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
1251
  while (1) {
1252
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
H
Haojun Liao 已提交
1253
    doFilter(pBInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
1254

1255
    bool hasRemain = hasRemainResults(&pInfo->groupResInfo);
1256
    if (!hasRemain) {
H
Haojun Liao 已提交
1257
      setOperatorCompleted(pOperator);
1258 1259
      break;
    }
1260

1261 1262 1263 1264
    if (pBInfo->pRes->info.rows > 0) {
      break;
    }
  }
1265

1266
  pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1267
  return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1268 1269
}

1270
static SSDataBlock* doBuildIntervalResult(SOperatorInfo* pOperator) {
1271
  SIntervalAggOperatorInfo* pInfo = pOperator->info;
L
Liu Jicong 已提交
1272
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
1273 1274 1275 1276 1277 1278

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

  SSDataBlock* pBlock = pInfo->binfo.pRes;
1279 1280 1281 1282
  pTaskInfo->code = pOperator->fpSet._openFn(pOperator);
  if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
    return NULL;
  }
1283

1284 1285 1286
  while (1) {
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
    doFilter(pBlock, pOperator->exprSupp.pFilterInfo, NULL);
1287

1288 1289
    bool hasRemain = hasRemainResults(&pInfo->groupResInfo);
    if (!hasRemain) {
H
Haojun Liao 已提交
1290
      setOperatorCompleted(pOperator);
1291
      break;
1292 1293
    }

1294 1295 1296
    if (pBlock->info.rows > 0) {
      break;
    }
1297
  }
1298 1299 1300 1301 1302

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

  return (rows == 0) ? NULL : pBlock;
1303 1304
}

5
54liuyao 已提交
1305
static void setInverFunction(SqlFunctionCtx* pCtx, int32_t num, EStreamType type) {
L
Liu Jicong 已提交
1306
  for (int i = 0; i < num; i++) {
5
54liuyao 已提交
1307 1308
    if (type == STREAM_INVERT) {
      fmSetInvertFunc(pCtx[i].functionId, &(pCtx[i].fpSet));
L
Liu Jicong 已提交
1309
    } else if (type == STREAM_NORMAL) {
5
54liuyao 已提交
1310 1311 1312 1313
      fmSetNormalFunc(pCtx[i].functionId, &(pCtx[i].fpSet));
    }
  }
}
5
54liuyao 已提交
1314

5
54liuyao 已提交
1315
static void doClearWindowImpl(SResultRowPosition* p1, SDiskbasedBuf* pResultBuf, SExprSupp* pSup, int32_t numOfOutput) {
1316
  SResultRow* pResult = getResultRowByPos(pResultBuf, p1, false);
1317 1318 1319
  if (NULL == pResult) {
    return;
  }
1320

1321
  SqlFunctionCtx* pCtx = pSup->pCtx;
5
54liuyao 已提交
1322
  for (int32_t i = 0; i < numOfOutput; ++i) {
1323
    pCtx[i].resultInfo = getResultEntryInfo(pResult, i, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
1324 1325 1326 1327 1328 1329 1330 1331 1332
    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 已提交
1333
  SFilePage* bufPage = getBufPage(pResultBuf, p1->pageId);
1334 1335 1336
  if (NULL == bufPage) {
    return;
  }
5
54liuyao 已提交
1337 1338
  setBufPageDirty(bufPage, true);
  releaseBufPage(pResultBuf, bufPage);
5
54liuyao 已提交
1339 1340
}

1341
static bool doDeleteWindow(SOperatorInfo* pOperator, TSKEY ts, uint64_t groupId) {
5
54liuyao 已提交
1342 1343 1344
  SStreamIntervalOperatorInfo* pInfo = pOperator->info;
  SWinKey                      key = {.ts = ts, .groupId = groupId};
  tSimpleHashRemove(pInfo->aggSup.pResultRowHashTable, &key, sizeof(SWinKey));
1345
  streamStateDel(pInfo->pState, &key);
5
54liuyao 已提交
1346 1347 1348
  return true;
}

1349
static void doDeleteWindows(SOperatorInfo* pOperator, SInterval* pInterval, SSDataBlock* pBlock, SArray* pUpWins,
1350
                            SSHashObj* pUpdatedMap) {
1351 1352 1353 1354 1355 1356 1357 1358 1359 1360 1361
  SStreamIntervalOperatorInfo* pInfo = pOperator->info;
  SColumnInfoData*             pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
  TSKEY*                       startTsCols = (TSKEY*)pStartTsCol->pData;
  SColumnInfoData*             pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
  TSKEY*                       endTsCols = (TSKEY*)pEndTsCol->pData;
  SColumnInfoData*             pCalStTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
  TSKEY*                       calStTsCols = (TSKEY*)pCalStTsCol->pData;
  SColumnInfoData*             pCalEnTsCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
  TSKEY*                       calEnTsCols = (TSKEY*)pCalEnTsCol->pData;
  SColumnInfoData*             pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
  uint64_t*                    pGpDatas = (uint64_t*)pGpCol->pData;
5
54liuyao 已提交
1362
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
H
Haojun Liao 已提交
1363
    SResultRowInfo dumyInfo = {0};
5
54liuyao 已提交
1364
    dumyInfo.cur.pageId = -1;
H
Haojun Liao 已提交
1365

1366 1367 1368 1369 1370 1371 1372 1373
    STimeWindow win = {0};
    if (IS_FINAL_OP(pInfo)) {
      win.skey = startTsCols[i];
      win.ekey = endTsCols[i];
    } else {
      win = getActiveTimeWindow(NULL, &dumyInfo, startTsCols[i], pInterval, TSDB_ORDER_ASC);
    }

5
54liuyao 已提交
1374
    do {
1375 1376 1377 1378
      if (!inCalSlidingWindow(pInterval, &win, calStTsCols[i], calEnTsCols[i])) {
        getNextTimeWindow(pInterval, pInterval->precision, TSDB_ORDER_ASC, &win);
        continue;
      }
5
54liuyao 已提交
1379
      uint64_t winGpId = pGpDatas[i];
1380
      bool     res = doDeleteWindow(pOperator, win.skey, winGpId);
5
54liuyao 已提交
1381 1382 1383 1384 1385
      SWinKey  winRes = {.ts = win.skey, .groupId = winGpId};
      if (pUpWins && res) {
        taosArrayPush(pUpWins, &winRes);
      }
      if (pUpdatedMap) {
1386
        tSimpleHashRemove(pUpdatedMap, &winRes, sizeof(SWinKey));
5
54liuyao 已提交
1387 1388
      }
      getNextTimeWindow(pInterval, pInterval->precision, TSDB_ORDER_ASC, &win);
5
54liuyao 已提交
1389
    } while (win.ekey <= endTsCols[i]);
5
54liuyao 已提交
1390 1391 1392
  }
}

1393
static int32_t getAllIntervalWindow(SSHashObj* pHashMap, SSHashObj* resWins) {
1394 1395 1396
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pHashMap, pIte, &iter)) != NULL) {
1397 1398 1399 1400
    SWinKey* pKey = tSimpleHashGetKey(pIte, NULL);
    uint64_t groupId = pKey->groupId;
    TSKEY    ts = pKey->ts;
    int32_t  code = saveWinResult(ts, groupId, resWins);
5
54liuyao 已提交
1401 1402 1403 1404 1405 1406 1407
    if (code != TSDB_CODE_SUCCESS) {
      return code;
    }
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
1408
static int32_t closeStreamIntervalWindow(SSHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SInterval* pInterval,
1409
                                         SHashObj* pPullDataMap, SSHashObj* closeWins, SArray* pDelWins,
1410
                                         SOperatorInfo* pOperator) {
5
54liuyao 已提交
1411
  qDebug("===stream===close interval window");
1412 1413 1414
  void*                        pIte = NULL;
  int32_t                      iter = 0;
  SStreamIntervalOperatorInfo* pInfo = pOperator->info;
1415
  int32_t                      delSize = taosArrayGetSize(pDelWins);
5
54liuyao 已提交
1416
  while ((pIte = tSimpleHashIterate(pHashMap, pIte, &iter)) != NULL) {
1417
    void*    key = tSimpleHashGetKey(pIte, NULL);
1418 1419 1420 1421 1422 1423 1424 1425 1426
    SWinKey* pWinKey = (SWinKey*)key;
    if (delSize > 0) {
      int32_t index = binarySearchCom(pDelWins, delSize, pWinKey, TSDB_ORDER_DESC, compareWinKey);
      if (index >= 0 && 0 == compareWinKey(pWinKey, pDelWins, index)) {
        taosArrayRemove(pDelWins, index);
        delSize = taosArrayGetSize(pDelWins);
      }
    }

5
54liuyao 已提交
1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451
    void*       chIds = taosHashGet(pPullDataMap, pWinKey, sizeof(SWinKey));
    STimeWindow win = {
        .skey = pWinKey->ts,
        .ekey = taosTimeAdd(win.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1,
    };
    if (isCloseWindow(&win, pTwSup)) {
      if (chIds && pPullDataMap) {
        SArray* chAy = *(SArray**)chIds;
        int32_t size = taosArrayGetSize(chAy);
        qDebug("===stream===window %" PRId64 " wait child size:%d", pWinKey->ts, size);
        for (int32_t i = 0; i < size; i++) {
          qDebug("===stream===window %" PRId64 " wait child id:%d", pWinKey->ts, *(int32_t*)taosArrayGet(chAy, i));
        }
        continue;
      } else if (pPullDataMap) {
        qDebug("===stream===close window %" PRId64, pWinKey->ts);
      }

      if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
        int32_t code = saveWinResultInfo(pWinKey->ts, pWinKey->groupId, closeWins);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
        }
      }
      tSimpleHashIterateRemove(pHashMap, pWinKey, sizeof(SWinKey), &pIte, &iter);
1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496
    }
  }
  return TSDB_CODE_SUCCESS;
}

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

static void deleteIntervalDiscBuf(SStreamState* pState, SHashObj* pPullDataMap, TSKEY mark, SInterval* pInterval,
                                  SWinKey* key) {
  STimeWindow tw = getFinalTimeWindow(key->ts, pInterval);
  SWinKey     next = {0};
  while (tw.ekey < mark) {
    SStreamStateCur* pCur = streamStateSeekKeyNext(pState, key);
    int32_t          code = streamStateGetKVByCur(pCur, &next, NULL, 0);
    streamStateFreeCur(pCur);

    void* chIds = taosHashGet(pPullDataMap, key, sizeof(SWinKey));
    if (chIds && pPullDataMap) {
      SArray* chAy = *(SArray**)chIds;
      int32_t size = taosArrayGetSize(chAy);
      qDebug("===stream===window %" PRId64 " wait child size:%d", key->ts, size);
      for (int32_t i = 0; i < size; i++) {
        qDebug("===stream===window %" PRId64 " wait child id:%d", key->ts, *(int32_t*)taosArrayGet(chAy, i));
      }
      break;
    }
    qDebug("===stream===delete window %" PRId64, key->ts);
    int32_t codeDel = streamStateDel(pState, key);
    if (codeDel != TSDB_CODE_SUCCESS) {
      code = streamStateGetFirst(pState, key);
      if (code != TSDB_CODE_SUCCESS) {
        qDebug("===stream===stream state first key: empty-empty");
        return;
      }
      continue;
    }
    if (code == TSDB_CODE_SUCCESS) {
      *key = next;
      tw = getFinalTimeWindow(key->ts, pInterval);
    }
  }
5
54liuyao 已提交
1497

5
54liuyao 已提交
1498 1499
  // for debug
  if (qDebugFlag & DEBUG_DEBUG && mark > 0) {
1500 1501 1502 1503 1504 1505 1506
    SStreamStateCur* pCur = streamStateGetCur(pState, key);
    int32_t          code = streamStateCurPrev(pState, pCur);
    if (code == TSDB_CODE_SUCCESS) {
      SWinKey tmpKey = {0};
      code = streamStateGetKVByCur(pCur, &tmpKey, NULL, 0);
      if (code == TSDB_CODE_SUCCESS) {
        STimeWindow tw = getFinalTimeWindow(tmpKey.ts, pInterval);
1507
        qDebug("===stream===error stream state first key:%" PRId64 "-%" PRId64 ",%" PRIu64 ",mark %" PRId64, tw.skey,
1508 1509 1510
               tw.ekey, tmpKey.groupId, mark);
      } else {
        STimeWindow tw = getFinalTimeWindow(key->ts, pInterval);
1511
        qDebug("===stream===stream state first key:%" PRId64 "-%" PRId64 ",%" PRIu64 ",mark %" PRId64, tw.skey, tw.ekey,
1512
               key->groupId, mark);
5
54liuyao 已提交
1513
      }
1514 1515
    } else {
      STimeWindow tw = getFinalTimeWindow(key->ts, pInterval);
1516
      qDebug("===stream===stream state first key:%" PRId64 "-%" PRId64 ",%" PRIu64 ",mark %" PRId64, tw.skey, tw.ekey,
1517
             key->groupId, mark);
5
54liuyao 已提交
1518
    }
1519
    streamStateFreeCur(pCur);
5
54liuyao 已提交
1520 1521 1522
  }
}

1523
static void closeChildIntervalWindow(SOperatorInfo* pOperator, SArray* pChildren, TSKEY maxTs) {
5
54liuyao 已提交
1524 1525
  int32_t size = taosArrayGetSize(pChildren);
  for (int32_t i = 0; i < size; i++) {
1526 1527
    SOperatorInfo*               pChildOp = taosArrayGetP(pChildren, i);
    SStreamIntervalOperatorInfo* pChInfo = pChildOp->info;
5
54liuyao 已提交
1528
    ASSERTS(pChInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE, "children trigger type should be at once");
5
54liuyao 已提交
1529
    pChInfo->twAggSup.maxTs = TMAX(pChInfo->twAggSup.maxTs, maxTs);
1530
    closeStreamIntervalWindow(pChInfo->aggSup.pResultRowHashTable, &pChInfo->twAggSup, &pChInfo->interval, NULL, NULL,
1531
                              NULL, pOperator);
1532 1533 1534
  }
}

1535 1536
static void doBuildDeleteResult(SStreamIntervalOperatorInfo* pInfo, SArray* pWins, int32_t* index,
                                SSDataBlock* pBlock) {
1537 1538 1539 1540 1541 1542 1543 1544
  blockDataCleanup(pBlock);
  int32_t size = taosArrayGetSize(pWins);
  if (*index == size) {
    *index = 0;
    taosArrayClear(pWins);
    return;
  }
  blockDataEnsureCapacity(pBlock, size - *index);
1545
  uint64_t uid = 0;
1546
  for (int32_t i = *index; i < size; i++) {
H
Haojun Liao 已提交
1547
    SWinKey* pWin = taosArrayGet(pWins, i);
1548 1549
    void*    tbname = NULL;
    streamStateGetParName(pInfo->pState, pWin->groupId, &tbname);
1550 1551 1552 1553 1554 1555 1556
    if (tbname == NULL) {
      appendOneRowToStreamSpecialBlock(pBlock, &pWin->ts, &pWin->ts, &uid, &pWin->groupId, NULL);
    } else {
      char parTbName[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN];
      STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));
      appendOneRowToStreamSpecialBlock(pBlock, &pWin->ts, &pWin->ts, &uid, &pWin->groupId, parTbName);
    }
1557
    tdbFree(tbname);
1558
    (*index)++;
5
54liuyao 已提交
1559 1560 1561
  }
}

1562
static void destroyStateWindowOperatorInfo(void* param) {
1563
  SStateWindowOperatorInfo* pInfo = (SStateWindowOperatorInfo*)param;
1564
  cleanupBasicInfo(&pInfo->binfo);
1565
  taosMemoryFreeClear(pInfo->stateKey.pData);
1566
  cleanupExprSupp(&pInfo->scalarSup);
D
dapan1121 已提交
1567 1568 1569
  colDataDestroy(&pInfo->twAggSup.timeWindowData);
  cleanupAggSup(&pInfo->aggSup);
  cleanupGroupResInfo(&pInfo->groupResInfo);
1570

D
dapan1121 已提交
1571
  taosMemoryFreeClear(param);
1572 1573
}

H
Haojun Liao 已提交
1574
static void freeItem(void* param) {
L
Liu Jicong 已提交
1575
  SGroupKeys* pKey = (SGroupKeys*)param;
H
Haojun Liao 已提交
1576 1577 1578
  taosMemoryFree(pKey->pData);
}

1579
void destroyIntervalOperatorInfo(void* param) {
1580
  SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)param;
1581
  cleanupBasicInfo(&pInfo->binfo);
1582
  cleanupAggSup(&pInfo->aggSup);
1583 1584 1585 1586
  cleanupExprSupp(&pInfo->scalarSupp);

  tdListFree(pInfo->binfo.resultRowInfo.openWindow);

H
Haojun Liao 已提交
1587 1588 1589 1590
  pInfo->pInterpCols = taosArrayDestroy(pInfo->pInterpCols);
  taosArrayDestroyEx(pInfo->pPrevValues, freeItem);

  pInfo->pPrevValues = NULL;
1591

H
Haojun Liao 已提交
1592 1593
  cleanupGroupResInfo(&pInfo->groupResInfo);
  colDataDestroy(&pInfo->twAggSup.timeWindowData);
D
dapan1121 已提交
1594
  taosMemoryFreeClear(param);
1595 1596
}

1597
void destroyStreamFinalIntervalOperatorInfo(void* param) {
1598
  SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)param;
1599
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
1600
  cleanupAggSup(&pInfo->aggSup);
L
Liu Jicong 已提交
1601
  // it should be empty.
5
54liuyao 已提交
1602 1603 1604
  taosHashCleanup(pInfo->pPullDataMap);
  taosArrayDestroy(pInfo->pPullWins);
  blockDataDestroy(pInfo->pPullDataRes);
L
Liu Jicong 已提交
1605 1606
  taosArrayDestroy(pInfo->pDelWins);
  blockDataDestroy(pInfo->pDelRes);
1607
  taosMemoryFreeClear(pInfo->pState);
5
54liuyao 已提交
1608

1609 1610 1611 1612
  if (pInfo->pChildren) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, i);
5
54liuyao 已提交
1613
      destroyOperatorInfo(pChildOp);
1614
    }
L
Liu Jicong 已提交
1615
    taosArrayDestroy(pInfo->pChildren);
1616
  }
1617
  nodesDestroyNode((SNode*)pInfo->pPhyNode);
5
54liuyao 已提交
1618
  colDataDestroy(&pInfo->twAggSup.timeWindowData);
5
54liuyao 已提交
1619
  pInfo->groupResInfo.pRows = taosArrayDestroy(pInfo->groupResInfo.pRows);
5
54liuyao 已提交
1620
  cleanupExprSupp(&pInfo->scalarSupp);
1621

D
dapan1121 已提交
1622
  taosMemoryFreeClear(param);
5
54liuyao 已提交
1623 1624
}

1625
static bool allInvertible(SqlFunctionCtx* pFCtx, int32_t numOfCols) {
5
54liuyao 已提交
1626
  for (int32_t i = 0; i < numOfCols; i++) {
5
54liuyao 已提交
1627
    if (fmIsUserDefinedFunc(pFCtx[i].functionId) || !fmIsInvertible(pFCtx[i].functionId)) {
5
54liuyao 已提交
1628 1629 1630 1631 1632 1633
      return false;
    }
  }
  return true;
}

1634
static bool timeWindowinterpNeeded(SqlFunctionCtx* pCtx, int32_t numOfCols, SIntervalAggOperatorInfo* pInfo) {
1635 1636 1637
  // the primary timestamp column
  bool needed = false;

L
Liu Jicong 已提交
1638
  for (int32_t i = 0; i < numOfCols; ++i) {
1639
    SExprInfo* pExpr = pCtx[i].pExpr;
H
Haojun Liao 已提交
1640
    if (fmIsIntervalInterpoFunc(pCtx[i].functionId)) {
1641
      needed = true;
H
Haojun Liao 已提交
1642
      break;
1643 1644 1645
    }
  }

H
Haojun Liao 已提交
1646 1647 1648
  if (needed) {
    pInfo->pInterpCols = taosArrayInit(4, sizeof(SColumn));
    pInfo->pPrevValues = taosArrayInit(4, sizeof(SGroupKeys));
1649

H
Haojun Liao 已提交
1650 1651 1652 1653 1654 1655 1656 1657 1658 1659 1660 1661 1662 1663 1664
    {  // ts column
      SColumn c = {0};
      c.colId = 1;
      c.slotId = pInfo->primaryTsIndex;
      c.type = TSDB_DATA_TYPE_TIMESTAMP;
      c.bytes = sizeof(int64_t);
      taosArrayPush(pInfo->pInterpCols, &c);

      SGroupKeys key;
      key.bytes = c.bytes;
      key.type = c.type;
      key.isNull = true;  // to denote no value is assigned yet
      key.pData = taosMemoryCalloc(1, c.bytes);
      taosArrayPush(pInfo->pPrevValues, &key);
    }
1665 1666
  }

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

H
Haojun Liao 已提交
1670
    if (fmIsIntervalInterpoFunc(pCtx[i].functionId)) {
1671 1672 1673
      SFunctParam* pParam = &pExpr->base.pParam[0];

      SColumn c = *pParam->pCol;
1674
      taosArrayPush(pInfo->pInterpCols, &c);
1675 1676

      SGroupKeys key = {0};
X
Xiaoyu Wang 已提交
1677 1678
      key.bytes = c.bytes;
      key.type = c.type;
1679
      key.isNull = false;
X
Xiaoyu Wang 已提交
1680
      key.pData = taosMemoryCalloc(1, c.bytes);
1681
      taosArrayPush(pInfo->pPrevValues, &key);
1682 1683 1684 1685 1686 1687
    }
  }

  return needed;
}

L
Liu Jicong 已提交
1688
void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSupporter* pSup, SInterval* pInterval,
5
54liuyao 已提交
1689
                            STimeWindowAggSupp* pTwSup) {
1690
  if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
5
54liuyao 已提交
1691
    initIntervalDownStream(downstream->pDownstream[0], type, pSup, pInterval, pTwSup);
1692 1693
    return;
  }
5
54liuyao 已提交
1694
  SStreamScanInfo* pScanInfo = downstream->info;
1695 1696
  pScanInfo->windowSup.parentType = type;
  pScanInfo->windowSup.pIntervalAggSup = pSup;
5
54liuyao 已提交
1697
  if (!pScanInfo->igCheckUpdate && !pScanInfo->pUpdateInfo) {
5
54liuyao 已提交
1698 1699
    pScanInfo->pUpdateInfo = updateInfoInitP(pInterval, pTwSup->waterMark);
  }
1700
  pScanInfo->interval = *pInterval;
5
54liuyao 已提交
1701
  pScanInfo->twAggSup = *pTwSup;
5
54liuyao 已提交
1702 1703
}

H
Haojun Liao 已提交
1704 1705
void initStreamFunciton(SqlFunctionCtx* pCtx, int32_t numOfExpr) {
  for (int32_t i = 0; i < numOfExpr; i++) {
L
Liu Jicong 已提交
1706
    //    pCtx[i].isStream = true;
H
Haojun Liao 已提交
1707 1708 1709
  }
}

H
Haojun Liao 已提交
1710
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SIntervalPhysiNode* pPhyNode,
5
54liuyao 已提交
1711
                                          SExecTaskInfo* pTaskInfo) {
1712
  SIntervalAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
L
Liu Jicong 已提交
1713
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
1714 1715 1716 1717
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

H
Haojun Liao 已提交
1718
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->window.node.pOutputDataBlockDesc);
H
Haojun Liao 已提交
1719 1720 1721 1722 1723 1724
  initBasicInfo(&pInfo->binfo, pResBlock);

  SExprSupp* pSup = &pOperator->exprSupp;
  pInfo->primaryTsIndex = ((SColumnNode*)pPhyNode->window.pTspk)->slotId;

  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
H
Haojun Liao 已提交
1725 1726
  initResultSizeInfo(&pOperator->resultInfo, 512);
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
H
Haojun Liao 已提交
1727 1728 1729

  int32_t    num = 0;
  SExprInfo* pExprInfo = createExprInfo(pPhyNode->window.pFuncs, NULL, &num);
L
Liu Jicong 已提交
1730 1731
  int32_t    code =
      initAggSup(pSup, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str, pTaskInfo->streamInfo.pState);
H
Haojun Liao 已提交
1732 1733 1734 1735 1736
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  SInterval interval = {.interval = pPhyNode->interval,
1737 1738 1739 1740 1741
                        .sliding = pPhyNode->sliding,
                        .intervalUnit = pPhyNode->intervalUnit,
                        .slidingUnit = pPhyNode->slidingUnit,
                        .offset = pPhyNode->offset,
                        .precision = ((SColumnNode*)pPhyNode->window.pTspk)->node.resType.precision};
H
Haojun Liao 已提交
1742 1743 1744 1745 1746 1747 1748

  STimeWindowAggSupp as = {
      .waterMark = pPhyNode->window.watermark,
      .calTrigger = pPhyNode->window.triggerType,
      .maxTs = INT64_MIN,
  };

L
Liu Jicong 已提交
1749
  pInfo->win = pTaskInfo->window;
1750 1751
  pInfo->inputOrder = (pPhyNode->window.inputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
  pInfo->resultTsOrder = (pPhyNode->window.outputTsOrder == ORDER_ASC) ? TSDB_ORDER_ASC : TSDB_ORDER_DESC;
H
Haojun Liao 已提交
1752 1753
  pInfo->interval = interval;
  pInfo->twAggSup = as;
1754
  pInfo->binfo.mergeResultBlock = pPhyNode->window.mergeDataBlock;
1755 1756 1757 1758

  if (pPhyNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pPhyNode->window.pExprs, NULL, &numOfScalar);
H
Haojun Liao 已提交
1759
    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
1760 1761 1762 1763
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }
1764

H
Haojun Liao 已提交
1765 1766 1767 1768 1769
  code = filterInitFromNode((SNode*)pPhyNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

1770
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->win);
H
Haojun Liao 已提交
1771
  pInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, num, pInfo);
1772
  if (pInfo->timeWindowInterpo) {
1773
    pInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SOpenWindowInfo));
H
Haojun Liao 已提交
1774 1775 1776
    if (pInfo->binfo.resultRowInfo.openWindow == NULL) {
      goto _error;
    }
1777
  }
1778

1779
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
L
Liu Jicong 已提交
1780 1781
  setOperatorInfo(pOperator, "TimeIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL, true, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
1782

L
Liu Jicong 已提交
1783 1784
  pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, NULL, destroyIntervalOperatorInfo,
                                         optrDefaultBufFn, NULL);
1785 1786 1787 1788 1789 1790 1791 1792

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

  return pOperator;

L
Liu Jicong 已提交
1793
_error:
H
Haojun Liao 已提交
1794 1795 1796
  if (pInfo != NULL) {
    destroyIntervalOperatorInfo(pInfo);
  }
1797 1798 1799 1800 1801
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}

1802
// todo handle multiple timeline cases. assume no timeline interweaving
1803 1804
static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperatorInfo* pInfo, SSDataBlock* pBlock) {
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
1805
  SExprSupp*     pSup = &pOperator->exprSupp;
1806

1807
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
1808 1809

  bool    masterScan = true;
1810
  int32_t numOfOutput = pOperator->exprSupp.numOfExprs;
H
Haojun Liao 已提交
1811
  int64_t gid = pBlock->info.id.groupId;
1812 1813 1814 1815 1816 1817 1818 1819 1820 1821 1822 1823 1824 1825

  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) {
1826 1827 1828
    if (gid != pRowSup->groupId || pInfo->winSup.prevTs == INT64_MIN) {
      doKeepNewWindowStartInfo(pRowSup, tsList, j, gid);
      doKeepTuple(pRowSup, tsList[j], gid);
H
Haojun Liao 已提交
1829 1830
    } else if (((tsList[j] - pRowSup->prevTs >= 0) && (tsList[j] - pRowSup->prevTs <= gap)) ||
               ((pRowSup->prevTs - tsList[j] >= 0) && (pRowSup->prevTs - tsList[j] <= gap))) {
1831
      // The gap is less than the threshold, so it belongs to current session window that has been opened already.
1832
      doKeepTuple(pRowSup, tsList[j], gid);
1833 1834 1835 1836 1837 1838 1839 1840 1841 1842
      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;
1843 1844
      int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &window, masterScan, &pResult, gid, pSup->pCtx,
                                           numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1845
      if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
S
Shengliang Guan 已提交
1846
        T_LONG_JMP(pTaskInfo->env, TSDB_CODE_APP_ERROR);
1847 1848 1849 1850
      }

      // pInfo->numOfRows data belong to the current session window
      updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &window, false);
H
Haojun Liao 已提交
1851
      applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
L
Liu Jicong 已提交
1852
                                      pRowSup->numOfRows, pBlock->info.rows, numOfOutput);
1853 1854

      // here we start a new session window
1855 1856
      doKeepNewWindowStartInfo(pRowSup, tsList, j, gid);
      doKeepTuple(pRowSup, tsList[j], gid);
1857 1858 1859 1860 1861
    }
  }

  SResultRow* pResult = NULL;
  pRowSup->win.ekey = tsList[pBlock->info.rows - 1];
1862 1863
  int32_t ret = setTimeWindowOutputBuf(&pInfo->binfo.resultRowInfo, &pRowSup->win, masterScan, &pResult, gid,
                                       pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
1864
  if (ret != TSDB_CODE_SUCCESS) {  // null data, too many state code
S
Shengliang Guan 已提交
1865
    T_LONG_JMP(pTaskInfo->env, TSDB_CODE_APP_ERROR);
1866 1867 1868
  }

  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, false);
L
Liu Jicong 已提交
1869 1870
  applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, pRowSup->startRowIndex,
                                  pRowSup->numOfRows, pBlock->info.rows, numOfOutput);
1871 1872
}

1873
static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
1874 1875 1876 1877 1878 1879
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SSessionAggOperatorInfo* pInfo = pOperator->info;
  SOptrBasicInfo*          pBInfo = &pInfo->binfo;
1880
  SExprSupp*               pSup = &pOperator->exprSupp;
1881 1882

  if (pOperator->status == OP_RES_TO_RETURN) {
1883
    while (1) {
1884
      doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
H
Haojun Liao 已提交
1885
      doFilter(pBInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
1886

1887
      bool hasRemain = hasRemainResults(&pInfo->groupResInfo);
1888
      if (!hasRemain) {
H
Haojun Liao 已提交
1889
        setOperatorCompleted(pOperator);
1890 1891
        break;
      }
1892

1893 1894 1895 1896 1897
      if (pBInfo->pRes->info.rows > 0) {
        break;
      }
    }
    pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1898
    return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1899 1900
  }

1901 1902 1903
  int64_t st = taosGetTimestampUs();
  int32_t order = TSDB_ORDER_ASC;

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

  while (1) {
1907
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
1908 1909 1910 1911 1912
    if (pBlock == NULL) {
      break;
    }

    // the pDataBlock are always the same one, no need to call this again
1913
    setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);
1914 1915
    blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);

1916 1917 1918
    doSessionWindowAggImpl(pOperator, pInfo, pBlock);
  }

1919 1920
  pOperator->cost.openCost = (taosGetTimestampUs() - st) / 1000.0;

1921 1922 1923
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;

1924
  initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, TSDB_ORDER_ASC);
1925
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
1926
  while (1) {
1927
    doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
H
Haojun Liao 已提交
1928
    doFilter(pBInfo->pRes, pOperator->exprSupp.pFilterInfo, NULL);
1929

1930
    bool hasRemain = hasRemainResults(&pInfo->groupResInfo);
1931
    if (!hasRemain) {
H
Haojun Liao 已提交
1932
      setOperatorCompleted(pOperator);
1933 1934
      break;
    }
1935

1936 1937 1938 1939 1940
    if (pBInfo->pRes->info.rows > 0) {
      break;
    }
  }
  pOperator->resultInfo.totalRows += pBInfo->pRes->info.rows;
1941
  return (pBInfo->pRes->info.rows == 0) ? NULL : pBInfo->pRes;
1942 1943
}

H
Haojun Liao 已提交
1944
// todo make this as an non-blocking operator
1945 1946
SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWinodwPhysiNode* pStateNode,
                                             SExecTaskInfo* pTaskInfo) {
1947 1948 1949 1950 1951 1952
  SStateWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStateWindowOperatorInfo));
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

1953 1954 1955
  int32_t      tsSlotId = ((SColumnNode*)pStateNode->window.pTspk)->slotId;
  SColumnNode* pColNode = (SColumnNode*)((STargetNode*)pStateNode->pStateKey)->pExpr;

1956 1957 1958
  if (pStateNode->window.pExprs != NULL) {
    int32_t    numOfScalarExpr = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pStateNode->window.pExprs, NULL, &numOfScalarExpr);
H
Hongze Cheng 已提交
1959
    int32_t    code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr);
1960 1961 1962 1963 1964
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }

1965
  pInfo->stateCol = extractColumnFromColumnNode(pColNode);
1966 1967 1968 1969 1970 1971 1972
  pInfo->stateKey.type = pInfo->stateCol.type;
  pInfo->stateKey.bytes = pInfo->stateCol.bytes;
  pInfo->stateKey.pData = taosMemoryCalloc(1, pInfo->stateCol.bytes);
  if (pInfo->stateKey.pData == NULL) {
    goto _error;
  }

H
Haojun Liao 已提交
1973 1974 1975 1976 1977
  int32_t code = filterInitFromNode((SNode*)pStateNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

1978 1979
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;

1980 1981
  int32_t    num = 0;
  SExprInfo* pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &num);
1982
  initResultSizeInfo(&pOperator->resultInfo, 4096);
H
Haojun Liao 已提交
1983

L
Liu Jicong 已提交
1984 1985
  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,
                    pTaskInfo->streamInfo.pState);
1986 1987 1988 1989
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
1990
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pStateNode->window.node.pOutputDataBlockDesc);
1991
  initBasicInfo(&pInfo->binfo, pResBlock);
1992
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
1993

L
Liu Jicong 已提交
1994 1995
  pInfo->twAggSup =
      (STimeWindowAggSupp){.waterMark = pStateNode->window.watermark, .calTrigger = pStateNode->window.triggerType};
1996

1997 1998
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

X
Xiaoyu Wang 已提交
1999
  pInfo->tsSlotId = tsSlotId;
2000

L
Liu Jicong 已提交
2001 2002
  setOperatorInfo(pOperator, "StateWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE, true, OP_NOT_OPENED, pInfo,
                  pTaskInfo);
L
Liu Jicong 已提交
2003 2004
  pOperator->fpSet = createOperatorFpSet(openStateWindowAggOptr, doStateWindowAgg, NULL, destroyStateWindowOperatorInfo,
                                         optrDefaultBufFn, NULL);
2005

2006 2007 2008 2009 2010
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

2011 2012
  return pOperator;

L
Liu Jicong 已提交
2013
_error:
H
Haojun Liao 已提交
2014 2015 2016 2017
  if (pInfo != NULL) {
    destroyStateWindowOperatorInfo(pInfo);
  }

2018 2019
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
2020 2021 2022
  return NULL;
}

2023
void destroySWindowOperatorInfo(void* param) {
2024
  SSessionAggOperatorInfo* pInfo = (SSessionAggOperatorInfo*)param;
2025 2026 2027
  if (pInfo == NULL) {
    return;
  }
2028

2029
  cleanupBasicInfo(&pInfo->binfo);
H
Haojun Liao 已提交
2030 2031 2032 2033
  colDataDestroy(&pInfo->twAggSup.timeWindowData);

  cleanupAggSup(&pInfo->aggSup);
  cleanupGroupResInfo(&pInfo->groupResInfo);
D
dapan1121 已提交
2034
  taosMemoryFreeClear(param);
2035 2036
}

H
Haojun Liao 已提交
2037
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SSessionWinodwPhysiNode* pSessionNode,
2038
                                            SExecTaskInfo* pTaskInfo) {
2039 2040 2041 2042 2043 2044
  SSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSessionAggOperatorInfo));
  SOperatorInfo*           pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

2045
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
2046
  initResultSizeInfo(&pOperator->resultInfo, 4096);
2047

2048
  int32_t      numOfCols = 0;
H
Haojun Liao 已提交
2049
  SExprInfo*   pExprInfo = createExprInfo(pSessionNode->window.pFuncs, NULL, &numOfCols);
H
Haojun Liao 已提交
2050
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pSessionNode->window.node.pOutputDataBlockDesc);
H
Haojun Liao 已提交
2051
  initBasicInfo(&pInfo->binfo, pResBlock);
H
Haojun Liao 已提交
2052

L
Liu Jicong 已提交
2053 2054
  int32_t code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str,
                            pTaskInfo->streamInfo.pState);
2055 2056 2057 2058
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
2059 2060 2061 2062
  pInfo->twAggSup.waterMark = pSessionNode->window.watermark;
  pInfo->twAggSup.calTrigger = pSessionNode->window.triggerType;
  pInfo->gap = pSessionNode->gap;

2063
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
2064 2065
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

2066
  pInfo->tsSlotId = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
L
Liu Jicong 已提交
2067 2068 2069
  pInfo->binfo.pRes = pResBlock;
  pInfo->winSup.prevTs = INT64_MIN;
  pInfo->reptScan = false;
H
Haojun Liao 已提交
2070 2071 2072 2073
  code = filterInitFromNode((SNode*)pSessionNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
H
Haojun Liao 已提交
2074

L
Liu Jicong 已提交
2075 2076
  setOperatorInfo(pOperator, "SessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION, true, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
L
Liu Jicong 已提交
2077 2078
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doSessionWindowAgg, NULL, destroySWindowOperatorInfo,
                                         optrDefaultBufFn, NULL);
2079 2080
  pOperator->pTaskInfo = pTaskInfo;
  code = appendDownstream(pOperator, &downstream, 1);
2081 2082 2083 2084
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

2085 2086
  return pOperator;

L
Liu Jicong 已提交
2087
_error:
2088
  destroySWindowOperatorInfo(pInfo);
2089 2090 2091
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
L
Liu Jicong 已提交
2092
}
5
54liuyao 已提交
2093

5
54liuyao 已提交
2094
void compactFunctions(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx, int32_t numOfOutput,
2095
                      SExecTaskInfo* pTaskInfo, SColumnInfoData* pTimeWindowData) {
5
54liuyao 已提交
2096 2097
  for (int32_t k = 0; k < numOfOutput; ++k) {
    if (fmIsWindowPseudoColumnFunc(pDestCtx[k].functionId)) {
2098 2099 2100 2101 2102
      if (!pTimeWindowData) {
        continue;
      }

      SResultRowEntryInfo* pEntryInfo = GET_RES_INFO(&pDestCtx[k]);
L
Liu Jicong 已提交
2103 2104
      char*                p = GET_ROWCELL_INTERBUF(pEntryInfo);
      SColumnInfoData      idata = {0};
2105 2106 2107 2108 2109 2110 2111 2112
      idata.info.type = TSDB_DATA_TYPE_BIGINT;
      idata.info.bytes = tDataTypes[TSDB_DATA_TYPE_BIGINT].bytes;
      idata.pData = p;

      SScalarParam out = {.columnData = &idata};
      SScalarParam tw = {.numOfRows = 5, .columnData = pTimeWindowData};
      pDestCtx[k].sfp.process(&tw, 1, &out);
      pEntryInfo->numOfRes = 1;
L
Liu Jicong 已提交
2113
    } else if (functionNeedToExecute(&pDestCtx[k]) && pDestCtx[k].fpSet.combine != NULL) {
2114
      int32_t code = pDestCtx[k].fpSet.combine(&pDestCtx[k], &pSourceCtx[k]);
5
54liuyao 已提交
2115
      if (code != TSDB_CODE_SUCCESS) {
5
54liuyao 已提交
2116
        qError("%s apply combine functions error, code: %s", GET_TASKID(pTaskInfo), tstrerror(code));
5
54liuyao 已提交
2117
      }
5
54liuyao 已提交
2118 2119 2120 2121
    } else if (pDestCtx[k].fpSet.combine == NULL) {
      char* funName = fmGetFuncName(pDestCtx[k].functionId);
      qError("%s error, combine funcion for %s is not implemented", GET_TASKID(pTaskInfo), funName);
      taosMemoryFreeClear(funName);
5
54liuyao 已提交
2122 2123 2124 2125
    }
  }
}

2126 2127
bool hasIntervalWindow(SStreamState* pState, SWinKey* pKey) {
  return TSDB_CODE_SUCCESS == streamStateGet(pState, pKey, NULL, 0);
2128 2129
}

2130
static void rebuildIntervalWindow(SOperatorInfo* pOperator, SArray* pWinArray, SSHashObj* pUpdatedMap) {
2131 2132 2133 2134
  SStreamIntervalOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
  int32_t                      size = taosArrayGetSize(pWinArray);
  int32_t                      numOfOutput = pOperator->exprSupp.numOfExprs;
5
54liuyao 已提交
2135
  SExprSupp*                   pSup = &pOperator->exprSupp;
5
54liuyao 已提交
2136 2137 2138
  if (!pInfo->pChildren) {
    return;
  }
5
54liuyao 已提交
2139
  for (int32_t i = 0; i < size; i++) {
H
Haojun Liao 已提交
2140
    SWinKey*    pWinRes = taosArrayGet(pWinArray, i);
2141
    SResultRow* pCurResult = NULL;
2142
    STimeWindow parentWin = getFinalTimeWindow(pWinRes->ts, &pInfo->interval);
5
54liuyao 已提交
2143
    if (isDeletedStreamWindow(&parentWin, pWinRes->groupId, pInfo->pState, &pInfo->twAggSup)) {
2144 2145
      continue;
    }
2146

5
54liuyao 已提交
2147
    int32_t numOfChildren = taosArrayGetSize(pInfo->pChildren);
2148
    int32_t num = 0;
5
54liuyao 已提交
2149
    for (int32_t j = 0; j < numOfChildren; j++) {
2150 2151 2152 2153
      SOperatorInfo*               pChildOp = taosArrayGetP(pInfo->pChildren, j);
      SStreamIntervalOperatorInfo* pChInfo = pChildOp->info;
      SExprSupp*                   pChildSup = &pChildOp->exprSupp;
      if (!hasIntervalWindow(pChInfo->pState, pWinRes)) {
2154 2155
        continue;
      }
2156 2157 2158 2159
      if (num == 0) {
        int32_t code = setOutputBuf(pInfo->pState, &parentWin, &pCurResult, pWinRes->groupId, pSup->pCtx, numOfOutput,
                                    pSup->rowEntryInfoOffset, &pInfo->aggSup);
        if (code != TSDB_CODE_SUCCESS || pCurResult == NULL) {
S
Shengliang Guan 已提交
2160
          T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
2161 2162
        }
      }
2163
      num++;
2164
      SResultRow* pChResult = NULL;
2165 2166
      setOutputBuf(pChInfo->pState, &parentWin, &pChResult, pWinRes->groupId, pChildSup->pCtx, pChildSup->numOfExprs,
                   pChildSup->rowEntryInfoOffset, &pChInfo->aggSup);
2167 2168
      updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &parentWin, true);
      compactFunctions(pSup->pCtx, pChildSup->pCtx, numOfOutput, pTaskInfo, &pInfo->twAggSup.timeWindowData);
5
54liuyao 已提交
2169
      releaseOutputBuf(pChInfo->pState, pWinRes, pChResult);
5
54liuyao 已提交
2170
    }
2171
    if (num > 0 && pUpdatedMap) {
2172 2173 2174
      saveWinResultInfo(pCurResult->win.skey, pWinRes->groupId, pUpdatedMap);
      saveOutputBuf(pInfo->pState, pWinRes, pCurResult, pInfo->aggSup.resultRowSize);
      releaseOutputBuf(pInfo->pState, pWinRes, pCurResult);
2175
    }
5
54liuyao 已提交
2176 2177 2178
  }
}

2179
bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup) {
5
54liuyao 已提交
2180 2181
  if (pWin->ekey < pTwSup->maxTs - pTwSup->deleteMark) {
    SWinKey key = {.ts = pWin->skey, .groupId = groupId};
5
54liuyao 已提交
2182
    if (streamStateGet(pState, &key, NULL, 0) == TSDB_CODE_SUCCESS) {
5
54liuyao 已提交
2183 2184
      return false;
    }
2185
    return true;
5
54liuyao 已提交
2186 2187 2188 2189
  }
  return false;
}

L
Liu Jicong 已提交
2190 2191 2192 2193
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 已提交
2194 2195 2196 2197
  int32_t prevEndPos = forwardRows - 1 + startPos;
  return getNextQualifiedWindow(pInterval, pNextWin, pBlockInfo, tsCols, prevEndPos, TSDB_ORDER_ASC);
}

H
Haojun Liao 已提交
2198
void addPullWindow(SHashObj* pMap, SWinKey* pWinRes, int32_t size) {
5
54liuyao 已提交
2199 2200 2201 2202
  SArray* childIds = taosArrayInit(8, sizeof(int32_t));
  for (int32_t i = 0; i < size; i++) {
    taosArrayPush(childIds, &i);
  }
H
Haojun Liao 已提交
2203
  taosHashPut(pMap, pWinRes, sizeof(SWinKey), &childIds, sizeof(void*));
5
54liuyao 已提交
2204 2205 2206 2207
}

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

2208
static void clearStreamIntervalOperator(SStreamIntervalOperatorInfo* pInfo) {
2209
  tSimpleHashClear(pInfo->aggSup.pResultRowHashTable);
5
54liuyao 已提交
2210
  clearDiskbasedBuf(pInfo->aggSup.pResultBuf);
2211
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
L
Liu Jicong 已提交
2212
  pInfo->aggSup.currentPageId = -1;
2213
  streamStateClear(pInfo->pState);
5
54liuyao 已提交
2214 2215
}

5
54liuyao 已提交
2216 2217 2218 2219
static void clearSpecialDataBlock(SSDataBlock* pBlock) {
  if (pBlock->info.rows <= 0) {
    return;
  }
5
54liuyao 已提交
2220 2221 2222
  blockDataCleanup(pBlock);
}

5
54liuyao 已提交
2223 2224 2225 2226 2227 2228
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 已提交
2229
  blockDataEnsureCapacity(pBlock, size - (*pIndex));
2230 2231 2232 2233 2234
  SColumnInfoData* pStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
  SColumnInfoData* pEndTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
  SColumnInfoData* pGroupId = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
  SColumnInfoData* pCalStartTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
  SColumnInfoData* pCalEndTs = (SColumnInfoData*)taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
5
54liuyao 已提交
2235
  for (; (*pIndex) < size; (*pIndex)++) {
L
Liu Jicong 已提交
2236
    SPullWindowInfo* pWin = taosArrayGet(array, (*pIndex));
2237 2238 2239 2240 2241
    colDataSetVal(pStartTs, pBlock->info.rows, (const char*)&pWin->window.skey, false);
    colDataSetVal(pEndTs, pBlock->info.rows, (const char*)&pWin->window.ekey, false);
    colDataSetVal(pGroupId, pBlock->info.rows, (const char*)&pWin->groupId, false);
    colDataSetVal(pCalStartTs, pBlock->info.rows, (const char*)&pWin->calWin.skey, false);
    colDataSetVal(pCalEndTs, pBlock->info.rows, (const char*)&pWin->calWin.ekey, false);
5
54liuyao 已提交
2242 2243 2244 2245 2246 2247 2248 2249 2250
    pBlock->info.rows++;
  }
  if ((*pIndex) == size) {
    *pIndex = 0;
    taosArrayClear(array);
  }
  blockDataUpdateTsWindow(pBlock, 0);
}

5
54liuyao 已提交
2251
void processPullOver(SSDataBlock* pBlock, SHashObj* pMap, SInterval* pInterval) {
5
54liuyao 已提交
2252
  SColumnInfoData* pStartCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
L
Liu Jicong 已提交
2253
  TSKEY*           tsData = (TSKEY*)pStartCol->pData;
5
54liuyao 已提交
2254 2255
  SColumnInfoData* pEndCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
  TSKEY*           tsEndData = (TSKEY*)pEndCol->pData;
5
54liuyao 已提交
2256
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
L
Liu Jicong 已提交
2257 2258
  uint64_t*        groupIdData = (uint64_t*)pGroupCol->pData;
  int32_t          chId = getChildIndex(pBlock);
5
54liuyao 已提交
2259
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
5
54liuyao 已提交
2260 2261 2262 2263 2264 2265 2266 2267 2268 2269 2270 2271 2272 2273 2274
    TSKEY winTs = tsData[i];
    while (winTs < tsEndData[i]) {
      SWinKey winRes = {.ts = winTs, .groupId = groupIdData[i]};
      void*   chIds = taosHashGet(pMap, &winRes, sizeof(SWinKey));
      if (chIds) {
        SArray* chArray = *(SArray**)chIds;
        int32_t index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
        if (index != -1) {
          qDebug("===stream===window %" PRId64 " delete child id %d", winRes.ts, chId);
          taosArrayRemove(chArray, index);
          if (taosArrayGetSize(chArray) == 0) {
            // pull data is over
            taosArrayDestroy(chArray);
            taosHashRemove(pMap, &winRes, sizeof(SWinKey));
          }
5
54liuyao 已提交
2275 2276
        }
      }
5
54liuyao 已提交
2277
      winTs = taosTimeAdd(winTs, pInterval->sliding, pInterval->slidingUnit, pInterval->precision);
5
54liuyao 已提交
2278 2279 2280
    }
  }
}
5
54liuyao 已提交
2281

2282
static void addRetriveWindow(SArray* wins, SStreamIntervalOperatorInfo* pInfo) {
2283 2284
  int32_t size = taosArrayGetSize(wins);
  for (int32_t i = 0; i < size; i++) {
L
Liu Jicong 已提交
2285
    SWinKey*    winKey = taosArrayGet(wins, i);
2286
    STimeWindow nextWin = getFinalTimeWindow(winKey->ts, &pInfo->interval);
2287
    if (needDeleteWindowBuf(&nextWin, &pInfo->twAggSup) && !pInfo->ignoreExpiredData) {
2288 2289
      void* chIds = taosHashGet(pInfo->pPullDataMap, winKey, sizeof(SWinKey));
      if (!chIds) {
L
Liu Jicong 已提交
2290 2291
        SPullWindowInfo pull = {
            .window = nextWin, .groupId = winKey->groupId, .calWin.skey = nextWin.skey, .calWin.ekey = nextWin.skey};
2292
        // add pull data request
5
54liuyao 已提交
2293 2294 2295 2296 2297
        if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) {
          int32_t size1 = taosArrayGetSize(pInfo->pChildren);
          addPullWindow(pInfo->pPullDataMap, winKey, size1);
          qDebug("===stream===prepare retrive for delete %" PRId64 ", size:%d", winKey->ts, size1);
        }
2298 2299 2300 2301 2302
      }
    }
  }
}

5
54liuyao 已提交
2303 2304 2305 2306 2307 2308
static void clearFunctionContext(SExprSupp* pSup) {
  for (int32_t i = 0; i < pSup->numOfExprs; i++) {
    pSup->pCtx[i].saveHandle.currentPage = -1;
  }
}

2309 2310
void doBuildStreamIntervalResult(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* pBlock,
                                 SGroupResInfo* pGroupResInfo) {
2311 2312 2313 2314 2315 2316 2317 2318 2319 2320
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
  // set output datablock version
  pBlock->info.version = pTaskInfo->version;

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

  // clear the existed group id
H
Haojun Liao 已提交
2321
  pBlock->info.id.groupId = 0;
2322
  buildDataBlockFromGroupRes(pOperator, pState, pBlock, &pOperator->exprSupp, pGroupResInfo);
2323 2324
}

5
54liuyao 已提交
2325 2326 2327 2328 2329 2330 2331 2332 2333 2334 2335
static int32_t getNextQualifiedFinalWindow(SInterval* pInterval, STimeWindow* pNext, SDataBlockInfo* pDataBlockInfo,
                                           TSKEY* primaryKeys, int32_t prevPosition) {
  int32_t startPos = prevPosition + 1;
  if (startPos == pDataBlockInfo->rows) {
    startPos = -1;
  } else {
    *pNext = getFinalTimeWindow(primaryKeys[startPos], pInterval);
  }
  return startPos;
}

2336
static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBlock, uint64_t groupId,
2337
                                    SSHashObj* pUpdatedMap) {
2338 2339 2340 2341 2342 2343 2344 2345 2346 2347 2348 2349 2350 2351 2352 2353
  SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)pOperatorInfo->info;

  SResultRowInfo* pResultRowInfo = &(pInfo->binfo.resultRowInfo);
  SExecTaskInfo*  pTaskInfo = pOperatorInfo->pTaskInfo;
  SExprSupp*      pSup = &pOperatorInfo->exprSupp;
  int32_t         numOfOutput = pSup->numOfExprs;
  int32_t         step = 1;
  TSKEY*          tsCols = NULL;
  SResultRow*     pResult = NULL;
  int32_t         forwardRows = 0;

  SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
  tsCols = (int64_t*)pColDataInfo->pData;

  int32_t     startPos = 0;
  TSKEY       ts = getStartTsKey(&pSDataBlock->info.window, tsCols);
5
54liuyao 已提交
2354 2355 2356 2357 2358 2359
  STimeWindow nextWin = {0};
  if (IS_FINAL_OP(pInfo)) {
    nextWin = getFinalTimeWindow(ts, &pInfo->interval);
  } else {
    nextWin = getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, TSDB_ORDER_ASC);
  }
2360 2361 2362 2363 2364 2365 2366 2367 2368 2369 2370 2371 2372 2373 2374 2375 2376 2377
  while (1) {
    bool isClosed = isCloseWindow(&nextWin, &pInfo->twAggSup);
    if ((pInfo->ignoreExpiredData && isClosed) || !inSlidingWindow(&pInfo->interval, &nextWin, &pSDataBlock->info)) {
      startPos = getNexWindowPos(&pInfo->interval, &pSDataBlock->info, tsCols, startPos, nextWin.ekey, &nextWin);
      if (startPos < 0) {
        break;
      }
      continue;
    }

    if (IS_FINAL_OP(pInfo) && isClosed && pInfo->pChildren) {
      bool    ignore = true;
      SWinKey winRes = {
          .ts = nextWin.skey,
          .groupId = groupId,
      };
      void* chIds = taosHashGet(pInfo->pPullDataMap, &winRes, sizeof(SWinKey));
      if (isDeletedStreamWindow(&nextWin, groupId, pInfo->pState, &pInfo->twAggSup) && !chIds) {
L
Liu Jicong 已提交
2378 2379
        SPullWindowInfo pull = {
            .window = nextWin, .groupId = groupId, .calWin.skey = nextWin.skey, .calWin.ekey = nextWin.skey};
2380
        // add pull data request
5
54liuyao 已提交
2381 2382 2383 2384 2385
        if (savePullWindow(&pull, pInfo->pPullWins) == TSDB_CODE_SUCCESS) {
          int32_t size = taosArrayGetSize(pInfo->pChildren);
          addPullWindow(pInfo->pPullDataMap, &winRes, size);
          qDebug("===stream===prepare retrive %" PRId64 ", size:%d", winRes.ts, size);
        }
2386 2387 2388 2389 2390 2391 2392 2393 2394 2395 2396 2397 2398 2399 2400 2401 2402 2403 2404 2405 2406 2407 2408 2409 2410 2411
      } else {
        int32_t index = -1;
        SArray* chArray = NULL;
        int32_t chId = 0;
        if (chIds) {
          chArray = *(void**)chIds;
          chId = getChildIndex(pSDataBlock);
          index = taosArraySearchIdx(chArray, &chId, compareInt32Val, TD_EQ);
        }
        if (index == -1 || pSDataBlock->info.type == STREAM_PULL_DATA) {
          ignore = false;
        }
      }

      if (ignore) {
        startPos = getNexWindowPos(&pInfo->interval, &pSDataBlock->info, tsCols, startPos, nextWin.ekey, &nextWin);
        if (startPos < 0) {
          break;
        }
        continue;
      }
    }

    int32_t code = setOutputBuf(pInfo->pState, &nextWin, &pResult, groupId, pSup->pCtx, numOfOutput,
                                pSup->rowEntryInfoOffset, &pInfo->aggSup);
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
S
Shengliang Guan 已提交
2412
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
2413 2414
    }

5
54liuyao 已提交
2415 2416 2417 2418 2419 2420
    if (IS_FINAL_OP(pInfo)) {
      forwardRows = 1;
    } else {
      forwardRows = getNumOfRowsInTimeWindow(&pSDataBlock->info, tsCols, startPos, nextWin.ekey, binarySearchForKey,
                                             NULL, TSDB_ORDER_ASC);
    }
2421 2422 2423
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pUpdatedMap) {
      saveWinResultInfo(pResult->win.skey, groupId, pUpdatedMap);
    }
5
54liuyao 已提交
2424 2425 2426 2427 2428 2429 2430 2431

    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
      SWinKey key = {
          .ts = pResult->win.skey,
          .groupId = groupId,
      };
      tSimpleHashPut(pInfo->aggSup.pResultRowHashTable, &key, sizeof(SWinKey), NULL, 0);
    }
2432
    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true);
H
Haojun Liao 已提交
2433
    applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, forwardRows,
L
Liu Jicong 已提交
2434
                                    pSDataBlock->info.rows, numOfOutput);
2435 2436 2437 2438 2439 2440 2441 2442 2443 2444
    SWinKey key = {
        .ts = nextWin.skey,
        .groupId = groupId,
    };
    saveOutputBuf(pInfo->pState, &key, pResult, pInfo->aggSup.resultRowSize);
    releaseOutputBuf(pInfo->pState, &key, pResult);
    if (pInfo->delKey.ts > key.ts) {
      pInfo->delKey = key;
    }
    int32_t prevEndPos = (forwardRows - 1) * step + startPos;
2445 2446 2447 2448 2449 2450 2451 2452 2453 2454 2455 2456 2457
    if (pSDataBlock->info.window.skey <= 0 || pSDataBlock->info.window.ekey <= 0) {
      qError("table uid %" PRIu64 " data block timestamp range may not be calculated! minKey %" PRId64
             ",maxKey %" PRId64,
             pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
      blockDataUpdateTsWindow(pSDataBlock, 0);

      // timestamp of the data is incorrect
      if (pSDataBlock->info.window.skey <= 0 || pSDataBlock->info.window.ekey <= 0) {
        qError("table uid %" PRIu64 " data block timestamp is out of range! minKey %" PRId64 ",maxKey %" PRId64,
               pSDataBlock->info.id.uid, pSDataBlock->info.window.skey, pSDataBlock->info.window.ekey);
      }
    }

5
54liuyao 已提交
2458 2459 2460 2461 2462 2463
    if (IS_FINAL_OP(pInfo)) {
      startPos = getNextQualifiedFinalWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos);
    } else {
      startPos =
          getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC);
    }
2464 2465 2466 2467 2468 2469
    if (startPos < 0) {
      break;
    }
  }
}

5
54liuyao 已提交
2470
static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
2471
  SStreamIntervalOperatorInfo* pInfo = pOperator->info;
L
Liu Jicong 已提交
2472
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
2473 2474
  SOperatorInfo*               downstream = pOperator->pDownstream[0];
  SExprSupp*                   pSup = &pOperator->exprSupp;
2475

5
54liuyao 已提交
2476
  qDebug("interval status %d %s", pOperator->status, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
L
Liu Jicong 已提交
2477

5
54liuyao 已提交
2478 2479 2480
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
5
54liuyao 已提交
2481 2482 2483
    doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
    if (pInfo->pPullDataRes->info.rows != 0) {
      // process the rest of the data
5
54liuyao 已提交
2484
      printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
5
54liuyao 已提交
2485 2486 2487
      return pInfo->pPullDataRes;
    }

2488
    doBuildDeleteResult(pInfo, pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
2489 2490 2491 2492 2493 2494
    if (pInfo->pDelRes->info.rows != 0) {
      // process the rest of the data
      printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
      return pInfo->pDelRes;
    }

2495
    doBuildStreamIntervalResult(pOperator, pInfo->pState, pInfo->binfo.pRes, &pInfo->groupResInfo);
5
54liuyao 已提交
2496 2497 2498
    if (pInfo->binfo.pRes->info.rows != 0) {
      printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
      return pInfo->binfo.pRes;
5
54liuyao 已提交
2499
    }
5
54liuyao 已提交
2500

H
Haojun Liao 已提交
2501
    setOperatorCompleted(pOperator);
5
54liuyao 已提交
2502 2503 2504 2505 2506 2507
    if (!IS_FINAL_OP(pInfo)) {
      clearFunctionContext(&pOperator->exprSupp);
      // semi interval operator clear disk buffer
      clearStreamIntervalOperator(pInfo);
      qDebug("===stream===clear semi operator");
    } else {
2508 2509
      deleteIntervalDiscBuf(pInfo->pState, pInfo->pPullDataMap, pInfo->twAggSup.maxTs - pInfo->twAggSup.deleteMark,
                            &pInfo->interval, &pInfo->delKey);
L
Liu Jicong 已提交
2510
      streamStateCommit(pTaskInfo->streamInfo.pState);
5
54liuyao 已提交
2511 2512
    }
    return NULL;
5
54liuyao 已提交
2513
  } else {
5
54liuyao 已提交
2514
    if (!IS_FINAL_OP(pInfo)) {
2515
      doBuildDeleteResult(pInfo, pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
2516 2517 2518 2519 2520 2521
      if (pInfo->pDelRes->info.rows != 0) {
        // process the rest of the data
        printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
        return pInfo->pDelRes;
      }

2522
      doBuildStreamIntervalResult(pOperator, pInfo->pState, pInfo->binfo.pRes, &pInfo->groupResInfo);
5
54liuyao 已提交
2523
      if (pInfo->binfo.pRes->info.rows != 0) {
5
54liuyao 已提交
2524
        printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
5
54liuyao 已提交
2525 2526
        return pInfo->binfo.pRes;
      }
5
54liuyao 已提交
2527
    }
5
54liuyao 已提交
2528 2529
  }

5
54liuyao 已提交
2530
  if (!pInfo->pUpdated) {
2531
    pInfo->pUpdated = taosArrayInit(4, sizeof(SWinKey));
5
54liuyao 已提交
2532 2533 2534
  }
  if (!pInfo->pUpdatedMap) {
    _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
2535
    pInfo->pUpdatedMap = tSimpleHashInit(1024, hashFn);
5
54liuyao 已提交
2536 2537
  }

5
54liuyao 已提交
2538 2539 2540
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
5
54liuyao 已提交
2541
      pOperator->status = OP_RES_TO_RETURN;
L
Liu Jicong 已提交
2542 2543
      qDebug("===stream===return data:%s. recv datablock num:%" PRIu64,
             IS_FINAL_OP(pInfo) ? "interval final" : "interval semi", pInfo->numOfDatapack);
5
54liuyao 已提交
2544
      pInfo->numOfDatapack = 0;
5
54liuyao 已提交
2545 2546
      break;
    }
5
54liuyao 已提交
2547
    pInfo->numOfDatapack++;
5
54liuyao 已提交
2548
    printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "interval final recv" : "interval semi recv");
2549

H
Haojun Liao 已提交
2550
    if (pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_PULL_DATA) {
5
54liuyao 已提交
2551
      pInfo->binfo.pRes->info.type = pBlock->info.type;
2552 2553
    } else if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
               pBlock->info.type == STREAM_CLEAR) {
2554
      SArray* delWins = taosArrayInit(8, sizeof(SWinKey));
5
54liuyao 已提交
2555
      doDeleteWindows(pOperator, &pInfo->interval, pBlock, delWins, pInfo->pUpdatedMap);
2556
      if (IS_FINAL_OP(pInfo)) {
2557 2558 2559 2560
        int32_t                      childIndex = getChildIndex(pBlock);
        SOperatorInfo*               pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
        SStreamIntervalOperatorInfo* pChildInfo = pChildOp->info;
        SExprSupp*                   pChildSup = &pChildOp->exprSupp;
2561
        doDeleteWindows(pChildOp, &pChildInfo->interval, pBlock, NULL, NULL);
5
54liuyao 已提交
2562
        rebuildIntervalWindow(pOperator, delWins, pInfo->pUpdatedMap);
2563 2564 2565
        addRetriveWindow(delWins, pInfo);
        taosArrayAddAll(pInfo->pDelWins, delWins);
        taosArrayDestroy(delWins);
2566 2567
        continue;
      }
5
54liuyao 已提交
2568
      removeResults(delWins, pInfo->pUpdatedMap);
2569 2570
      taosArrayAddAll(pInfo->pDelWins, delWins);
      taosArrayDestroy(delWins);
2571
      break;
5
54liuyao 已提交
2572
    } else if (pBlock->info.type == STREAM_GET_ALL && IS_FINAL_OP(pInfo)) {
5
54liuyao 已提交
2573
      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap);
5
54liuyao 已提交
2574
      continue;
5
54liuyao 已提交
2575
    } else if (pBlock->info.type == STREAM_RETRIEVE && !IS_FINAL_OP(pInfo)) {
5
54liuyao 已提交
2576 2577
      doDeleteWindows(pOperator, &pInfo->interval, pBlock, NULL, pInfo->pUpdatedMap);
      if (taosArrayGetSize(pInfo->pUpdated) > 0) {
5
54liuyao 已提交
2578 2579 2580
        break;
      }
      continue;
L
Liu Jicong 已提交
2581
    } else if (pBlock->info.type == STREAM_PULL_OVER && IS_FINAL_OP(pInfo)) {
5
54liuyao 已提交
2582
      processPullOver(pBlock, pInfo->pPullDataMap, &pInfo->interval);
5
54liuyao 已提交
2583
      continue;
5
54liuyao 已提交
2584 2585 2586 2587
    } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
      return pBlock;
    } else {
      ASSERTS(pBlock->info.type == STREAM_INVALID, "invalid SSDataBlock type");
5
54liuyao 已提交
2588
    }
5
54liuyao 已提交
2589

5
54liuyao 已提交
2590 2591 2592 2593
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
2594
    setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
2595
    doStreamIntervalAggImpl(pOperator, pBlock, pBlock->info.id.groupId, pInfo->pUpdatedMap);
5
54liuyao 已提交
2596
    if (IS_FINAL_OP(pInfo)) {
S
shenglian zhou 已提交
2597
      int32_t chIndex = getChildIndex(pBlock);
5
54liuyao 已提交
2598 2599 2600 2601 2602
      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) {
S
Shengliang Guan 已提交
2603
          T_LONG_JMP(pOperator->pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
5
54liuyao 已提交
2604
        }
2605
        SStreamIntervalOperatorInfo* pTmpInfo = pChildOp->info;
2606
        pTmpInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
5
54liuyao 已提交
2607
        taosArrayPush(pInfo->pChildren, &pChildOp);
5
54liuyao 已提交
2608
        qDebug("===stream===add child, id:%d", chIndex);
5
54liuyao 已提交
2609
      }
2610 2611
      SOperatorInfo*               pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
      SStreamIntervalOperatorInfo* pChInfo = pChildOp->info;
2612
      setInputDataBlock(&pChildOp->exprSupp, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
H
Haojun Liao 已提交
2613
      doStreamIntervalAggImpl(pChildOp, pBlock, pBlock->info.id.groupId, NULL);
5
54liuyao 已提交
2614
    }
5
54liuyao 已提交
2615 2616 2617
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.watermark);
    pInfo->twAggSup.minTs = TMIN(pInfo->twAggSup.minTs, pBlock->info.window.skey);
5
54liuyao 已提交
2618
  }
S
shenglian zhou 已提交
2619

5
54liuyao 已提交
2620
  removeDeleteResults(pInfo->pUpdatedMap, pInfo->pDelWins);
5
54liuyao 已提交
2621
  if (IS_FINAL_OP(pInfo)) {
2622
    closeStreamIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval,
5
54liuyao 已提交
2623
                              pInfo->pPullDataMap, pInfo->pUpdatedMap, pInfo->pDelWins, pOperator);
2624
    closeChildIntervalWindow(pOperator, pInfo->pChildren, pInfo->twAggSup.maxTs);
5
54liuyao 已提交
2625
  }
2626
  pInfo->binfo.pRes->info.watermark = pInfo->twAggSup.maxTs;
5
54liuyao 已提交
2627

2628 2629 2630
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pInfo->pUpdatedMap, pIte, &iter)) != NULL) {
5
54liuyao 已提交
2631
    taosArrayPush(pInfo->pUpdated, pIte);
5
54liuyao 已提交
2632
  }
2633
  tSimpleHashCleanup(pInfo->pUpdatedMap);
5
54liuyao 已提交
2634
  pInfo->pUpdatedMap = NULL;
2635
  taosArraySort(pInfo->pUpdated, winKeyCmprImpl);
5
54liuyao 已提交
2636

5
54liuyao 已提交
2637 2638
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);
  pInfo->pUpdated = NULL;
5
54liuyao 已提交
2639
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
5
54liuyao 已提交
2640 2641 2642 2643

  doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
  if (pInfo->pPullDataRes->info.rows != 0) {
    // process the rest of the data
5
54liuyao 已提交
2644
    printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
5
54liuyao 已提交
2645 2646 2647
    return pInfo->pPullDataRes;
  }

2648
  doBuildDeleteResult(pInfo, pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
5
54liuyao 已提交
2649 2650 2651 2652 2653 2654
  if (pInfo->pDelRes->info.rows != 0) {
    // process the rest of the data
    printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
    return pInfo->pDelRes;
  }

2655
  doBuildStreamIntervalResult(pOperator, pInfo->pState, pInfo->binfo.pRes, &pInfo->groupResInfo);
5
54liuyao 已提交
2656
  if (pInfo->binfo.pRes->info.rows != 0) {
5
54liuyao 已提交
2657
    printDataBlock(pInfo->binfo.pRes, IS_FINAL_OP(pInfo) ? "interval final" : "interval semi");
5
54liuyao 已提交
2658 2659 2660 2661 2662 2663
    return pInfo->binfo.pRes;
  }

  return NULL;
}

5
54liuyao 已提交
2664 2665 2666 2667
int64_t getDeleteMark(SIntervalPhysiNode* pIntervalPhyNode) {
  if (pIntervalPhyNode->window.deleteMark <= 0) {
    return DEAULT_DELETE_MARK;
  }
L
Liu Jicong 已提交
2668
  int64_t deleteMark = TMAX(pIntervalPhyNode->window.deleteMark, pIntervalPhyNode->window.watermark);
5
54liuyao 已提交
2669 2670 2671 2672
  deleteMark = TMAX(deleteMark, pIntervalPhyNode->interval);
  return deleteMark;
}

S
shenglian zhou 已提交
2673 2674
SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                     SExecTaskInfo* pTaskInfo, int32_t numOfChild) {
2675 2676 2677
  SIntervalPhysiNode*          pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
  SStreamIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamIntervalOperatorInfo));
  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
2678 2679 2680
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
2681

2682
  pOperator->pTaskInfo = pTaskInfo;
S
shenglian zhou 已提交
2683 2684 2685 2686 2687 2688 2689 2690
  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 已提交
2691 2692
      .calTrigger = pIntervalPhyNode->window.triggerType,
      .maxTs = INT64_MIN,
5
54liuyao 已提交
2693
      .minTs = INT64_MAX,
5
54liuyao 已提交
2694
      .deleteMark = getDeleteMark(pIntervalPhyNode),
L
Liu Jicong 已提交
2695 2696
      .deleteMarkSaved = 0,
      .calTriggerSaved = 0,
S
shenglian zhou 已提交
2697
  };
5
54liuyao 已提交
2698
  ASSERTS(pInfo->twAggSup.calTrigger != STREAM_TRIGGER_MAX_DELAY, "trigger type should not be max delay");
5
54liuyao 已提交
2699 2700
  pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
2701
  initResultSizeInfo(&pOperator->resultInfo, 4096);
5
54liuyao 已提交
2702 2703 2704 2705 2706 2707 2708 2709 2710
  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 已提交
2711 2712
  int32_t      numOfCols = 0;
  SExprInfo*   pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
H
Haojun Liao 已提交
2713
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);
5
54liuyao 已提交
2714
  initBasicInfo(&pInfo->binfo, pResBlock);
2715

L
Liu Jicong 已提交
2716 2717
  int32_t code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str,
                            pTaskInfo->streamInfo.pState);
2718 2719 2720 2721
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
2722
  initStreamFunciton(pOperator->exprSupp.pCtx, pOperator->exprSupp.numOfExprs);
2723

5
54liuyao 已提交
2724
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
2725

2726 2727 2728 2729
  pInfo->pState = taosMemoryCalloc(1, sizeof(SStreamState));
  *(pInfo->pState) = *(pTaskInfo->streamInfo.pState);
  streamStateSetNumber(pInfo->pState, -1);

2730
  initResultRowInfo(&pInfo->binfo.resultRowInfo);
5
54liuyao 已提交
2731 2732
  pInfo->pChildren = NULL;
  if (numOfChild > 0) {
2733
    pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
5
54liuyao 已提交
2734 2735 2736
    if (!pInfo->pChildren) {
      goto _error;
    }
5
54liuyao 已提交
2737 2738 2739
    for (int32_t i = 0; i < numOfChild; i++) {
      SOperatorInfo* pChildOp = createStreamFinalIntervalOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
      if (pChildOp) {
2740
        SStreamIntervalOperatorInfo* pChInfo = pChildOp->info;
2741
        pChInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
5
54liuyao 已提交
2742
        taosArrayPush(pInfo->pChildren, &pChildOp);
2743
        streamStateSetNumber(pChInfo->pState, i);
5
54liuyao 已提交
2744 2745 2746 2747 2748
        continue;
      }
      goto _error;
    }
  }
5
54liuyao 已提交
2749

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

5
54liuyao 已提交
2752 2753 2754 2755
  if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL) {
    pInfo->isFinal = true;
    pOperator->name = "StreamFinalIntervalOperator";
  } else {
5
54liuyao 已提交
2756
    // semi interval operator does not catch result
5
54liuyao 已提交
2757 2758 2759 2760
    pInfo->isFinal = false;
    pOperator->name = "StreamSemiIntervalOperator";
  }

5
54liuyao 已提交
2761
  if (!IS_FINAL_OP(pInfo) || numOfChild == 0) {
5
54liuyao 已提交
2762 2763
    pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
  }
5
54liuyao 已提交
2764 2765 2766 2767
  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);
2768
  pInfo->pPullDataRes = createSpecialDataBlock(STREAM_RETRIEVE);
5
54liuyao 已提交
2769
  pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired;
2770
  pInfo->ignoreExpiredDataSaved = false;
2771
  pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT);
2772
  pInfo->delIndex = 0;
H
Haojun Liao 已提交
2773
  pInfo->pDelWins = taosArrayInit(4, sizeof(SWinKey));
2774 2775
  pInfo->delKey.ts = INT64_MAX;
  pInfo->delKey.groupId = 0;
5
54liuyao 已提交
2776
  pInfo->numOfDatapack = 0;
5
54liuyao 已提交
2777 2778
  pInfo->pUpdated = NULL;
  pInfo->pUpdatedMap = NULL;
5
54liuyao 已提交
2779

5
54liuyao 已提交
2780
  pOperator->operatorType = pPhyNode->type;
5
54liuyao 已提交
2781 2782 2783 2784
  pOperator->blocking = true;
  pOperator->status = OP_NOT_OPENED;
  pOperator->info = pInfo;

L
Liu Jicong 已提交
2785 2786
  pOperator->fpSet = createOperatorFpSet(NULL, doStreamFinalIntervalAgg, NULL, destroyStreamFinalIntervalOperatorInfo,
                                         optrDefaultBufFn, NULL);
2787
  if (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL) {
5
54liuyao 已提交
2788
    initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup);
2789
  }
5
54liuyao 已提交
2790 2791 2792 2793 2794 2795 2796 2797
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

_error:
2798
  destroyStreamFinalIntervalOperatorInfo(pInfo);
5
54liuyao 已提交
2799 2800 2801
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
5
54liuyao 已提交
2802
}
5
54liuyao 已提交
2803 2804

void destroyStreamAggSupporter(SStreamAggSupporter* pSup) {
5
54liuyao 已提交
2805
  tSimpleHashCleanup(pSup->pResultRows);
5
54liuyao 已提交
2806 2807
  destroyDiskbasedBuf(pSup->pResultBuf);
  blockDataDestroy(pSup->pScanBlock);
5
54liuyao 已提交
2808 2809
  taosMemoryFreeClear(pSup->pState);
  taosMemoryFreeClear(pSup->pDummyCtx);
5
54liuyao 已提交
2810 2811
}

2812
void destroyStreamSessionAggOperatorInfo(void* param) {
5
54liuyao 已提交
2813
  SStreamSessionAggOperatorInfo* pInfo = (SStreamSessionAggOperatorInfo*)param;
2814
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
2815
  destroyStreamAggSupporter(&pInfo->streamAggSup);
5
54liuyao 已提交
2816

2817 2818 2819
  if (pInfo->pChildren != NULL) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
5
54liuyao 已提交
2820 2821
      SOperatorInfo* pChild = taosArrayGetP(pInfo->pChildren, i);
      destroyOperatorInfo(pChild);
2822
    }
5
54liuyao 已提交
2823
    taosArrayDestroy(pInfo->pChildren);
2824
  }
5
54liuyao 已提交
2825 2826 2827 2828
  colDataDestroy(&pInfo->twAggSup.timeWindowData);
  blockDataDestroy(pInfo->pDelRes);
  blockDataDestroy(pInfo->pWinBlock);
  blockDataDestroy(pInfo->pUpdateRes);
5
54liuyao 已提交
2829
  tSimpleHashCleanup(pInfo->pStDeleted);
2830

D
dapan1121 已提交
2831
  taosMemoryFreeClear(param);
5
54liuyao 已提交
2832 2833
}

2834 2835
int32_t initBasicInfoEx(SOptrBasicInfo* pBasicInfo, SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfCols,
                        SSDataBlock* pResultBlock) {
H
Haojun Liao 已提交
2836
  initBasicInfo(pBasicInfo, pResultBlock);
2837 2838 2839 2840
  int32_t code = initExprSupp(pSup, pExprInfo, numOfCols);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }
2841

H
Haojun Liao 已提交
2842
  initStreamFunciton(pSup->pCtx, pSup->numOfExprs);
5
54liuyao 已提交
2843
  for (int32_t i = 0; i < numOfCols; ++i) {
2844
    pSup->pCtx[i].saveHandle.pBuf = NULL;
5
54liuyao 已提交
2845
  }
2846

2847
  ASSERT(numOfCols > 0);
5
54liuyao 已提交
2848 2849 2850 2851 2852 2853
  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;
2854 2855
    pDummy[i].isNotNullFunc = pCtx[i].isNotNullFunc;
    pDummy[i].isPseudoFunc = pCtx[i].isPseudoFunc;
5
54liuyao 已提交
2856 2857
  }
}
5
54liuyao 已提交
2858

5
54liuyao 已提交
2859 2860
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex,
                    STimeWindowAggSupp* pTwSup) {
2861 2862 2863 2864 2865 2866
  if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) {
    SStreamPartitionOperatorInfo* pScanInfo = downstream->info;
    pScanInfo->tsColIndex = tsColIndex;
  }

  if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
5
54liuyao 已提交
2867
    initDownStream(downstream->pDownstream[0], pAggSup, type, tsColIndex, pTwSup);
2868 2869
    return;
  }
2870
  SStreamScanInfo* pScanInfo = downstream->info;
5
54liuyao 已提交
2871
  pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type};
5
54liuyao 已提交
2872
  if (!pScanInfo->igCheckUpdate && !pScanInfo->pUpdateInfo) {
5
54liuyao 已提交
2873
    pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, pTwSup->waterMark);
5
54liuyao 已提交
2874
  }
5
54liuyao 已提交
2875
  pScanInfo->twAggSup = *pTwSup;
5
54liuyao 已提交
2876 2877
}

5
54liuyao 已提交
2878 2879 2880 2881 2882 2883 2884 2885 2886 2887
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, int64_t gap,
                               SStreamState* pState, int32_t keySize, int16_t keyType) {
  pSup->resultRowSize = keySize + getResultRowSize(pCtx, numOfOutput);
  pSup->pScanBlock = createSpecialDataBlock(STREAM_CLEAR);
  pSup->gap = gap;
  pSup->stateKeySize = keySize;
  pSup->stateKeyType = keyType;
  pSup->pDummyCtx = (SqlFunctionCtx*)taosMemoryCalloc(numOfOutput, sizeof(SqlFunctionCtx));
  if (pSup->pDummyCtx == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
5
54liuyao 已提交
2888
  }
H
Haojun Liao 已提交
2889

5
54liuyao 已提交
2890 2891 2892 2893
  initDummyFunction(pSup->pDummyCtx, pCtx, numOfOutput);
  pSup->pState = taosMemoryCalloc(1, sizeof(SStreamState));
  *(pSup->pState) = *pState;
  streamStateSetNumber(pSup->pState, -1);
2894

5
54liuyao 已提交
2895 2896
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  pSup->pResultRows = tSimpleHashInit(32, hashFn);
X
Xiaoyu Wang 已提交
2897

5
54liuyao 已提交
2898 2899 2900
  int32_t pageSize = 4096;
  while (pageSize < pSup->resultRowSize * 4) {
    pageSize <<= 1u;
5
54liuyao 已提交
2901
  }
5
54liuyao 已提交
2902 2903 2904 2905
  // at least four pages need to be in buffer
  int32_t bufSize = 4096 * 256;
  if (bufSize <= pageSize) {
    bufSize = pageSize * 4;
5
54liuyao 已提交
2906
  }
5
54liuyao 已提交
2907 2908 2909 2910
  if (!osTempSpaceAvailable()) {
    terrno = TSDB_CODE_NO_AVAIL_DISK;
    qError("Init stream agg supporter failed since %s", terrstr(terrno));
    return terrno;
5
54liuyao 已提交
2911
  }
5
54liuyao 已提交
2912 2913 2914
  int32_t code = createDiskbasedBuf(&pSup->pResultBuf, pageSize, bufSize, "function", tsTempDir);
  for (int32_t i = 0; i < numOfOutput; ++i) {
    pCtx[i].saveHandle.pBuf = pSup->pResultBuf;
5
54liuyao 已提交
2915 2916
  }

5
54liuyao 已提交
2917
  return TSDB_CODE_SUCCESS;
5
54liuyao 已提交
2918
}
5
54liuyao 已提交
2919 2920

bool isInTimeWindow(STimeWindow* pWin, TSKEY ts, int64_t gap) {
5
54liuyao 已提交
2921
  if (ts + gap >= pWin->skey && ts - gap <= pWin->ekey) {
5
54liuyao 已提交
2922 2923 2924 2925 2926
    return true;
  }
  return false;
}

5
54liuyao 已提交
2927 2928
bool isInWindow(SResultWindowInfo* pWinInfo, TSKEY ts, int64_t gap) {
  return isInTimeWindow(&pWinInfo->sessionWin.win, ts, gap);
5
54liuyao 已提交
2929 2930
}

5
54liuyao 已提交
2931 2932 2933 2934 2935
void getCurSessionWindow(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
                         SSessionKey* pKey) {
  pKey->win.skey = startTs;
  pKey->win.ekey = endTs;
  pKey->groupId = groupId;
2936
  int32_t code = streamStateSessionGetKeyByRange(pAggSup->pState, pKey, pKey);
5
54liuyao 已提交
2937 2938
  if (code != TSDB_CODE_SUCCESS) {
    SET_SESSION_WIN_KEY_INVALID(pKey);
2939 2940 2941
  }
}

5
54liuyao 已提交
2942
bool isInvalidSessionWin(SResultWindowInfo* pWinInfo) { return pWinInfo->sessionWin.win.skey == 0; }
5
54liuyao 已提交
2943

5
54liuyao 已提交
2944 2945 2946
void setSessionOutputBuf(SStreamAggSupporter* pAggSup, TSKEY startTs, TSKEY endTs, uint64_t groupId,
                         SResultWindowInfo* pCurWin) {
  pCurWin->sessionWin.groupId = groupId;
2947 2948
  pCurWin->sessionWin.win.skey = startTs;
  pCurWin->sessionWin.win.ekey = endTs;
5
54liuyao 已提交
2949
  int32_t size = pAggSup->resultRowSize;
2950 2951
  int32_t code =
      streamStateSessionAddIfNotExist(pAggSup->pState, &pCurWin->sessionWin, pAggSup->gap, &pCurWin->pOutputBuf, &size);
5
54liuyao 已提交
2952 2953 2954 2955 2956
  if (code == TSDB_CODE_SUCCESS) {
    pCurWin->isOutput = true;
  } else {
    pCurWin->sessionWin.win.skey = startTs;
    pCurWin->sessionWin.win.ekey = endTs;
5
54liuyao 已提交
2957
  }
5
54liuyao 已提交
2958
}
5
54liuyao 已提交
2959

5
54liuyao 已提交
2960 2961
int32_t getSessionWinBuf(SStreamAggSupporter* pAggSup, SStreamStateCur* pCur, SResultWindowInfo* pWinInfo) {
  int32_t size = 0;
2962
  int32_t code = streamStateSessionGetKVByCur(pCur, &pWinInfo->sessionWin, &pWinInfo->pOutputBuf, &size);
5
54liuyao 已提交
2963 2964
  if (code != TSDB_CODE_SUCCESS) {
    return code;
5
54liuyao 已提交
2965
  }
5
54liuyao 已提交
2966 2967 2968 2969 2970 2971
  streamStateCurNext(pAggSup->pState, pCur);
  return TSDB_CODE_SUCCESS;
}
void saveDeleteInfo(SArray* pWins, SSessionKey key) {
  // key.win.ekey = key.win.skey;
  taosArrayPush(pWins, &key);
5
54liuyao 已提交
2972 2973
}

5
54liuyao 已提交
2974 2975 2976 2977
void saveDeleteRes(SSHashObj* pStDelete, SSessionKey key) {
  key.win.ekey = key.win.skey;
  tSimpleHashPut(pStDelete, &key, sizeof(SSessionKey), NULL, 0);
}
2978

5
54liuyao 已提交
2979 2980 2981 2982 2983
static void removeSessionResult(SSHashObj* pHashMap, SSHashObj* pResMap, SSessionKey key) {
  key.win.ekey = key.win.skey;
  tSimpleHashRemove(pHashMap, &key, sizeof(SSessionKey));
  tSimpleHashRemove(pResMap, &key, sizeof(SSessionKey));
}
5
54liuyao 已提交
2984

5
54liuyao 已提交
2985 2986 2987 2988 2989
static void getSessionHashKey(const SSessionKey* pKey, SSessionKey* pHashKey) {
  *pHashKey = *pKey;
  pHashKey->win.ekey = pKey->win.skey;
}

5
54liuyao 已提交
2990 2991 2992
static void removeSessionResults(SSHashObj* pHashMap, SArray* pWins) {
  if (tSimpleHashGetSize(pHashMap) == 0) {
    return;
5
54liuyao 已提交
2993
  }
5
54liuyao 已提交
2994 2995 2996 2997
  int32_t size = taosArrayGetSize(pWins);
  for (int32_t i = 0; i < size; i++) {
    SSessionKey* pWin = taosArrayGet(pWins, i);
    if (!pWin) continue;
5
54liuyao 已提交
2998 2999
    SSessionKey key = {0};
    getSessionHashKey(pWin, &key);
5
54liuyao 已提交
3000
    tSimpleHashRemove(pHashMap, &key, sizeof(SSessionKey));
5
54liuyao 已提交
3001 3002 3003
  }
}

dengyihao's avatar
dengyihao 已提交
3004
int32_t updateSessionWindowInfo(SResultWindowInfo* pWinInfo, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t groupId,
5
54liuyao 已提交
3005 3006
                                int32_t rows, int32_t start, int64_t gap, SSHashObj* pResultRows, SSHashObj* pStUpdated,
                                SSHashObj* pStDeleted) {
5
54liuyao 已提交
3007
  for (int32_t i = start; i < rows; ++i) {
3008
    if (!isInWindow(pWinInfo, pStartTs[i], gap) && (!pEndTs || !isInWindow(pWinInfo, pEndTs[i], gap))) {
5
54liuyao 已提交
3009 3010
      return i - start;
    }
5
54liuyao 已提交
3011
    if (pWinInfo->sessionWin.win.skey > pStartTs[i]) {
5
54liuyao 已提交
3012
      if (pStDeleted && pWinInfo->isOutput) {
5
54liuyao 已提交
3013
        saveDeleteRes(pStDeleted, pWinInfo->sessionWin);
5
54liuyao 已提交
3014
      }
5
54liuyao 已提交
3015 3016
      removeSessionResult(pStUpdated, pResultRows, pWinInfo->sessionWin);
      pWinInfo->sessionWin.win.skey = pStartTs[i];
5
54liuyao 已提交
3017
    }
5
54liuyao 已提交
3018
    pWinInfo->sessionWin.win.ekey = TMAX(pWinInfo->sessionWin.win.ekey, pStartTs[i]);
5
54liuyao 已提交
3019
    if (pEndTs) {
5
54liuyao 已提交
3020
      pWinInfo->sessionWin.win.ekey = TMAX(pWinInfo->sessionWin.win.ekey, pEndTs[i]);
5
54liuyao 已提交
3021 3022 3023 3024 3025
    }
  }
  return rows - start;
}

5
54liuyao 已提交
3026 3027
static int32_t initSessionOutputBuf(SResultWindowInfo* pWinInfo, SResultRow** pResult, SqlFunctionCtx* pCtx,
                                    int32_t numOfOutput, int32_t* rowEntryInfoOffset) {
3028
  ASSERT(pWinInfo->sessionWin.win.skey <= pWinInfo->sessionWin.win.ekey);
5
54liuyao 已提交
3029
  *pResult = (SResultRow*)pWinInfo->pOutputBuf;
5
54liuyao 已提交
3030
  // set time window for current result
5
54liuyao 已提交
3031
  (*pResult)->win = pWinInfo->sessionWin.win;
3032
  setResultRowInitCtx(*pResult, pCtx, numOfOutput, rowEntryInfoOffset);
5
54liuyao 已提交
3033 3034 3035
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3036 3037 3038
static int32_t doOneWindowAggImpl(SColumnInfoData* pTimeWindowData, SResultWindowInfo* pCurWin, SResultRow** pResult,
                                  int32_t startIndex, int32_t winRows, int32_t rows, int32_t numOutput,
                                  SOperatorInfo* pOperator) {
3039
  SExprSupp*     pSup = &pOperator->exprSupp;
3040
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
3041
  int32_t        code = initSessionOutputBuf(pCurWin, pResult, pSup->pCtx, numOutput, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
3042
  if (code != TSDB_CODE_SUCCESS || (*pResult) == NULL) {
S
Shengliang Guan 已提交
3043
    return TSDB_CODE_OUT_OF_MEMORY;
5
54liuyao 已提交
3044
  }
5
54liuyao 已提交
3045
  updateTimeWindowInfo(pTimeWindowData, &pCurWin->sessionWin.win, false);
H
Haojun Liao 已提交
3046
  applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, pTimeWindowData, startIndex, winRows, rows, numOutput);
5
54liuyao 已提交
3047 3048 3049
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3050 3051
static bool doDeleteSessionWindow(SStreamAggSupporter* pAggSup, SSessionKey* pKey) {
  streamStateSessionDel(pAggSup->pState, pKey);
5
54liuyao 已提交
3052 3053 3054
  SSessionKey hashKey = {0};
  getSessionHashKey(pKey, &hashKey);
  tSimpleHashRemove(pAggSup->pResultRows, &hashKey, sizeof(SSessionKey));
5
54liuyao 已提交
3055 3056 3057 3058 3059 3060 3061 3062 3063 3064
  return true;
}

static int32_t setSessionWinOutputInfo(SSHashObj* pStUpdated, SResultWindowInfo* pWinInfo) {
  void* pVal = tSimpleHashGet(pStUpdated, &pWinInfo->sessionWin, sizeof(SSessionKey));
  if (pVal) {
    SResultWindowInfo* pWin = pVal;
    pWinInfo->isOutput = pWin->isOutput;
  }
  return TSDB_CODE_SUCCESS;
5
54liuyao 已提交
3065 3066
}

5
54liuyao 已提交
3067 3068 3069 3070 3071 3072 3073
SStreamStateCur* getNextSessionWinInfo(SStreamAggSupporter* pAggSup, SSHashObj* pStUpdated, SResultWindowInfo* pCurWin,
                                       SResultWindowInfo* pNextWin) {
  SStreamStateCur* pCur = streamStateSessionSeekKeyNext(pAggSup->pState, &pCurWin->sessionWin);
  pNextWin->isOutput = true;
  setSessionWinOutputInfo(pStUpdated, pNextWin);
  int32_t size = 0;
  pNextWin->sessionWin = pCurWin->sessionWin;
3074
  int32_t code = streamStateSessionGetKVByCur(pCur, &pNextWin->sessionWin, &pNextWin->pOutputBuf, &size);
5
54liuyao 已提交
3075 3076 3077 3078
  if (code != TSDB_CODE_SUCCESS) {
    SET_SESSION_WIN_INVALID(*pNextWin);
  }
  return pCur;
5
54liuyao 已提交
3079 3080
}

5
54liuyao 已提交
3081 3082 3083 3084 3085 3086 3087 3088 3089
static void compactSessionWindow(SOperatorInfo* pOperator, SResultWindowInfo* pCurWin, SSHashObj* pStUpdated,
                                 SSHashObj* pStDeleted) {
  SExprSupp*                     pSup = &pOperator->exprSupp;
  SExecTaskInfo*                 pTaskInfo = pOperator->pTaskInfo;
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
  SResultRow*                    pCurResult = NULL;
  int32_t                        numOfOutput = pOperator->exprSupp.numOfExprs;
  SStreamAggSupporter*           pAggSup = &pInfo->streamAggSup;
  initSessionOutputBuf(pCurWin, &pCurResult, pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset);
5
54liuyao 已提交
3090
  // Just look for the window behind StartIndex
5
54liuyao 已提交
3091 3092 3093 3094 3095 3096
  while (1) {
    SResultWindowInfo winInfo = {0};
    SStreamStateCur*  pCur = getNextSessionWinInfo(pAggSup, pStUpdated, pCurWin, &winInfo);
    if (!IS_VALID_SESSION_WIN(winInfo) || !isInWindow(pCurWin, winInfo.sessionWin.win.skey, pAggSup->gap)) {
      streamStateFreeCur(pCur);
      break;
5
54liuyao 已提交
3097
    }
5
54liuyao 已提交
3098 3099 3100 3101 3102 3103 3104 3105 3106 3107 3108 3109
    SResultRow* pWinResult = NULL;
    initSessionOutputBuf(&winInfo, &pWinResult, pAggSup->pDummyCtx, numOfOutput, pSup->rowEntryInfoOffset);
    pCurWin->sessionWin.win.ekey = TMAX(pCurWin->sessionWin.win.ekey, winInfo.sessionWin.win.ekey);
    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pCurWin->sessionWin.win, true);
    compactFunctions(pSup->pCtx, pAggSup->pDummyCtx, numOfOutput, pTaskInfo, &pInfo->twAggSup.timeWindowData);
    tSimpleHashRemove(pStUpdated, &winInfo.sessionWin, sizeof(SSessionKey));
    if (winInfo.isOutput && pStDeleted) {
      saveDeleteRes(pStDeleted, winInfo.sessionWin);
    }
    removeSessionResult(pStUpdated, pAggSup->pResultRows, winInfo.sessionWin);
    doDeleteSessionWindow(pAggSup, &winInfo.sessionWin);
    streamStateFreeCur(pCur);
5
54liuyao 已提交
3110 3111 3112
  }
}

5
54liuyao 已提交
3113 3114 3115
int32_t saveSessionOutputBuf(SStreamAggSupporter* pAggSup, SResultWindowInfo* pWinInfo) {
  saveSessionDiscBuf(pAggSup->pState, &pWinInfo->sessionWin, pWinInfo->pOutputBuf, pAggSup->resultRowSize);
  return TSDB_CODE_SUCCESS;
5
54liuyao 已提交
3116 3117
}

5
54liuyao 已提交
3118 3119
static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SSHashObj* pStUpdated,
                                   SSHashObj* pStDeleted, bool hasEndTs) {
X
Xiaoyu Wang 已提交
3120
  SExecTaskInfo*                 pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
3121
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
3122
  int32_t                        numOfOutput = pOperator->exprSupp.numOfExprs;
H
Haojun Liao 已提交
3123
  uint64_t                       groupId = pSDataBlock->info.id.groupId;
X
Xiaoyu Wang 已提交
3124
  int64_t                        code = TSDB_CODE_SUCCESS;
5
54liuyao 已提交
3125 3126 3127
  SResultRow*                    pResult = NULL;
  int32_t                        rows = pSDataBlock->info.rows;
  int32_t                        winRows = 0;
X
Xiaoyu Wang 已提交
3128

5
54liuyao 已提交
3129
  SColumnInfoData* pStartTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
3130
  TSKEY*           startTsCols = (int64_t*)pStartTsCol->pData;
5
54liuyao 已提交
3131 3132 3133
  SColumnInfoData* pEndTsCol = NULL;
  if (hasEndTs) {
    pEndTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->endTsIndex);
5
54liuyao 已提交
3134
  } else {
5
54liuyao 已提交
3135
    pEndTsCol = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
5
54liuyao 已提交
3136
  }
X
Xiaoyu Wang 已提交
3137

5
54liuyao 已提交
3138
  TSKEY*               endTsCols = (int64_t*)pEndTsCol->pData;
5
54liuyao 已提交
3139
  SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
5
54liuyao 已提交
3140
  for (int32_t i = 0; i < rows;) {
5
54liuyao 已提交
3141
    if (pInfo->ignoreExpiredData && isOverdue(endTsCols[i], &pInfo->twAggSup)) {
5
54liuyao 已提交
3142 3143 3144
      i++;
      continue;
    }
5
54liuyao 已提交
3145 3146 3147 3148 3149
    SResultWindowInfo winInfo = {0};
    setSessionOutputBuf(pAggSup, startTsCols[i], endTsCols[i], groupId, &winInfo);
    setSessionWinOutputInfo(pStUpdated, &winInfo);
    winRows = updateSessionWindowInfo(&winInfo, startTsCols, endTsCols, groupId, rows, i, pAggSup->gap,
                                      pAggSup->pResultRows, pStUpdated, pStDeleted);
5
54liuyao 已提交
3150 3151
    // coverity scan error
    if (!winInfo.pOutputBuf) {
S
Shengliang Guan 已提交
3152
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
5
54liuyao 已提交
3153
    }
L
Liu Jicong 已提交
3154

5
54liuyao 已提交
3155 3156
    code = doOneWindowAggImpl(&pInfo->twAggSup.timeWindowData, &winInfo, &pResult, i, winRows, rows, numOfOutput,
                              pOperator);
5
54liuyao 已提交
3157
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
S
Shengliang Guan 已提交
3158
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
5
54liuyao 已提交
3159
    }
5
54liuyao 已提交
3160 3161
    compactSessionWindow(pOperator, &winInfo, pStUpdated, pStDeleted);
    saveSessionOutputBuf(pAggSup, &winInfo);
5
54liuyao 已提交
3162 3163

    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE && pStUpdated) {
5
54liuyao 已提交
3164
      code = saveResult(winInfo, pStUpdated);
5
54liuyao 已提交
3165
      if (code != TSDB_CODE_SUCCESS) {
S
Shengliang Guan 已提交
3166
        T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
5
54liuyao 已提交
3167
      }
5
54liuyao 已提交
3168
    }
5
54liuyao 已提交
3169
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
5
54liuyao 已提交
3170 3171
      SSessionKey key = {0};
      getSessionHashKey(&winInfo.sessionWin, &key);
5
54liuyao 已提交
3172 3173 3174
      tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SSessionKey), &winInfo, sizeof(SResultWindowInfo));
    }

5
54liuyao 已提交
3175 3176 3177 3178
    i += winRows;
  }
}

5
54liuyao 已提交
3179
static void doDeleteTimeWindows(SStreamAggSupporter* pAggSup, SSDataBlock* pBlock, SArray* result) {
5
54liuyao 已提交
3180
  SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
3181
  TSKEY*           startDatas = (TSKEY*)pStartTsCol->pData;
5
54liuyao 已提交
3182
  SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
3183
  TSKEY*           endDatas = (TSKEY*)pEndTsCol->pData;
3184
  SColumnInfoData* pGroupCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
3185
  uint64_t*        gpDatas = (uint64_t*)pGroupCol->pData;
5
54liuyao 已提交
3186
  for (int32_t i = 0; i < pBlock->info.rows; i++) {
5
54liuyao 已提交
3187 3188 3189 3190
    while (1) {
      SSessionKey curWin = {0};
      getCurSessionWindow(pAggSup, startDatas[i], endDatas[i], gpDatas[i], &curWin);
      if (IS_INVALID_SESSION_WIN_KEY(curWin)) {
3191 3192
        break;
      }
5
54liuyao 已提交
3193 3194 3195 3196
      doDeleteSessionWindow(pAggSup, &curWin);
      if (result) {
        saveDeleteInfo(result, curWin);
      }
3197
    }
5
54liuyao 已提交
3198 3199 3200
  }
}

5
54liuyao 已提交
3201 3202 3203 3204 3205 3206 3207 3208
static inline int32_t sessionKeyCompareAsc(const void* pKey1, const void* pKey2) {
  SSessionKey* pWin1 = (SSessionKey*)pKey1;
  SSessionKey* pWin2 = (SSessionKey*)pKey2;

  if (pWin1->groupId > pWin2->groupId) {
    return 1;
  } else if (pWin1->groupId < pWin2->groupId) {
    return -1;
5
54liuyao 已提交
3209 3210
  }

5
54liuyao 已提交
3211 3212 3213 3214 3215 3216 3217 3218 3219 3220 3221 3222 3223
  if (pWin1->win.skey > pWin2->win.skey) {
    return 1;
  } else if (pWin1->win.skey < pWin2->win.skey) {
    return -1;
  }

  return 0;
}

static int32_t copyUpdateResult(SSHashObj* pStUpdated, SArray* pUpdated) {
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pStUpdated, pIte, &iter)) != NULL) {
3224
    void* key = tSimpleHashGetKey(pIte, NULL);
5
54liuyao 已提交
3225 3226 3227 3228 3229 3230
    taosArrayPush(pUpdated, key);
  }
  taosArraySort(pUpdated, sessionKeyCompareAsc);
  return TSDB_CODE_SUCCESS;
}

3231
void doBuildDeleteDataBlock(SOperatorInfo* pOp, SSHashObj* pStDeleted, SSDataBlock* pBlock, void** Ite) {
5
54liuyao 已提交
3232 3233 3234 3235
  blockDataCleanup(pBlock);
  int32_t size = tSimpleHashGetSize(pStDeleted);
  if (size == 0) {
    return;
3236 3237
  }
  blockDataEnsureCapacity(pBlock, size);
5
54liuyao 已提交
3238 3239 3240 3241 3242
  int32_t iter = 0;
  while (((*Ite) = tSimpleHashIterate(pStDeleted, *Ite, &iter)) != NULL) {
    if (pBlock->info.rows + 1 > pBlock->info.capacity) {
      break;
    }
3243
    SSessionKey*     res = tSimpleHashGetKey(*Ite, NULL);
3244
    SColumnInfoData* pStartTsCol = taosArrayGet(pBlock->pDataBlock, START_TS_COLUMN_INDEX);
3245
    colDataSetVal(pStartTsCol, pBlock->info.rows, (const char*)&res->win.skey, false);
3246
    SColumnInfoData* pEndTsCol = taosArrayGet(pBlock->pDataBlock, END_TS_COLUMN_INDEX);
3247
    colDataSetVal(pEndTsCol, pBlock->info.rows, (const char*)&res->win.skey, false);
3248
    SColumnInfoData* pUidCol = taosArrayGet(pBlock->pDataBlock, UID_COLUMN_INDEX);
3249
    colDataSetNULL(pUidCol, pBlock->info.rows);
5
54liuyao 已提交
3250
    SColumnInfoData* pGpCol = taosArrayGet(pBlock->pDataBlock, GROUPID_COLUMN_INDEX);
3251
    colDataSetVal(pGpCol, pBlock->info.rows, (const char*)&res->groupId, false);
3252
    SColumnInfoData* pCalStCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_START_TS_COLUMN_INDEX);
3253
    colDataSetNULL(pCalStCol, pBlock->info.rows);
3254
    SColumnInfoData* pCalEdCol = taosArrayGet(pBlock->pDataBlock, CALCULATE_END_TS_COLUMN_INDEX);
3255
    colDataSetNULL(pCalEdCol, pBlock->info.rows);
3256 3257

    SColumnInfoData* pTableCol = taosArrayGet(pBlock->pDataBlock, TABLE_NAME_COLUMN_INDEX);
3258 3259 3260

    void* tbname = NULL;
    streamStateGetParName(pOp->pTaskInfo->streamInfo.pState, res->groupId, &tbname);
3261
    if (tbname == NULL) {
3262
      colDataSetNULL(pTableCol, pBlock->info.rows);
3263 3264 3265
    } else {
      char parTbName[VARSTR_HEADER_SIZE + TSDB_TABLE_NAME_LEN];
      STR_WITH_MAXSIZE_TO_VARSTR(parTbName, tbname, sizeof(parTbName));
3266
      colDataSetVal(pTableCol, pBlock->info.rows, (const char*)parTbName, false);
L
Liu Jicong 已提交
3267
      tdbFree(tbname);
3268
    }
5
54liuyao 已提交
3269 3270 3271
    pBlock->info.rows += 1;
  }
  if ((*Ite) == NULL) {
5
54liuyao 已提交
3272
    tSimpleHashClear(pStDeleted);
5
54liuyao 已提交
3273 3274 3275
  }
}

5
54liuyao 已提交
3276 3277 3278 3279 3280 3281 3282 3283
static void rebuildSessionWindow(SOperatorInfo* pOperator, SArray* pWinArray, SSHashObj* pStUpdated) {
  SExprSupp*                     pSup = &pOperator->exprSupp;
  SExecTaskInfo*                 pTaskInfo = pOperator->pTaskInfo;
  int32_t                        size = taosArrayGetSize(pWinArray);
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
  SStreamAggSupporter*           pAggSup = &pInfo->streamAggSup;
  int32_t                        numOfOutput = pSup->numOfExprs;
  int32_t                        numOfChildren = taosArrayGetSize(pInfo->pChildren);
3284

3285
  for (int32_t i = 0; i < size; i++) {
5
54liuyao 已提交
3286 3287 3288
    SSessionKey*      pWinKey = taosArrayGet(pWinArray, i);
    int32_t           num = 0;
    SResultWindowInfo parentWin = {0};
3289
    for (int32_t j = 0; j < numOfChildren; j++) {
X
Xiaoyu Wang 已提交
3290
      SOperatorInfo*                 pChild = taosArrayGetP(pInfo->pChildren, j);
3291
      SStreamSessionAggOperatorInfo* pChInfo = pChild->info;
5
54liuyao 已提交
3292
      SStreamAggSupporter*           pChAggSup = &pChInfo->streamAggSup;
5
54liuyao 已提交
3293 3294
      SSessionKey                    chWinKey = {0};
      getSessionHashKey(pWinKey, &chWinKey);
3295 3296 3297
      SStreamStateCur* pCur = streamStateSessionSeekKeyCurrentNext(pChAggSup->pState, &chWinKey);
      SResultRow*      pResult = NULL;
      SResultRow*      pChResult = NULL;
5
54liuyao 已提交
3298 3299 3300 3301 3302 3303 3304 3305 3306 3307 3308 3309
      while (1) {
        SResultWindowInfo childWin = {0};
        childWin.sessionWin = *pWinKey;
        int32_t code = getSessionWinBuf(pChAggSup, pCur, &childWin);
        if (code == TSDB_CODE_SUCCESS && pWinKey->win.skey <= childWin.sessionWin.win.skey &&
            childWin.sessionWin.win.ekey <= pWinKey->win.ekey) {
          if (num == 0) {
            setSessionOutputBuf(pAggSup, pWinKey->win.skey, pWinKey->win.ekey, pWinKey->groupId, &parentWin);
            code = initSessionOutputBuf(&parentWin, &pResult, pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset);
            if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
              break;
            }
3310
          }
5
54liuyao 已提交
3311 3312 3313 3314 3315 3316 3317 3318
          num++;
          updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &parentWin.sessionWin.win, true);
          initSessionOutputBuf(&childWin, &pChResult, pChild->exprSupp.pCtx, numOfOutput,
                               pChild->exprSupp.rowEntryInfoOffset);
          compactFunctions(pSup->pCtx, pChild->exprSupp.pCtx, numOfOutput, pTaskInfo, &pInfo->twAggSup.timeWindowData);
          compactSessionWindow(pOperator, &parentWin, pStUpdated, NULL);
          saveResult(parentWin, pStUpdated);
        } else {
5
54liuyao 已提交
3319
          break;
3320 3321
        }
      }
5
54liuyao 已提交
3322 3323 3324 3325
      streamStateFreeCur(pCur);
    }
    if (num > 0) {
      saveSessionOutputBuf(pAggSup, &parentWin);
3326 3327 3328 3329
    }
  }
}

5
54liuyao 已提交
3330 3331 3332 3333 3334 3335 3336 3337 3338 3339
int32_t closeSessionWindow(SSHashObj* pHashMap, STimeWindowAggSupp* pTwSup, SSHashObj* pClosed) {
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pHashMap, pIte, &iter)) != NULL) {
    SResultWindowInfo* pWinInfo = pIte;
    if (isCloseWindow(&pWinInfo->sessionWin.win, pTwSup)) {
      if (pTwSup->calTrigger == STREAM_TRIGGER_WINDOW_CLOSE && pClosed) {
        int32_t code = saveResult(*pWinInfo, pClosed);
        if (code != TSDB_CODE_SUCCESS) {
          return code;
5
54liuyao 已提交
3340 3341
        }
      }
3342
      SSessionKey* pKey = tSimpleHashGetKey(pIte, NULL);
3343
      tSimpleHashIterateRemove(pHashMap, pKey, sizeof(SSessionKey), &pIte, &iter);
5
54liuyao 已提交
3344 3345 3346 3347 3348
    }
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3349
static void closeChildSessionWindow(SArray* pChildren, TSKEY maxTs) {
5
54liuyao 已提交
3350 3351 3352 3353 3354
  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);
5
54liuyao 已提交
3355
    closeSessionWindow(pChInfo->streamAggSup.pResultRows, &pChInfo->twAggSup, NULL);
5
54liuyao 已提交
3356 3357 3358
  }
}

5
54liuyao 已提交
3359 3360 3361 3362
int32_t getAllSessionWindow(SSHashObj* pHashMap, SSHashObj* pStUpdated) {
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pHashMap, pIte, &iter)) != NULL) {
3363
    SResultWindowInfo* pWinInfo = pIte;
5
54liuyao 已提交
3364
    saveResult(*pWinInfo, pStUpdated);
5
54liuyao 已提交
3365 3366 3367 3368
  }
  return TSDB_CODE_SUCCESS;
}

5
54liuyao 已提交
3369
static void copyDeleteWindowInfo(SArray* pResWins, SSHashObj* pStDeleted) {
5
54liuyao 已提交
3370 3371
  int32_t size = taosArrayGetSize(pResWins);
  for (int32_t i = 0; i < size; i++) {
5
54liuyao 已提交
3372 3373
    SSessionKey* pWinKey = taosArrayGet(pResWins, i);
    if (!pWinKey) continue;
5
54liuyao 已提交
3374 3375
    SSessionKey winInfo = {0};
    getSessionHashKey(pWinKey, &winInfo);
5
54liuyao 已提交
3376
    tSimpleHashPut(pStDeleted, &winInfo, sizeof(SSessionKey), NULL, 0);
3377 3378 3379
  }
}

H
Haojun Liao 已提交
3380
// the allocated memory comes from outer function.
5
54liuyao 已提交
3381 3382 3383
void initGroupResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList) {
  pGroupResInfo->pRows = pArrayList;
  pGroupResInfo->index = 0;
H
Haojun Liao 已提交
3384
  pGroupResInfo->pBuf = NULL;
3385 3386
}

5
54liuyao 已提交
3387 3388 3389 3390 3391 3392 3393 3394
void doBuildSessionResult(SOperatorInfo* pOperator, SStreamState* pState, SGroupResInfo* pGroupResInfo,
                          SSDataBlock* pBlock) {
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
  // set output datablock version
  pBlock->info.version = pTaskInfo->version;

  blockDataCleanup(pBlock);
  if (!hasRemainResults(pGroupResInfo)) {
H
Haojun Liao 已提交
3395
    cleanupGroupResInfo(pGroupResInfo);
3396 3397 3398
    return;
  }

5
54liuyao 已提交
3399
  // clear the existed group id
H
Haojun Liao 已提交
3400
  pBlock->info.id.groupId = 0;
3401
  buildSessionResultDataBlock(pOperator, pState, pBlock, &pOperator->exprSupp, pGroupResInfo);
5
54liuyao 已提交
3402 3403
}

5
54liuyao 已提交
3404
static SSDataBlock* doStreamSessionAgg(SOperatorInfo* pOperator) {
5
54liuyao 已提交
3405
  SExprSupp*                     pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3406
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
3407
  SOptrBasicInfo*                pBInfo = &pInfo->binfo;
5
54liuyao 已提交
3408
  SStreamAggSupporter*           pAggSup = &pInfo->streamAggSup;
5
54liuyao 已提交
3409 3410 3411
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  } else if (pOperator->status == OP_RES_TO_RETURN) {
3412
    doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
5
54liuyao 已提交
3413
    if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
3414
      printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
5
54liuyao 已提交
3415 3416
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
3417 3418 3419 3420
    doBuildSessionResult(pOperator, pAggSup->pState, &pInfo->groupResInfo, pBInfo->pRes);
    if (pBInfo->pRes->info.rows > 0) {
      printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
      return pBInfo->pRes;
5
54liuyao 已提交
3421
    }
5
54liuyao 已提交
3422

H
Haojun Liao 已提交
3423
    setOperatorCompleted(pOperator);
5
54liuyao 已提交
3424
    return NULL;
5
54liuyao 已提交
3425 3426 3427
  }

  SOperatorInfo* downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
3428 3429 3430 3431
  if (!pInfo->pUpdated) {
    pInfo->pUpdated = taosArrayInit(16, sizeof(SSessionKey));
  }
  if (!pInfo->pStUpdated) {
3432
    _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
5
54liuyao 已提交
3433 3434
    pInfo->pStUpdated = tSimpleHashInit(64, hashFn);
  }
5
54liuyao 已提交
3435 3436 3437 3438 3439
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
      break;
    }
5
54liuyao 已提交
3440
    printDataBlock(pBlock, IS_FINAL_OP(pInfo) ? "final session recv" : "single session recv");
3441

5
54liuyao 已提交
3442 3443 3444
    if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
        pBlock->info.type == STREAM_CLEAR) {
      SArray* pWins = taosArrayInit(16, sizeof(SSessionKey));
5
54liuyao 已提交
3445
      // gap must be 0
5
54liuyao 已提交
3446
      doDeleteTimeWindows(pAggSup, pBlock, pWins);
5
54liuyao 已提交
3447
      removeSessionResults(pInfo->pStUpdated, pWins);
5
54liuyao 已提交
3448 3449 3450 3451 3452
      if (IS_FINAL_OP(pInfo)) {
        int32_t                        childIndex = getChildIndex(pBlock);
        SOperatorInfo*                 pChildOp = taosArrayGetP(pInfo->pChildren, childIndex);
        SStreamSessionAggOperatorInfo* pChildInfo = pChildOp->info;
        // gap must be 0
5
54liuyao 已提交
3453
        doDeleteTimeWindows(&pChildInfo->streamAggSup, pBlock, NULL);
5
54liuyao 已提交
3454
        rebuildSessionWindow(pOperator, pWins, pInfo->pStUpdated);
5
54liuyao 已提交
3455 3456 3457 3458
      }
      copyDeleteWindowInfo(pWins, pInfo->pStDeleted);
      taosArrayDestroy(pWins);
      continue;
3459
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
3460
      getAllSessionWindow(pAggSup->pResultRows, pInfo->pStUpdated);
5
54liuyao 已提交
3461
      continue;
5
54liuyao 已提交
3462 3463 3464 3465
    } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
      return pBlock;
    } else {
      ASSERTS(pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_INVALID, "invalid SSDataBlock type");
5
54liuyao 已提交
3466
    }
5
54liuyao 已提交
3467

5
54liuyao 已提交
3468 3469 3470 3471
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
3472
    // the pDataBlock are always the same one, no need to call this again
3473
    setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
3474
    doStreamSessionAggImpl(pOperator, pBlock, pInfo->pStUpdated, pInfo->pStDeleted, IS_FINAL_OP(pInfo));
5
54liuyao 已提交
3475 3476 3477 3478 3479
    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++) {
3480 3481
        SOperatorInfo* pChildOp =
            createStreamFinalSessionAggOperatorInfo(NULL, pInfo->pPhyNode, pOperator->pTaskInfo, 0);
5
54liuyao 已提交
3482
        if (!pChildOp) {
S
Shengliang Guan 已提交
3483
          T_LONG_JMP(pOperator->pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
5
54liuyao 已提交
3484 3485 3486
        }
        taosArrayPush(pInfo->pChildren, &pChildOp);
      }
3487
      SOperatorInfo* pChildOp = taosArrayGetP(pInfo->pChildren, chIndex);
3488
      setInputDataBlock(&pChildOp->exprSupp, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
3489
      doStreamSessionAggImpl(pChildOp, pBlock, NULL, NULL, true);
3490
    }
5
54liuyao 已提交
3491 3492
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.watermark);
5
54liuyao 已提交
3493 3494 3495
  }
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
H
Haojun Liao 已提交
3496

5
54liuyao 已提交
3497
  closeSessionWindow(pAggSup->pResultRows, &pInfo->twAggSup, pInfo->pStUpdated);
5
54liuyao 已提交
3498
  closeChildSessionWindow(pInfo->pChildren, pInfo->twAggSup.maxTs);
5
54liuyao 已提交
3499 3500 3501 3502 3503 3504
  copyUpdateResult(pInfo->pStUpdated, pInfo->pUpdated);
  removeSessionResults(pInfo->pStDeleted, pInfo->pUpdated);
  tSimpleHashCleanup(pInfo->pStUpdated);
  pInfo->pStUpdated = NULL;
  initGroupResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);
  pInfo->pUpdated = NULL;
5
54liuyao 已提交
3505
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
5
54liuyao 已提交
3506

3507 3508 3509 3510 3511 3512
#if 0
  char* pBuf = streamStateSessionDump(pAggSup->pState);
  qDebug("===stream===final session%s", pBuf);
  taosMemoryFree(pBuf);
#endif

3513
  doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
5
54liuyao 已提交
3514
  if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
3515
    printDataBlock(pInfo->pDelRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
5
54liuyao 已提交
3516 3517
    return pInfo->pDelRes;
  }
5
54liuyao 已提交
3518 3519 3520 3521 3522 3523 3524

  doBuildSessionResult(pOperator, pAggSup->pState, &pInfo->groupResInfo, pBInfo->pRes);
  if (pBInfo->pRes->info.rows > 0) {
    printDataBlock(pBInfo->pRes, IS_FINAL_OP(pInfo) ? "final session" : "single session");
    return pBInfo->pRes;
  }

H
Haojun Liao 已提交
3525
  setOperatorCompleted(pOperator);
5
54liuyao 已提交
3526
  return NULL;
5
54liuyao 已提交
3527 3528
}

5
54liuyao 已提交
3529 3530 3531 3532 3533 3534 3535 3536 3537 3538 3539 3540 3541 3542 3543 3544 3545 3546 3547 3548
SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                  SExecTaskInfo* pTaskInfo) {
  SSessionWinodwPhysiNode*       pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
  int32_t                        numOfCols = 0;
  int32_t                        code = TSDB_CODE_OUT_OF_MEMORY;
  SStreamSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamSessionAggOperatorInfo));
  SOperatorInfo*                 pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }

  pOperator->pTaskInfo = pTaskInfo;

  initResultSizeInfo(&pOperator->resultInfo, 4096);
  if (pSessionNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pSessionNode->window.pExprs, NULL, &numOfScalar);
    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
5
54liuyao 已提交
3549 3550
    }
  }
5
54liuyao 已提交
3551 3552 3553
  SExprSupp* pSup = &pOperator->exprSupp;

  SExprInfo*   pExprInfo = createExprInfo(pSessionNode->window.pFuncs, NULL, &numOfCols);
H
Haojun Liao 已提交
3554
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);
5
54liuyao 已提交
3555 3556 3557 3558 3559 3560 3561 3562 3563 3564 3565 3566 3567 3568 3569 3570 3571 3572 3573 3574 3575 3576 3577 3578 3579 3580 3581 3582 3583 3584 3585 3586 3587 3588
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  code = initStreamAggSupporter(&pInfo->streamAggSup, pSup->pCtx, numOfCols, pSessionNode->gap,
                                pTaskInfo->streamInfo.pState, 0, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pSessionNode->window.watermark,
      .calTrigger = pSessionNode->window.triggerType,
      .maxTs = INT64_MIN,
      .minTs = INT64_MAX,
  };

  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

  pInfo->primaryTsIndex = ((SColumnNode*)pSessionNode->window.pTspk)->slotId;
  if (pSessionNode->window.pTsEnd) {
    pInfo->endTsIndex = ((SColumnNode*)pSessionNode->window.pTsEnd)->slotId;
  }
  pInfo->binfo.pRes = pResBlock;
  pInfo->order = TSDB_ORDER_ASC;
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
  pInfo->pStDeleted = tSimpleHashInit(64, hashFn);
  pInfo->pDelIterator = NULL;
  pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT);
  pInfo->pChildren = NULL;
  pInfo->isFinal = false;
  pInfo->pPhyNode = pPhyNode;
  pInfo->ignoreExpiredData = pSessionNode->window.igExpired;
3589
  pInfo->ignoreExpiredDataSaved = false;
5
54liuyao 已提交
3590 3591
  pInfo->pUpdated = NULL;
  pInfo->pStUpdated = NULL;
5
54liuyao 已提交
3592

H
Haojun Liao 已提交
3593 3594
  setOperatorInfo(pOperator, "StreamSessionWindowAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION, true,
                  OP_NOT_OPENED, pInfo, pTaskInfo);
L
Liu Jicong 已提交
3595 3596
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo,
                                         optrDefaultBufFn, NULL);
H
Haojun Liao 已提交
3597

5
54liuyao 已提交
3598
  if (downstream) {
5
54liuyao 已提交
3599
    initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
5
54liuyao 已提交
3600 3601 3602 3603 3604 3605 3606 3607 3608 3609 3610 3611 3612 3613 3614 3615 3616
    code = appendDownstream(pOperator, &downstream, 1);
  }
  return pOperator;

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

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

static void clearStreamSessionOperator(SStreamSessionAggOperatorInfo* pInfo) {
  tSimpleHashClear(pInfo->streamAggSup.pResultRows);
  streamStateSessionClear(pInfo->streamAggSup.pState);
5
54liuyao 已提交
3617 3618 3619 3620 3621 3622 3623
}

static SSDataBlock* doStreamSessionSemiAgg(SOperatorInfo* pOperator) {
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
  SOptrBasicInfo*                pBInfo = &pInfo->binfo;
  TSKEY                          maxTs = INT64_MIN;
  SExprSupp*                     pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3624
  SStreamAggSupporter*           pAggSup = &pInfo->streamAggSup;
3625

5
54liuyao 已提交
3626 3627
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
3628
  }
L
Liu Jicong 已提交
3629

3630
  {
5
54liuyao 已提交
3631
    doBuildSessionResult(pOperator, pAggSup->pState, &pInfo->groupResInfo, pBInfo->pRes);
5
54liuyao 已提交
3632
    if (pBInfo->pRes->info.rows > 0) {
H
Haojun Liao 已提交
3633
      printDataBlock(pBInfo->pRes, "semi session");
5
54liuyao 已提交
3634 3635 3636
      return pBInfo->pRes;
    }

3637
    doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
3638
    if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
3639
      printDataBlock(pInfo->pDelRes, "semi session delete");
5
54liuyao 已提交
3640 3641
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
3642

3643
    if (pOperator->status == OP_RES_TO_RETURN) {
5
54liuyao 已提交
3644
      clearFunctionContext(&pOperator->exprSupp);
3645 3646
      // semi interval operator clear disk buffer
      clearStreamSessionOperator(pInfo);
H
Haojun Liao 已提交
3647
      setOperatorCompleted(pOperator);
3648 3649
      return NULL;
    }
5
54liuyao 已提交
3650 3651 3652
  }

  SOperatorInfo* downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
3653 3654 3655 3656
  if (!pInfo->pUpdated) {
    pInfo->pUpdated = taosArrayInit(16, sizeof(SSessionKey));
  }
  if (!pInfo->pStUpdated) {
3657
    _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
5
54liuyao 已提交
3658 3659
    pInfo->pStUpdated = tSimpleHashInit(64, hashFn);
  }
5
54liuyao 已提交
3660 3661 3662
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
5
54liuyao 已提交
3663
      clearSpecialDataBlock(pInfo->pUpdateRes);
3664
      pOperator->status = OP_RES_TO_RETURN;
5
54liuyao 已提交
3665 3666
      break;
    }
H
Haojun Liao 已提交
3667
    printDataBlock(pBlock, "semi session recv");
5
54liuyao 已提交
3668

5
54liuyao 已提交
3669 3670
    if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
        pBlock->info.type == STREAM_CLEAR) {
5
54liuyao 已提交
3671
      // gap must be 0
3672
      SArray* pWins = taosArrayInit(16, sizeof(SSessionKey));
5
54liuyao 已提交
3673
      doDeleteTimeWindows(&pInfo->streamAggSup, pBlock, pWins);
5
54liuyao 已提交
3674
      removeSessionResults(pInfo->pStUpdated, pWins);
5
54liuyao 已提交
3675
      copyDeleteWindowInfo(pWins, pInfo->pStDeleted);
3676
      taosArrayDestroy(pWins);
5
54liuyao 已提交
3677
      break;
5
54liuyao 已提交
3678
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
3679
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pInfo->pStUpdated);
5
54liuyao 已提交
3680
      continue;
5
54liuyao 已提交
3681 3682 3683 3684
    } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
      return pBlock;
    } else {
      ASSERTS(pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_INVALID, "invalid SSDataBlock type");
5
54liuyao 已提交
3685 3686
    }

5
54liuyao 已提交
3687 3688 3689 3690
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
5
54liuyao 已提交
3691
    // the pDataBlock are always the same one, no need to call this again
3692
    setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
3693
    doStreamSessionAggImpl(pOperator, pBlock, pInfo->pStUpdated, NULL, false);
5
54liuyao 已提交
3694 3695 3696 3697
    maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
  }

  pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
3698
  pBInfo->pRes->info.watermark = pInfo->twAggSup.maxTs;
3699

5
54liuyao 已提交
3700 3701 3702 3703
  copyUpdateResult(pInfo->pStUpdated, pInfo->pUpdated);
  removeSessionResults(pInfo->pStDeleted, pInfo->pUpdated);
  tSimpleHashCleanup(pInfo->pStUpdated);
  pInfo->pStUpdated = NULL;
5
54liuyao 已提交
3704

5
54liuyao 已提交
3705 3706
  initGroupResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);
  pInfo->pUpdated = NULL;
5
54liuyao 已提交
3707
  blockDataEnsureCapacity(pBInfo->pRes, pOperator->resultInfo.capacity);
5
54liuyao 已提交
3708

3709 3710 3711 3712 3713 3714
#if 0
  char* pBuf = streamStateSessionDump(pAggSup->pState);
  qDebug("===stream===semi session%s", pBuf);
  taosMemoryFree(pBuf);
#endif

5
54liuyao 已提交
3715
  doBuildSessionResult(pOperator, pAggSup->pState, &pInfo->groupResInfo, pBInfo->pRes);
5
54liuyao 已提交
3716
  if (pBInfo->pRes->info.rows > 0) {
H
Haojun Liao 已提交
3717
    printDataBlock(pBInfo->pRes, "semi session");
5
54liuyao 已提交
3718 3719 3720
    return pBInfo->pRes;
  }

3721
  doBuildDeleteDataBlock(pOperator, pInfo->pStDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
3722
  if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
3723
    printDataBlock(pInfo->pDelRes, "semi session delete");
5
54liuyao 已提交
3724 3725
    return pInfo->pDelRes;
  }
5
54liuyao 已提交
3726

5
54liuyao 已提交
3727 3728 3729
  clearFunctionContext(&pOperator->exprSupp);
  // semi interval operator clear disk buffer
  clearStreamSessionOperator(pInfo);
H
Haojun Liao 已提交
3730
  setOperatorCompleted(pOperator);
5
54liuyao 已提交
3731
  return NULL;
5
54liuyao 已提交
3732
}
3733

3734 3735
SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                       SExecTaskInfo* pTaskInfo, int32_t numOfChild) {
3736 3737
  int32_t        code = TSDB_CODE_OUT_OF_MEMORY;
  SOperatorInfo* pOperator = createStreamSessionAggOperatorInfo(downstream, pPhyNode, pTaskInfo);
3738 3739 3740
  if (pOperator == NULL) {
    goto _error;
  }
H
Haojun Liao 已提交
3741

3742
  SStreamSessionAggOperatorInfo* pInfo = pOperator->info;
5
54liuyao 已提交
3743

H
Haojun Liao 已提交
3744
  pInfo->isFinal = (pPhyNode->type == QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION);
L
Liu Jicong 已提交
3745
  char* name = (pInfo->isFinal) ? "StreamSessionFinalAggOperator" : "StreamSessionSemiAggOperator";
H
Haojun Liao 已提交
3746 3747

  if (pPhyNode->type != QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION) {
H
Haojun Liao 已提交
3748
    pInfo->pUpdateRes = createSpecialDataBlock(STREAM_CLEAR);
5
54liuyao 已提交
3749
    blockDataEnsureCapacity(pInfo->pUpdateRes, 128);
3750 3751
    pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamSessionSemiAgg, NULL,
                                           destroyStreamSessionAggOperatorInfo, optrDefaultBufFn, NULL);
5
54liuyao 已提交
3752
  }
3753

L
Liu Jicong 已提交
3754
  setOperatorInfo(pOperator, name, pPhyNode->type, false, OP_NOT_OPENED, pInfo, pTaskInfo);
H
Haojun Liao 已提交
3755

5
54liuyao 已提交
3756 3757 3758 3759
  pOperator->operatorType = pPhyNode->type;
  if (numOfChild > 0) {
    pInfo->pChildren = taosArrayInit(numOfChild, sizeof(void*));
    for (int32_t i = 0; i < numOfChild; i++) {
5
54liuyao 已提交
3760 3761
      SOperatorInfo* pChildOp = createStreamFinalSessionAggOperatorInfo(NULL, pPhyNode, pTaskInfo, 0);
      if (pChildOp == NULL) {
5
54liuyao 已提交
3762 3763
        goto _error;
      }
5
54liuyao 已提交
3764 3765 3766 3767
      SStreamSessionAggOperatorInfo* pChInfo = pChildOp->info;
      pChInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
      streamStateSetNumber(pChInfo->streamAggSup.pState, i);
      taosArrayPush(pInfo->pChildren, &pChildOp);
3768 3769
    }
  }
3770 3771 3772 3773 3774

  if (!IS_FINAL_OP(pInfo) || numOfChild == 0) {
    pInfo->twAggSup.calTrigger = STREAM_TRIGGER_AT_ONCE;
  }

3775 3776 3777 3778
  return pOperator;

_error:
  if (pInfo != NULL) {
3779
    destroyStreamSessionAggOperatorInfo(pInfo);
3780 3781 3782 3783 3784
  }
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
5
54liuyao 已提交
3785

3786
void destroyStreamStateOperatorInfo(void* param) {
X
Xiaoyu Wang 已提交
3787
  SStreamStateAggOperatorInfo* pInfo = (SStreamStateAggOperatorInfo*)param;
3788
  cleanupBasicInfo(&pInfo->binfo);
5
54liuyao 已提交
3789
  destroyStreamAggSupporter(&pInfo->streamAggSup);
5
54liuyao 已提交
3790 3791 3792 3793
  cleanupGroupResInfo(&pInfo->groupResInfo);
  if (pInfo->pChildren != NULL) {
    int32_t size = taosArrayGetSize(pInfo->pChildren);
    for (int32_t i = 0; i < size; i++) {
5
54liuyao 已提交
3794 3795
      SOperatorInfo* pChild = taosArrayGetP(pInfo->pChildren, i);
      destroyOperatorInfo(pChild);
5
54liuyao 已提交
3796
    }
5
54liuyao 已提交
3797
    taosArrayDestroy(pInfo->pChildren);
5
54liuyao 已提交
3798
  }
5
54liuyao 已提交
3799 3800
  colDataDestroy(&pInfo->twAggSup.timeWindowData);
  blockDataDestroy(pInfo->pDelRes);
5
54liuyao 已提交
3801
  tSimpleHashCleanup(pInfo->pSeDeleted);
D
dapan1121 已提交
3802
  taosMemoryFreeClear(param);
5
54liuyao 已提交
3803 3804 3805
}

bool isTsInWindow(SStateWindowInfo* pWin, TSKEY ts) {
5
54liuyao 已提交
3806
  if (pWin->winInfo.sessionWin.win.skey <= ts && ts <= pWin->winInfo.sessionWin.win.ekey) {
5
54liuyao 已提交
3807 3808 3809 3810 3811 3812
    return true;
  }
  return false;
}

bool isEqualStateKey(SStateWindowInfo* pWin, char* pKeyData) {
5
54liuyao 已提交
3813 3814 3815 3816 3817 3818 3819 3820 3821 3822 3823 3824 3825 3826 3827 3828 3829 3830 3831 3832 3833 3834 3835 3836 3837 3838 3839 3840 3841 3842 3843 3844
  return pKeyData && compareVal(pKeyData, pWin->pStateKey);
}

bool compareStateKey(void* data, void* key) {
  SStateKeys* stateKey = (SStateKeys*)key;
  stateKey->pData = (char*)key + sizeof(SStateKeys);
  return compareVal(data, stateKey);
}

void setStateOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId, char* pKeyData,
                       SStateWindowInfo* pCurWin, SStateWindowInfo* pNextWin) {
  int32_t size = pAggSup->resultRowSize;
  pCurWin->winInfo.sessionWin.groupId = groupId;
  pCurWin->winInfo.sessionWin.win.skey = ts;
  pCurWin->winInfo.sessionWin.win.ekey = ts;
  int32_t code =
      streamStateStateAddIfNotExist(pAggSup->pState, &pCurWin->winInfo.sessionWin, pKeyData, pAggSup->stateKeySize,
                                    compareStateKey, &pCurWin->winInfo.pOutputBuf, &size);
  pCurWin->pStateKey =
      (SStateKeys*)((char*)pCurWin->winInfo.pOutputBuf + (pAggSup->resultRowSize - pAggSup->stateKeySize));
  pCurWin->pStateKey->bytes = pAggSup->stateKeySize - sizeof(SStateKeys);
  pCurWin->pStateKey->type = pAggSup->stateKeyType;
  pCurWin->pStateKey->pData = (char*)pCurWin->pStateKey + sizeof(SStateKeys);
  pCurWin->pStateKey->isNull = false;

  if (code == TSDB_CODE_SUCCESS) {
    pCurWin->winInfo.isOutput = true;
  } else {
    if (IS_VAR_DATA_TYPE(pAggSup->stateKeyType)) {
      varDataCopy(pCurWin->pStateKey->pData, pKeyData);
    } else {
      memcpy(pCurWin->pStateKey->pData, pKeyData, pCurWin->pStateKey->bytes);
5
54liuyao 已提交
3845 3846 3847
    }
  }

5
54liuyao 已提交
3848 3849 3850 3851 3852 3853
  pNextWin->winInfo.sessionWin = pCurWin->winInfo.sessionWin;
  pNextWin->winInfo.pOutputBuf = NULL;
  SStreamStateCur* pCur = streamStateSessionSeekKeyNext(pAggSup->pState, &pCurWin->winInfo.sessionWin);
  code = streamStateSessionGetKVByCur(pCur, &pNextWin->winInfo.sessionWin, NULL, 0);
  if (code != TSDB_CODE_SUCCESS) {
    SET_SESSION_WIN_INVALID(pNextWin->winInfo);
5
54liuyao 已提交
3854
  }
5
54liuyao 已提交
3855
  streamStateFreeCur(pCur);
5
54liuyao 已提交
3856 3857
}

5
54liuyao 已提交
3858
int32_t updateStateWindowInfo(SStateWindowInfo* pWinInfo, SStateWindowInfo* pNextWin, TSKEY* pTs, uint64_t groupId,
H
Haojun Liao 已提交
3859
                              SColumnInfoData* pKeyCol, int32_t rows, int32_t start, bool* allEqual,
5
54liuyao 已提交
3860
                              SSHashObj* pResultRows, SSHashObj* pSeUpdated, SSHashObj* pSeDeleted) {
5
54liuyao 已提交
3861 3862 3863 3864
  *allEqual = true;
  for (int32_t i = start; i < rows; ++i) {
    char* pKeyData = colDataGetData(pKeyCol, i);
    if (!isTsInWindow(pWinInfo, pTs[i])) {
X
Xiaoyu Wang 已提交
3865
      if (isEqualStateKey(pWinInfo, pKeyData)) {
5
54liuyao 已提交
3866
        if (IS_VALID_SESSION_WIN(pNextWin->winInfo)) {
5
54liuyao 已提交
3867
          // ts belongs to the next window
5
54liuyao 已提交
3868
          if (pTs[i] >= pNextWin->winInfo.sessionWin.win.skey) {
5
54liuyao 已提交
3869 3870 3871 3872 3873 3874 3875
            return i - start;
          }
        }
      } else {
        return i - start;
      }
    }
5
54liuyao 已提交
3876 3877

    if (pWinInfo->winInfo.sessionWin.win.skey > pTs[i]) {
H
Haojun Liao 已提交
3878
      if (pSeDeleted && pWinInfo->winInfo.isOutput) {
5
54liuyao 已提交
3879
        saveDeleteRes(pSeDeleted, pWinInfo->winInfo.sessionWin);
5
54liuyao 已提交
3880
      }
5
54liuyao 已提交
3881 3882
      removeSessionResult(pSeUpdated, pResultRows, pWinInfo->winInfo.sessionWin);
      pWinInfo->winInfo.sessionWin.win.skey = pTs[i];
5
54liuyao 已提交
3883
    }
5
54liuyao 已提交
3884
    pWinInfo->winInfo.sessionWin.win.ekey = TMAX(pWinInfo->winInfo.sessionWin.win.ekey, pTs[i]);
5
54liuyao 已提交
3885 3886 3887 3888 3889 3890 3891
    if (!isEqualStateKey(pWinInfo, pKeyData)) {
      *allEqual = false;
    }
  }
  return rows - start;
}

5
54liuyao 已提交
3892 3893
static void doStreamStateAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBlock, SSHashObj* pSeUpdated,
                                 SSHashObj* pStDeleted) {
X
Xiaoyu Wang 已提交
3894
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
5
54liuyao 已提交
3895
  SStreamStateAggOperatorInfo* pInfo = pOperator->info;
3896
  int32_t                      numOfOutput = pOperator->exprSupp.numOfExprs;
H
Haojun Liao 已提交
3897
  uint64_t                     groupId = pSDataBlock->info.id.groupId;
X
Xiaoyu Wang 已提交
3898 3899 3900 3901
  int64_t                      code = TSDB_CODE_SUCCESS;
  TSKEY*                       tsCols = NULL;
  SResultRow*                  pResult = NULL;
  int32_t                      winRows = 0;
5
54liuyao 已提交
3902
  if (pSDataBlock->pDataBlock != NULL) {
X
Xiaoyu Wang 已提交
3903 3904
    SColumnInfoData* pColDataInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->primaryTsIndex);
    tsCols = (int64_t*)pColDataInfo->pData;
5
54liuyao 已提交
3905
  } else {
X
Xiaoyu Wang 已提交
3906
    return;
5
54liuyao 已提交
3907
  }
L
Liu Jicong 已提交
3908

5
54liuyao 已提交
3909
  SStreamAggSupporter* pAggSup = &pInfo->streamAggSup;
5
54liuyao 已提交
3910 3911
  int32_t              rows = pSDataBlock->info.rows;
  blockDataEnsureCapacity(pAggSup->pScanBlock, rows);
L
Liu Jicong 已提交
3912
  SColumnInfoData* pKeyColInfo = taosArrayGet(pSDataBlock->pDataBlock, pInfo->stateCol.slotId);
5
54liuyao 已提交
3913
  for (int32_t i = 0; i < rows; i += winRows) {
3914
    if (pInfo->ignoreExpiredData && isOverdue(tsCols[i], &pInfo->twAggSup) || colDataIsNull_s(pKeyColInfo, i)) {
5
54liuyao 已提交
3915 3916 3917
      i++;
      continue;
    }
5
54liuyao 已提交
3918 3919 3920 3921 3922 3923 3924 3925 3926
    char*            pKeyData = colDataGetData(pKeyColInfo, i);
    int32_t          winIndex = 0;
    bool             allEqual = true;
    SStateWindowInfo curWin = {0};
    SStateWindowInfo nextWin = {0};
    setStateOutputBuf(pAggSup, tsCols[i], groupId, pKeyData, &curWin, &nextWin);
    setSessionWinOutputInfo(pSeUpdated, &curWin.winInfo);
    winRows = updateStateWindowInfo(&curWin, &nextWin, tsCols, groupId, pKeyColInfo, rows, i, &allEqual,
                                    pAggSup->pResultRows, pSeUpdated, pStDeleted);
5
54liuyao 已提交
3927
    if (!allEqual) {
3928
      uint64_t uid = 0;
5
54liuyao 已提交
3929 3930 3931 3932 3933
      appendOneRowToStreamSpecialBlock(pAggSup->pScanBlock, &curWin.winInfo.sessionWin.win.skey,
                                       &curWin.winInfo.sessionWin.win.ekey, &uid, &groupId, NULL);
      tSimpleHashRemove(pSeUpdated, &curWin.winInfo.sessionWin, sizeof(SSessionKey));
      doDeleteSessionWindow(pAggSup, &curWin.winInfo.sessionWin);
      releaseOutputBuf(pAggSup->pState, NULL, (SResultRow*)curWin.winInfo.pOutputBuf);
5
54liuyao 已提交
3934 3935
      continue;
    }
5
54liuyao 已提交
3936 3937
    code = doOneWindowAggImpl(&pInfo->twAggSup.timeWindowData, &curWin.winInfo, &pResult, i, winRows, rows, numOfOutput,
                              pOperator);
5
54liuyao 已提交
3938
    if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
S
Shengliang Guan 已提交
3939
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
5
54liuyao 已提交
3940
    }
5
54liuyao 已提交
3941 3942
    saveSessionOutputBuf(pAggSup, &curWin.winInfo);

5
54liuyao 已提交
3943
    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
5
54liuyao 已提交
3944
      code = saveResult(curWin.winInfo, pSeUpdated);
5
54liuyao 已提交
3945
      if (code != TSDB_CODE_SUCCESS) {
S
Shengliang Guan 已提交
3946
        T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
5
54liuyao 已提交
3947 3948
      }
    }
3949 3950

    if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE) {
5
54liuyao 已提交
3951 3952
      SSessionKey key = {0};
      getSessionHashKey(&curWin.winInfo.sessionWin, &key);
3953 3954
      tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo, sizeof(SResultWindowInfo));
    }
5
54liuyao 已提交
3955 3956 3957 3958 3959 3960 3961 3962
  }
}

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

3963
  SExprSupp*                   pSup = &pOperator->exprSupp;
5
54liuyao 已提交
3964
  SStreamStateAggOperatorInfo* pInfo = pOperator->info;
X
Xiaoyu Wang 已提交
3965
  SOptrBasicInfo*              pBInfo = &pInfo->binfo;
5
54liuyao 已提交
3966
  if (pOperator->status == OP_RES_TO_RETURN) {
3967
    doBuildDeleteDataBlock(pOperator, pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
5
54liuyao 已提交
3968
    if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
3969
      printDataBlock(pInfo->pDelRes, "single state delete");
5
54liuyao 已提交
3970 3971
      return pInfo->pDelRes;
    }
5
54liuyao 已提交
3972 3973 3974 3975 3976

    doBuildSessionResult(pOperator, pInfo->streamAggSup.pState, &pInfo->groupResInfo, pBInfo->pRes);
    if (pBInfo->pRes->info.rows > 0) {
      printDataBlock(pBInfo->pRes, "single state");
      return pBInfo->pRes;
5
54liuyao 已提交
3977
    }
5
54liuyao 已提交
3978

H
Haojun Liao 已提交
3979
    setOperatorCompleted(pOperator);
5
54liuyao 已提交
3980
    return NULL;
5
54liuyao 已提交
3981 3982 3983
  }

  SOperatorInfo* downstream = pOperator->pDownstream[0];
5
54liuyao 已提交
3984 3985 3986 3987
  if (!pInfo->pUpdated) {
    pInfo->pUpdated = taosArrayInit(16, sizeof(SSessionKey));
  }
  if (!pInfo->pSeUpdated) {
3988
    _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
5
54liuyao 已提交
3989 3990
    pInfo->pSeUpdated = tSimpleHashInit(64, hashFn);
  }
5
54liuyao 已提交
3991 3992 3993 3994 3995
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
      break;
    }
5
54liuyao 已提交
3996
    printDataBlock(pBlock, "single state recv");
3997

5
54liuyao 已提交
3998 3999 4000 4001
    if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
        pBlock->info.type == STREAM_CLEAR) {
      SArray* pWins = taosArrayInit(16, sizeof(SSessionKey));
      doDeleteTimeWindows(&pInfo->streamAggSup, pBlock, pWins);
5
54liuyao 已提交
4002
      removeSessionResults(pInfo->pSeUpdated, pWins);
5
54liuyao 已提交
4003
      copyDeleteWindowInfo(pWins, pInfo->pSeDeleted);
4004 4005
      taosArrayDestroy(pWins);
      continue;
4006
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
4007
      getAllSessionWindow(pInfo->streamAggSup.pResultRows, pInfo->pSeUpdated);
5
54liuyao 已提交
4008
      continue;
5
54liuyao 已提交
4009 4010 4011 4012
    } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
      return pBlock;
    } else {
      ASSERTS(pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_INVALID, "invalid SSDataBlock type");
5
54liuyao 已提交
4013
    }
4014

5
54liuyao 已提交
4015 4016 4017 4018
    if (pInfo->scalarSupp.pExprInfo != NULL) {
      SExprSupp* pExprSup = &pInfo->scalarSupp;
      projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
    }
4019
    // the pDataBlock are always the same one, no need to call this again
4020
    setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
5
54liuyao 已提交
4021 4022
    doStreamStateAggImpl(pOperator, pBlock, pInfo->pSeUpdated, pInfo->pSeDeleted);
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
5
54liuyao 已提交
4023 4024 4025
  }
  // restore the value
  pOperator->status = OP_RES_TO_RETURN;
X
Xiaoyu Wang 已提交
4026

5
54liuyao 已提交
4027 4028 4029 4030 4031
  closeSessionWindow(pInfo->streamAggSup.pResultRows, &pInfo->twAggSup, pInfo->pSeUpdated);
  copyUpdateResult(pInfo->pSeUpdated, pInfo->pUpdated);
  removeSessionResults(pInfo->pSeDeleted, pInfo->pUpdated);
  tSimpleHashCleanup(pInfo->pSeUpdated);
  pInfo->pSeUpdated = NULL;
5
54liuyao 已提交
4032

5
54liuyao 已提交
4033 4034
  initGroupResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);
  pInfo->pUpdated = NULL;
5
54liuyao 已提交
4035
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
5
54liuyao 已提交
4036

5
54liuyao 已提交
4037 4038 4039 4040 4041 4042
#if 0
  char* pBuf = streamStateSessionDump(pInfo->streamAggSup.pState);
  qDebug("===stream===final session%s", pBuf);
  taosMemoryFree(pBuf);
#endif

4043
  doBuildDeleteDataBlock(pOperator, pInfo->pSeDeleted, pInfo->pDelRes, &pInfo->pDelIterator);
5
54liuyao 已提交
4044
  if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
4045
    printDataBlock(pInfo->pDelRes, "single state delete");
5
54liuyao 已提交
4046 4047 4048
    return pInfo->pDelRes;
  }

5
54liuyao 已提交
4049 4050 4051 4052 4053
  doBuildSessionResult(pOperator, pInfo->streamAggSup.pState, &pInfo->groupResInfo, pBInfo->pRes);
  if (pBInfo->pRes->info.rows > 0) {
    printDataBlock(pBInfo->pRes, "single state");
    return pBInfo->pRes;
  }
H
Haojun Liao 已提交
4054
  setOperatorCompleted(pOperator);
5
54liuyao 已提交
4055
  return NULL;
4056 4057
}

X
Xiaoyu Wang 已提交
4058 4059 4060 4061 4062
SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                SExecTaskInfo* pTaskInfo) {
  SStreamStateWinodwPhysiNode* pStateNode = (SStreamStateWinodwPhysiNode*)pPhyNode;
  int32_t                      tsSlotId = ((SColumnNode*)pStateNode->window.pTspk)->slotId;
  SColumnNode*                 pColNode = (SColumnNode*)((STargetNode*)pStateNode->pStateKey)->pExpr;
H
Haojun Liao 已提交
4063
  int32_t                      code = TSDB_CODE_SUCCESS;
5
54liuyao 已提交
4064

X
Xiaoyu Wang 已提交
4065 4066
  SStreamStateAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamStateAggOperatorInfo));
  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
5
54liuyao 已提交
4067
  if (pInfo == NULL || pOperator == NULL) {
H
Haojun Liao 已提交
4068
    code = TSDB_CODE_OUT_OF_MEMORY;
5
54liuyao 已提交
4069 4070 4071 4072
    goto _error;
  }

  pInfo->stateCol = extractColumnFromColumnNode(pColNode);
4073
  initResultSizeInfo(&pOperator->resultInfo, 4096);
5
54liuyao 已提交
4074 4075 4076
  if (pStateNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pStateNode->window.pExprs, NULL, &numOfScalar);
H
Haojun Liao 已提交
4077
    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
5
54liuyao 已提交
4078 4079 4080 4081 4082
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }

X
Xiaoyu Wang 已提交
4083 4084
  pInfo->twAggSup = (STimeWindowAggSupp){
      .waterMark = pStateNode->window.watermark,
5
54liuyao 已提交
4085 4086
      .calTrigger = pStateNode->window.triggerType,
      .maxTs = INT64_MIN,
5
54liuyao 已提交
4087
      .minTs = INT64_MAX,
X
Xiaoyu Wang 已提交
4088
  };
4089

5
54liuyao 已提交
4090
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
4091

5
54liuyao 已提交
4092 4093 4094
  SExprSupp*   pSup = &pOperator->exprSupp;
  int32_t      numOfCols = 0;
  SExprInfo*   pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &numOfCols);
H
Haojun Liao 已提交
4095
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);
4096
  code = initBasicInfoEx(&pInfo->binfo, pSup, pExprInfo, numOfCols, pResBlock);
5
54liuyao 已提交
4097 4098 4099
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
5
54liuyao 已提交
4100 4101 4102 4103
  int32_t keySize = sizeof(SStateKeys) + pColNode->node.resType.bytes;
  int16_t type = pColNode->node.resType.type;
  code = initStreamAggSupporter(&pInfo->streamAggSup, pSup->pCtx, numOfCols, 0, pTaskInfo->streamInfo.pState, keySize,
                                type);
5
54liuyao 已提交
4104 4105 4106 4107 4108 4109
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  pInfo->primaryTsIndex = tsSlotId;
  _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
5
54liuyao 已提交
4110
  pInfo->pSeDeleted = tSimpleHashInit(64, hashFn);
5
54liuyao 已提交
4111
  pInfo->pDelIterator = NULL;
H
Haojun Liao 已提交
4112
  pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT);
5
54liuyao 已提交
4113
  pInfo->pChildren = NULL;
5
54liuyao 已提交
4114
  pInfo->ignoreExpiredData = pStateNode->window.igExpired;
4115
  pInfo->ignoreExpiredDataSaved = false;
5
54liuyao 已提交
4116 4117
  pInfo->pUpdated = NULL;
  pInfo->pSeUpdated = NULL;
5
54liuyao 已提交
4118

L
Liu Jicong 已提交
4119 4120
  setOperatorInfo(pOperator, "StreamStateAggOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE, true, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
L
Liu Jicong 已提交
4121 4122
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo,
                                         optrDefaultBufFn, NULL);
5
54liuyao 已提交
4123
  initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
5
54liuyao 已提交
4124 4125 4126 4127 4128 4129 4130
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
  return pOperator;

_error:
4131
  destroyStreamStateOperatorInfo(pInfo);
5
54liuyao 已提交
4132 4133 4134 4135
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
4136

4137
void destroyMAIOperatorInfo(void* param) {
4138
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = (SMergeAlignedIntervalAggOperatorInfo*)param;
4139
  destroyIntervalOperatorInfo(miaInfo->intervalAggOperatorInfo);
D
dapan1121 已提交
4140
  taosMemoryFreeClear(param);
4141 4142
}

4143
static SResultRow* doSetSingleOutputTupleBuf(SResultRowInfo* pResultRowInfo, SAggSupporter* pSup) {
H
Haojun Liao 已提交
4144
  SResultRow* pResult = getNewResultRow(pSup->pResultBuf, &pSup->currentPageId, pSup->resultRowSize);
4145 4146 4147
  if (NULL == pResult) {
    return pResult;
  }
H
Haojun Liao 已提交
4148
  pResultRowInfo->cur = (SResultRowPosition){.pageId = pResult->pageId, .offset = pResult->offset};
4149 4150
  return pResult;
}
4151

4152 4153 4154 4155 4156 4157 4158 4159
static int32_t setSingleOutputTupleBuf(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,
                                       SExprSupp* pExprSup, SAggSupporter* pAggSup) {
  if (*pResult == NULL) {
    *pResult = doSetSingleOutputTupleBuf(pResultRowInfo, pAggSup);
    if (*pResult == NULL) {
      return terrno;
    }
  }
4160

4161
  // set time window for current result
4162 4163
  (*pResult)->win = (*win);
  setResultRowInitCtx((*pResult), pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);
4164
  return TSDB_CODE_SUCCESS;
4165 4166
}

4167
static void doMergeAlignedIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo,
4168
                                          SSDataBlock* pBlock, SSDataBlock* pResultBlock) {
4169
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
D
dapan1121 已提交
4170
  SIntervalAggOperatorInfo*             iaInfo = miaInfo->intervalAggOperatorInfo;
4171 4172

  SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
4173
  SExprSupp*     pSup = &pOperatorInfo->exprSupp;
4174
  SInterval*     pInterval = &iaInfo->interval;
4175

5
54liuyao 已提交
4176 4177
  int32_t  startPos = 0;
  int64_t* tsCols = extractTsCol(pBlock, iaInfo);
4178

4179 4180
  TSKEY ts = getStartTsKey(&pBlock->info.window, tsCols);

4181 4182
  // there is an result exists
  if (miaInfo->curTs != INT64_MIN) {
4183
    if (ts != miaInfo->curTs) {
4184
      finalizeResultRows(iaInfo->aggSup.pResultBuf, &pResultRowInfo->cur, pSup, pResultBlock, pTaskInfo);
4185
      resetResultRow(miaInfo->pResultRow, iaInfo->aggSup.resultRowSize - sizeof(SResultRow));
4186
      miaInfo->curTs = ts;
4187
    }
4188 4189
  } else {
    miaInfo->curTs = ts;
4190 4191 4192
  }

  STimeWindow win = {0};
4193
  win.skey = miaInfo->curTs;
4194
  win.ekey = taosTimeAdd(win.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
4195

5
54liuyao 已提交
4196
  int32_t ret = setSingleOutputTupleBuf(pResultRowInfo, &win, &miaInfo->pResultRow, pSup, &iaInfo->aggSup);
4197 4198
  if (ret != TSDB_CODE_SUCCESS || miaInfo->pResultRow == NULL) {
    T_LONG_JMP(pTaskInfo->env, ret);
4199 4200
  }

4201 4202
  int32_t currPos = startPos;

4203
  STimeWindow currWin = win;
4204
  while (++currPos < pBlock->info.rows) {
4205
    if (tsCols[currPos] == miaInfo->curTs) {
4206
      continue;
4207 4208 4209
    }

    updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true);
H
Haojun Liao 已提交
4210 4211
    applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &iaInfo->twAggSup.timeWindowData, startPos,
                                    currPos - startPos, pBlock->info.rows, pSup->numOfExprs);
4212

4213
    finalizeResultRows(iaInfo->aggSup.pResultBuf, &pResultRowInfo->cur, pSup, pResultBlock, pTaskInfo);
4214
    resetResultRow(miaInfo->pResultRow, iaInfo->aggSup.resultRowSize - sizeof(SResultRow));
4215
    miaInfo->curTs = tsCols[currPos];
4216

4217
    currWin.skey = miaInfo->curTs;
4218
    currWin.ekey = taosTimeAdd(currWin.skey, pInterval->interval, pInterval->intervalUnit, pInterval->precision) - 1;
4219 4220

    startPos = currPos;
5
54liuyao 已提交
4221
    ret = setSingleOutputTupleBuf(pResultRowInfo, &win, &miaInfo->pResultRow, pSup, &iaInfo->aggSup);
4222 4223
    if (ret != TSDB_CODE_SUCCESS || miaInfo->pResultRow == NULL) {
      T_LONG_JMP(pTaskInfo->env, ret);
4224
    }
4225 4226

    miaInfo->curTs = currWin.skey;
4227
  }
4228

4229
  updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &currWin, true);
H
Haojun Liao 已提交
4230
  applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &iaInfo->twAggSup.timeWindowData, startPos, currPos - startPos,
L
Liu Jicong 已提交
4231
                                  pBlock->info.rows, pSup->numOfExprs);
4232 4233
}

4234
static void cleanupAfterGroupResultGen(SMergeAlignedIntervalAggOperatorInfo* pMiaInfo, SSDataBlock* pRes) {
H
Haojun Liao 已提交
4235
  pRes->info.id.groupId = pMiaInfo->groupId;
4236 4237
  pMiaInfo->curTs = INT64_MIN;
  pMiaInfo->groupId = 0;
4238 4239
}

4240
static void doMergeAlignedIntervalAgg(SOperatorInfo* pOperator) {
S
shenglian zhou 已提交
4241
  SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
4242

4243 4244
  SMergeAlignedIntervalAggOperatorInfo* pMiaInfo = pOperator->info;
  SIntervalAggOperatorInfo*             pIaInfo = pMiaInfo->intervalAggOperatorInfo;
4245

4246 4247 4248 4249 4250
  SExprSupp*      pSup = &pOperator->exprSupp;
  SSDataBlock*    pRes = pIaInfo->binfo.pRes;
  SResultRowInfo* pResultRowInfo = &pIaInfo->binfo.resultRowInfo;
  SOperatorInfo*  downstream = pOperator->pDownstream[0];
  int32_t         scanFlag = MAIN_SCAN;
4251

4252 4253
  while (1) {
    SSDataBlock* pBlock = NULL;
4254
    if (pMiaInfo->prefetchedBlock == NULL) {
4255 4256
      pBlock = downstream->fpSet.getNextFn(downstream);
    } else {
4257 4258
      pBlock = pMiaInfo->prefetchedBlock;
      pMiaInfo->prefetchedBlock = NULL;
4259

H
Haojun Liao 已提交
4260
      pMiaInfo->groupId = pBlock->info.id.groupId;
4261
    }
4262

4263
    // no data exists, all query processing is done
4264
    if (pBlock == NULL) {
4265 4266 4267
      // close last unclosed time window
      if (pMiaInfo->curTs != INT64_MIN) {
        finalizeResultRows(pIaInfo->aggSup.pResultBuf, &pResultRowInfo->cur, pSup, pRes, pTaskInfo);
4268 4269
        resetResultRow(pMiaInfo->pResultRow, pIaInfo->aggSup.resultRowSize - sizeof(SResultRow));
        cleanupAfterGroupResultGen(pMiaInfo, pRes);
4270
      }
4271

H
Haojun Liao 已提交
4272
      setOperatorCompleted(pOperator);
4273
      break;
4274
    }
4275

H
Haojun Liao 已提交
4276
    if (pMiaInfo->groupId == 0) {
H
Haojun Liao 已提交
4277 4278 4279
      if (pMiaInfo->groupId != pBlock->info.id.groupId) {
        pMiaInfo->groupId = pBlock->info.id.groupId;
        pRes->info.id.groupId = pMiaInfo->groupId;
H
Haojun Liao 已提交
4280 4281
      }
    } else {
H
Haojun Liao 已提交
4282
      if (pMiaInfo->groupId != pBlock->info.id.groupId) {
H
Haojun Liao 已提交
4283
        // if there are unclosed time window, close it firstly.
4284
        ASSERT(pMiaInfo->curTs != INT64_MIN);
H
Haojun Liao 已提交
4285
        finalizeResultRows(pIaInfo->aggSup.pResultBuf, &pResultRowInfo->cur, pSup, pRes, pTaskInfo);
4286
        resetResultRow(pMiaInfo->pResultRow, pIaInfo->aggSup.resultRowSize - sizeof(SResultRow));
H
Haojun Liao 已提交
4287

4288 4289
        pMiaInfo->prefetchedBlock = pBlock;
        cleanupAfterGroupResultGen(pMiaInfo, pRes);
H
Haojun Liao 已提交
4290
        break;
5
54liuyao 已提交
4291
      } else {
H
Haojun Liao 已提交
4292
        // continue
H
Haojun Liao 已提交
4293
        pRes->info.id.groupId = pMiaInfo->groupId;
H
Haojun Liao 已提交
4294
      }
4295
    }
4296

4297
    getTableScanInfo(pOperator, &pIaInfo->inputOrder, &scanFlag, false);
4298
    setInputDataBlock(pSup, pBlock, pIaInfo->inputOrder, scanFlag, true);
4299
    doMergeAlignedIntervalAggImpl(pOperator, &pIaInfo->binfo.resultRowInfo, pBlock, pRes);
4300

H
Haojun Liao 已提交
4301
    doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);
4302 4303 4304
    if (pRes->info.rows >= pOperator->resultInfo.capacity) {
      break;
    }
4305
  }
4306 4307 4308 4309 4310 4311 4312 4313 4314 4315 4316 4317 4318 4319 4320
}

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

  SMergeAlignedIntervalAggOperatorInfo* pMiaInfo = pOperator->info;
  SIntervalAggOperatorInfo*             iaInfo = pMiaInfo->intervalAggOperatorInfo;
  if (pOperator->status == OP_EXEC_DONE) {
    return NULL;
  }

  SSDataBlock* pRes = iaInfo->binfo.pRes;
  blockDataCleanup(pRes);

  if (iaInfo->binfo.mergeResultBlock) {
dengyihao's avatar
dengyihao 已提交
4321
    while (1) {
4322
      if (pOperator->status == OP_EXEC_DONE) {
4323 4324
        break;
      }
4325

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

4330 4331 4332 4333
      doMergeAlignedIntervalAgg(pOperator);
    }
  } else {
    doMergeAlignedIntervalAgg(pOperator);
4334 4335 4336 4337 4338 4339 4340
  }

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

4341
SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, SMergeAlignedIntervalPhysiNode* pNode,
4342
                                                      SExecTaskInfo* pTaskInfo) {
4343
  SMergeAlignedIntervalAggOperatorInfo* miaInfo = taosMemoryCalloc(1, sizeof(SMergeAlignedIntervalAggOperatorInfo));
4344
  SOperatorInfo*                        pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
4345 4346 4347 4348
  if (miaInfo == NULL || pOperator == NULL) {
    goto _error;
  }

D
dapan1121 已提交
4349 4350 4351 4352 4353
  miaInfo->intervalAggOperatorInfo = taosMemoryCalloc(1, sizeof(SIntervalAggOperatorInfo));
  if (miaInfo->intervalAggOperatorInfo == NULL) {
    goto _error;
  }

4354 4355 4356 4357 4358 4359 4360
  SInterval interval = {.interval = pNode->interval,
                        .sliding = pNode->sliding,
                        .intervalUnit = pNode->intervalUnit,
                        .slidingUnit = pNode->slidingUnit,
                        .offset = pNode->offset,
                        .precision = ((SColumnNode*)pNode->window.pTspk)->node.resType.precision};

D
dapan1121 已提交
4361
  SIntervalAggOperatorInfo* iaInfo = miaInfo->intervalAggOperatorInfo;
4362
  SExprSupp*                pSup = &pOperator->exprSupp;
4363

H
Haojun Liao 已提交
4364 4365 4366 4367 4368
  int32_t code = filterInitFromNode((SNode*)pNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

L
Liu Jicong 已提交
4369 4370 4371 4372
  miaInfo->curTs = INT64_MIN;
  iaInfo->win = pTaskInfo->window;
  iaInfo->inputOrder = TSDB_ORDER_ASC;
  iaInfo->interval = interval;
4373 4374
  iaInfo->primaryTsIndex = ((SColumnNode*)pNode->window.pTspk)->slotId;
  iaInfo->binfo.mergeResultBlock = pNode->window.mergeDataBlock;
4375 4376

  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
H
Haojun Liao 已提交
4377
  initResultSizeInfo(&pOperator->resultInfo, 512);
4378

H
Haojun Liao 已提交
4379 4380
  int32_t    num = 0;
  SExprInfo* pExprInfo = createExprInfo(pNode->window.pFuncs, NULL, &num);
H
Haojun Liao 已提交
4381

L
Liu Jicong 已提交
4382 4383
  code = initAggSup(&pOperator->exprSupp, &iaInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,
                    pTaskInfo->streamInfo.pState);
4384 4385 4386
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
4387

H
Haojun Liao 已提交
4388
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pNode->window.node.pOutputDataBlockDesc);
4389
  initBasicInfo(&iaInfo->binfo, pResBlock);
4390
  initExecTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &iaInfo->win);
4391

4392
  iaInfo->timeWindowInterpo = timeWindowinterpNeeded(pSup->pCtx, num, iaInfo);
4393
  if (iaInfo->timeWindowInterpo) {
4394
    iaInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SOpenWindowInfo));
4395 4396
  }

4397
  initResultRowInfo(&iaInfo->binfo.resultRowInfo);
4398
  blockDataEnsureCapacity(iaInfo->binfo.pRes, pOperator->resultInfo.capacity);
L
Liu Jicong 已提交
4399 4400
  setOperatorInfo(pOperator, "TimeMergeAlignedIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL,
                  false, OP_NOT_OPENED, miaInfo, pTaskInfo);
4401

L
Liu Jicong 已提交
4402 4403
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, mergeAlignedIntervalAgg, NULL, destroyMAIOperatorInfo,
                                         optrDefaultBufFn, NULL);
4404 4405 4406 4407 4408 4409 4410 4411 4412

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

  return pOperator;

_error:
4413
  destroyMAIOperatorInfo(miaInfo);
4414 4415 4416 4417
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}
4418 4419 4420 4421 4422

//=====================================================================================================================
// merge interval operator
typedef struct SMergeIntervalAggOperatorInfo {
  SIntervalAggOperatorInfo intervalAggOperatorInfo;
L
Liu Jicong 已提交
4423 4424 4425 4426 4427 4428
  SList*                   groupIntervals;
  SListIter                groupIntervalsIter;
  bool                     hasGroupId;
  uint64_t                 groupId;
  SSDataBlock*             prefetchedBlock;
  bool                     inputBlocksFinished;
4429 4430
} SMergeIntervalAggOperatorInfo;

S
slzhou 已提交
4431
typedef struct SGroupTimeWindow {
L
Liu Jicong 已提交
4432
  uint64_t    groupId;
S
slzhou 已提交
4433 4434 4435
  STimeWindow window;
} SGroupTimeWindow;

4436
void destroyMergeIntervalOperatorInfo(void* param) {
4437
  SMergeIntervalAggOperatorInfo* miaInfo = (SMergeIntervalAggOperatorInfo*)param;
S
slzhou 已提交
4438
  tdListFree(miaInfo->groupIntervals);
4439
  destroyIntervalOperatorInfo(&miaInfo->intervalAggOperatorInfo);
4440

D
dapan1121 已提交
4441
  taosMemoryFreeClear(param);
4442 4443
}

L
Liu Jicong 已提交
4444 4445
static int32_t finalizeWindowResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, STimeWindow* win,
                                    SSDataBlock* pResultBlock) {
4446 4447 4448
  SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
  SIntervalAggOperatorInfo*      iaInfo = &miaInfo->intervalAggOperatorInfo;
  SExecTaskInfo*                 pTaskInfo = pOperatorInfo->pTaskInfo;
4449
  bool                           ascScan = (iaInfo->inputOrder == TSDB_ORDER_ASC);
4450 4451 4452
  SExprSupp*                     pExprSup = &pOperatorInfo->exprSupp;

  SET_RES_WINDOW_KEY(iaInfo->aggSup.keyBuf, &win->skey, TSDB_KEYSIZE, tableGroupId);
L
Liu Jicong 已提交
4453 4454
  SResultRowPosition* p1 = (SResultRowPosition*)tSimpleHashGet(
      iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf, GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
4455
  ASSERT(p1 != NULL);
5
54liuyao 已提交
4456
  //  finalizeResultRows(iaInfo->aggSup.pResultBuf, p1, pResultBlock, pTaskInfo);
4457
  tSimpleHashRemove(iaInfo->aggSup.pResultRowHashTable, iaInfo->aggSup.keyBuf, GET_RES_WINDOW_KEY_LEN(TSDB_KEYSIZE));
4458 4459 4460
  return TSDB_CODE_SUCCESS;
}

4461 4462 4463 4464
static int32_t outputPrevIntervalResult(SOperatorInfo* pOperatorInfo, uint64_t tableGroupId, SSDataBlock* pResultBlock,
                                        STimeWindow* newWin) {
  SMergeIntervalAggOperatorInfo* miaInfo = pOperatorInfo->info;
  SIntervalAggOperatorInfo*      iaInfo = &miaInfo->intervalAggOperatorInfo;
4465
  bool                           ascScan = (iaInfo->inputOrder == TSDB_ORDER_ASC);
4466

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

S
slzhou 已提交
4470 4471 4472 4473 4474
  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 已提交
4475
    if (prevGrpWin->groupId != tableGroupId) {
S
slzhou 已提交
4476 4477
      continue;
    }
4478

S
slzhou 已提交
4479
    STimeWindow* prevWin = &prevGrpWin->window;
H
Haojun Liao 已提交
4480
    if ((ascScan && newWin->skey > prevWin->ekey) || ((!ascScan) && newWin->skey < prevWin->ekey)) {
5
54liuyao 已提交
4481
      //      finalizeWindowResult(pOperatorInfo, tableGroupId, prevWin, pResultBlock);
S
slzhou 已提交
4482 4483
      tdListPopNode(miaInfo->groupIntervals, listNode);
    }
4484 4485 4486 4487 4488 4489 4490 4491 4492 4493 4494 4495 4496 4497 4498 4499
  }

  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);
H
Haojun Liao 已提交
4500
  uint64_t    tableGroupId = pBlock->info.id.groupId;
4501
  bool        ascScan = (iaInfo->inputOrder == TSDB_ORDER_ASC);
4502 4503 4504
  TSKEY       blockStartTs = getStartTsKey(&pBlock->info.window, tsCols);
  SResultRow* pResult = NULL;

4505 4506
  STimeWindow win = getActiveTimeWindow(iaInfo->aggSup.pResultBuf, pResultRowInfo, blockStartTs, &iaInfo->interval,
                                        iaInfo->inputOrder);
4507 4508 4509 4510 4511

  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) {
S
Shengliang Guan 已提交
4512
    T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
4513 4514 4515 4516
  }

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

  // prev time window not interpolation yet.
  if (iaInfo->timeWindowInterpo) {
4522
    SResultRowPosition pos = addToOpenWindowList(pResultRowInfo, pResult, tableGroupId);
4523 4524 4525 4526 4527 4528
    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) {
S
Shengliang Guan 已提交
4529
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
4530 4531 4532 4533 4534 4535 4536
    }

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

  updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &win, true);
H
Haojun Liao 已提交
4537
  applyAggFunctionOnPartialTuples(pTaskInfo, pExprSup->pCtx, &iaInfo->twAggSup.timeWindowData, startPos, forwardRows,
L
Liu Jicong 已提交
4538
                                  pBlock->info.rows, numOfOutput);
4539 4540 4541 4542 4543 4544 4545 4546
  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;
4547 4548
    startPos =
        getNextQualifiedWindow(&iaInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, iaInfo->inputOrder);
4549 4550 4551 4552 4553 4554 4555 4556 4557
    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) {
S
Shengliang Guan 已提交
4558
      T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY);
4559 4560 4561 4562
    }

    ekey = ascScan ? nextWin.ekey : nextWin.skey;
    forwardRows =
4563
        getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, iaInfo->inputOrder);
4564 4565 4566 4567 4568

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

    updateTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &nextWin, true);
H
Haojun Liao 已提交
4569
    applyAggFunctionOnPartialTuples(pTaskInfo, pExprSup->pCtx, &iaInfo->twAggSup.timeWindowData, startPos, forwardRows,
L
Liu Jicong 已提交
4570
                                    pBlock->info.rows, numOfOutput);
4571 4572 4573 4574 4575 4576 4577 4578 4579 4580 4581 4582 4583 4584 4585 4586 4587 4588 4589 4590 4591 4592 4593 4594 4595 4596 4597 4598 4599 4600 4601 4602 4603 4604 4605
    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;
H
Haojun Liao 已提交
4606
        miaInfo->groupId = pBlock->info.id.groupId;
4607 4608 4609 4610
        miaInfo->prefetchedBlock = NULL;
      }

      if (pBlock == NULL) {
S
slzhou 已提交
4611
        tdListInitIter(miaInfo->groupIntervals, &miaInfo->groupIntervalsIter, TD_LIST_FORWARD);
4612 4613 4614 4615 4616 4617
        miaInfo->inputBlocksFinished = true;
        break;
      }

      if (!miaInfo->hasGroupId) {
        miaInfo->hasGroupId = true;
H
Haojun Liao 已提交
4618 4619
        miaInfo->groupId = pBlock->info.id.groupId;
      } else if (miaInfo->groupId != pBlock->info.id.groupId) {
4620 4621 4622 4623
        miaInfo->prefetchedBlock = pBlock;
        break;
      }

4624
      getTableScanInfo(pOperator, &iaInfo->inputOrder, &scanFlag, false);
4625
      setInputDataBlock(pExpSupp, pBlock, iaInfo->inputOrder, scanFlag, true);
4626 4627 4628 4629 4630 4631 4632
      doMergeIntervalAggImpl(pOperator, &iaInfo->binfo.resultRowInfo, pBlock, scanFlag, pRes);

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

H
Haojun Liao 已提交
4633
    pRes->info.id.groupId = miaInfo->groupId;
4634 4635 4636
  }

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

S
slzhou 已提交
4639 4640
    if (listNode != NULL) {
      SGroupTimeWindow* grpWin = (SGroupTimeWindow*)(listNode->data);
5
54liuyao 已提交
4641
      //      finalizeWindowResult(pOperator, grpWin->groupId, &grpWin->window, pRes);
H
Haojun Liao 已提交
4642
      pRes->info.id.groupId = grpWin->groupId;
4643 4644 4645 4646
    }
  }

  if (pRes->info.rows == 0) {
H
Haojun Liao 已提交
4647
    setOperatorCompleted(pOperator);
4648 4649 4650 4651 4652 4653 4654
  }

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

4655 4656 4657
SOperatorInfo* createMergeIntervalOperatorInfo(SOperatorInfo* downstream, SMergeIntervalPhysiNode* pIntervalPhyNode,
                                               SExecTaskInfo* pTaskInfo) {
  SMergeIntervalAggOperatorInfo* pMergeIntervalInfo = taosMemoryCalloc(1, sizeof(SMergeIntervalAggOperatorInfo));
4658
  SOperatorInfo*                 pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
4659
  if (pMergeIntervalInfo == NULL || pOperator == NULL) {
4660 4661 4662
    goto _error;
  }

5
54liuyao 已提交
4663 4664
  int32_t    num = 0;
  SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &num);
4665 4666 4667 4668 4669 4670 4671

  SInterval interval = {.interval = pIntervalPhyNode->interval,
                        .sliding = pIntervalPhyNode->sliding,
                        .intervalUnit = pIntervalPhyNode->intervalUnit,
                        .slidingUnit = pIntervalPhyNode->slidingUnit,
                        .offset = pIntervalPhyNode->offset,
                        .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision};
4672

4673
  pMergeIntervalInfo->groupIntervals = tdListNew(sizeof(SGroupTimeWindow));
4674

4675
  SIntervalAggOperatorInfo* pIntervalInfo = &pMergeIntervalInfo->intervalAggOperatorInfo;
L
Liu Jicong 已提交
4676
  pIntervalInfo->win = pTaskInfo->window;
4677
  pIntervalInfo->inputOrder = TSDB_ORDER_ASC;
L
Liu Jicong 已提交
4678
  pIntervalInfo->interval = interval;
4679 4680
  pIntervalInfo->binfo.mergeResultBlock = pIntervalPhyNode->window.mergeDataBlock;
  pIntervalInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
4681 4682 4683 4684

  SExprSupp* pExprSupp = &pOperator->exprSupp;

  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
4685
  initResultSizeInfo(&pOperator->resultInfo, 4096);
4686

L
Liu Jicong 已提交
4687 4688
  int32_t code = initAggSup(pExprSupp, &pIntervalInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,
                            pTaskInfo->streamInfo.pState);
4689 4690 4691
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }
4692

H
Haojun Liao 已提交
4693
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pIntervalPhyNode->window.node.pOutputDataBlockDesc);
4694 4695
  initBasicInfo(&pIntervalInfo->binfo, pResBlock);
  initExecTimeWindowInfo(&pIntervalInfo->twAggSup.timeWindowData, &pIntervalInfo->win);
4696

4697 4698
  pIntervalInfo->timeWindowInterpo = timeWindowinterpNeeded(pExprSupp->pCtx, num, pIntervalInfo);
  if (pIntervalInfo->timeWindowInterpo) {
4699
    pIntervalInfo->binfo.resultRowInfo.openWindow = tdListNew(sizeof(SOpenWindowInfo));
4700
    if (pIntervalInfo->binfo.resultRowInfo.openWindow == NULL) {
4701 4702 4703 4704
      goto _error;
    }
  }

4705
  initResultRowInfo(&pIntervalInfo->binfo.resultRowInfo);
L
Liu Jicong 已提交
4706 4707
  setOperatorInfo(pOperator, "TimeMergeIntervalAggOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_INTERVAL, false,
                  OP_NOT_OPENED, pMergeIntervalInfo, pTaskInfo);
L
Liu Jicong 已提交
4708 4709
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doMergeIntervalAgg, NULL, destroyMergeIntervalOperatorInfo,
                                         optrDefaultBufFn, NULL);
4710 4711 4712 4713 4714 4715 4716 4717 4718

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

  return pOperator;

_error:
H
Haojun Liao 已提交
4719 4720 4721 4722
  if (pMergeIntervalInfo != NULL) {
    destroyMergeIntervalOperatorInfo(pMergeIntervalInfo);
  }

4723 4724 4725
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
L
Liu Jicong 已提交
4726
}
4727 4728 4729

static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
  SStreamIntervalOperatorInfo* pInfo = pOperator->info;
4730 4731
  SExecTaskInfo*               pTaskInfo = pOperator->pTaskInfo;
  SExprSupp*                   pSup = &pOperator->exprSupp;
4732 4733 4734 4735 4736 4737

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

  if (pOperator->status == OP_RES_TO_RETURN) {
4738
    doBuildDeleteResult(pInfo, pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
4739
    if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
4740
      printDataBlock(pInfo->pDelRes, "single interval delete");
4741 4742 4743
      return pInfo->pDelRes;
    }

4744
    doBuildStreamIntervalResult(pOperator, pInfo->pState, pInfo->binfo.pRes, &pInfo->groupResInfo);
5
54liuyao 已提交
4745 4746 4747
    if (pInfo->binfo.pRes->info.rows > 0) {
      printDataBlock(pInfo->binfo.pRes, "single interval");
      return pInfo->binfo.pRes;
4748
    }
4749 4750
    deleteIntervalDiscBuf(pInfo->pState, NULL, pInfo->twAggSup.maxTs - pInfo->twAggSup.deleteMark, &pInfo->interval,
                          &pInfo->delKey);
H
Haojun Liao 已提交
4751
    setOperatorCompleted(pOperator);
L
Liu Jicong 已提交
4752
    streamStateCommit(pTaskInfo->streamInfo.pState);
5
54liuyao 已提交
4753
    return NULL;
4754 4755 4756 4757
  }

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

5
54liuyao 已提交
4758
  if (!pInfo->pUpdated) {
4759
    pInfo->pUpdated = taosArrayInit(4, sizeof(SWinKey));
5
54liuyao 已提交
4760 4761 4762
  }
  if (!pInfo->pUpdatedMap) {
    _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
4763
    pInfo->pUpdatedMap = tSimpleHashInit(1024, hashFn);
5
54liuyao 已提交
4764 4765
  }

4766 4767 4768
  while (1) {
    SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
    if (pBlock == NULL) {
5
54liuyao 已提交
4769 4770
      qDebug("===stream===return data:single interval. recv datablock num:%" PRIu64, pInfo->numOfDatapack);
      pInfo->numOfDatapack = 0;
4771 4772
      break;
    }
5
54liuyao 已提交
4773
    pInfo->numOfDatapack++;
4774 4775
    printDataBlock(pBlock, "single interval recv");

4776 4777
    if (pBlock->info.type == STREAM_DELETE_DATA || pBlock->info.type == STREAM_DELETE_RESULT ||
        pBlock->info.type == STREAM_CLEAR) {
5
54liuyao 已提交
4778
      doDeleteWindows(pOperator, &pInfo->interval, pBlock, pInfo->pDelWins, pInfo->pUpdatedMap);
4779 4780
      continue;
    } else if (pBlock->info.type == STREAM_GET_ALL) {
5
54liuyao 已提交
4781
      getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pInfo->pUpdatedMap);
4782
      continue;
5
54liuyao 已提交
4783
    } else if (pBlock->info.type == STREAM_CREATE_CHILD_TABLE) {
5
54liuyao 已提交
4784
      printDataBlock(pBlock, "single interval");
5
54liuyao 已提交
4785 4786 4787
      return pBlock;
    } else {
      ASSERTS(pBlock->info.type == STREAM_NORMAL || pBlock->info.type == STREAM_INVALID, "invalid SSDataBlock type");
4788 4789 4790 4791 4792 4793 4794 4795 4796 4797 4798 4799 4800 4801 4802
    }

    if (pBlock->info.type == STREAM_NORMAL && pBlock->info.version != 0) {
      // set input version
      pTaskInfo->version = pBlock->info.version;
    }

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

    // 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
4803
    setInputDataBlock(pSup, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
4804 4805 4806 4807
    if (pInfo->invertible) {
      setInverFunction(pSup->pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.type);
    }

5
54liuyao 已提交
4808
    doStreamIntervalAggImpl(pOperator, pBlock, pBlock->info.id.groupId, pInfo->pUpdatedMap);
5
54liuyao 已提交
4809 4810
    pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
    pInfo->twAggSup.minTs = TMIN(pInfo->twAggSup.minTs, pBlock->info.window.skey);
4811 4812
  }
  pOperator->status = OP_RES_TO_RETURN;
5
54liuyao 已提交
4813
  removeDeleteResults(pInfo->pUpdatedMap, pInfo->pDelWins);
4814 4815
  closeStreamIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, NULL,
                            pInfo->pUpdatedMap, pInfo->pDelWins, pOperator);
4816

4817 4818 4819 4820 4821
  void*   pIte = NULL;
  int32_t iter = 0;
  while ((pIte = tSimpleHashIterate(pInfo->pUpdatedMap, pIte, &iter)) != NULL) {
    SWinKey* pKey = tSimpleHashGetKey(pIte, NULL);
    taosArrayPush(pInfo->pUpdated, pKey);
4822
  }
4823
  taosArraySort(pInfo->pUpdated, winKeyCmprImpl);
4824

5
54liuyao 已提交
4825 4826
  initMultiResInfoFromArrayList(&pInfo->groupResInfo, pInfo->pUpdated);
  pInfo->pUpdated = NULL;
4827
  blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
4828
  tSimpleHashCleanup(pInfo->pUpdatedMap);
5
54liuyao 已提交
4829
  pInfo->pUpdatedMap = NULL;
5
54liuyao 已提交
4830

5
54liuyao 已提交
4831 4832 4833 4834 4835 4836
#if 0
  char* pBuf = streamStateIntervalDump(pInfo->pState);
  qDebug("===stream===interval state%s", pBuf);
  taosMemoryFree(pBuf);
#endif

4837
  doBuildDeleteResult(pInfo, pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
4838
  if (pInfo->pDelRes->info.rows > 0) {
5
54liuyao 已提交
4839
    printDataBlock(pInfo->pDelRes, "single interval delete");
4840 4841 4842
    return pInfo->pDelRes;
  }

4843
  doBuildStreamIntervalResult(pOperator, pInfo->pState, pInfo->binfo.pRes, &pInfo->groupResInfo);
5
54liuyao 已提交
4844 4845 4846 4847 4848 4849
  if (pInfo->binfo.pRes->info.rows > 0) {
    printDataBlock(pInfo->binfo.pRes, "single interval");
    return pInfo->binfo.pRes;
  }

  return NULL;
4850 4851 4852 4853 4854
}

SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
                                                SExecTaskInfo* pTaskInfo) {
  SStreamIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamIntervalOperatorInfo));
4855
  SOperatorInfo*               pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
4856 4857 4858 4859 4860
  if (pInfo == NULL || pOperator == NULL) {
    goto _error;
  }
  SStreamIntervalPhysiNode* pIntervalPhyNode = (SStreamIntervalPhysiNode*)pPhyNode;

H
Haojun Liao 已提交
4861
  int32_t    code = TSDB_CODE_SUCCESS;
4862 4863
  int32_t    numOfCols = 0;
  SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
H
Haojun Liao 已提交
4864

H
Haojun Liao 已提交
4865
  SSDataBlock* pResBlock = createDataBlockFromDescNode(pPhyNode->pOutputDataBlockDesc);
4866 4867 4868 4869 4870 4871 4872 4873
  SInterval    interval = {
         .interval = pIntervalPhyNode->interval,
         .sliding = pIntervalPhyNode->sliding,
         .intervalUnit = pIntervalPhyNode->intervalUnit,
         .slidingUnit = pIntervalPhyNode->slidingUnit,
         .offset = pIntervalPhyNode->offset,
         .precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision,
  };
H
Haojun Liao 已提交
4874

4875 4876 4877 4878
  STimeWindowAggSupp twAggSupp = {
      .waterMark = pIntervalPhyNode->window.watermark,
      .calTrigger = pIntervalPhyNode->window.triggerType,
      .maxTs = INT64_MIN,
5
54liuyao 已提交
4879
      .minTs = INT64_MAX,
5
54liuyao 已提交
4880
      .deleteMark = getDeleteMark(pIntervalPhyNode),
4881
  };
H
Haojun Liao 已提交
4882

5
54liuyao 已提交
4883
  ASSERTS(twAggSupp.calTrigger != STREAM_TRIGGER_MAX_DELAY, "trigger type should not be max delay");
4884

4885 4886 4887 4888
  pOperator->pTaskInfo = pTaskInfo;
  pInfo->interval = interval;
  pInfo->twAggSup = twAggSupp;
  pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired;
4889
  pInfo->ignoreExpiredDataSaved = false;
4890 4891 4892
  pInfo->isFinal = false;

  SExprSupp* pSup = &pOperator->exprSupp;
H
Haojun Liao 已提交
4893 4894 4895 4896
  initBasicInfo(&pInfo->binfo, pResBlock);
  initStreamFunciton(pSup->pCtx, pSup->numOfExprs);
  initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);

4897
  pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;
4898
  initResultSizeInfo(&pOperator->resultInfo, 4096);
4899

4900
  size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
L
Liu Jicong 已提交
4901 4902
  code = initAggSup(pSup, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str,
                    pTaskInfo->streamInfo.pState);
4903 4904 4905 4906
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

H
Haojun Liao 已提交
4907 4908 4909 4910 4911 4912 4913 4914
  if (pIntervalPhyNode->window.pExprs != NULL) {
    int32_t    numOfScalar = 0;
    SExprInfo* pScalarExprInfo = createExprInfo(pIntervalPhyNode->window.pExprs, NULL, &numOfScalar);
    code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
    if (code != TSDB_CODE_SUCCESS) {
      goto _error;
    }
  }
4915 4916

  pInfo->invertible = allInvertible(pSup->pCtx, numOfCols);
4917
  pInfo->invertible = false;
4918 4919 4920 4921 4922
  pInfo->pDelWins = taosArrayInit(4, sizeof(SWinKey));
  pInfo->delIndex = 0;
  pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT);
  initResultRowInfo(&pInfo->binfo.resultRowInfo);

4923 4924 4925 4926 4927 4928 4929 4930 4931 4932 4933 4934 4935
  pInfo->pState = taosMemoryCalloc(1, sizeof(SStreamState));
  *(pInfo->pState) = *(pTaskInfo->streamInfo.pState);
  streamStateSetNumber(pInfo->pState, -1);

  pInfo->pPhyNode = NULL;  // create new child
  pInfo->pPullDataMap = NULL;
  pInfo->pPullWins = NULL;  // SPullWindowInfo
  pInfo->pullIndex = 0;
  pInfo->pPullDataRes = NULL;
  pInfo->isFinal = false;
  pInfo->pChildren = NULL;
  pInfo->delKey.ts = INT64_MAX;
  pInfo->delKey.groupId = 0;
5
54liuyao 已提交
4936
  pInfo->numOfDatapack = 0;
5
54liuyao 已提交
4937 4938
  pInfo->pUpdated = NULL;
  pInfo->pUpdatedMap = NULL;
4939

L
Liu Jicong 已提交
4940 4941
  setOperatorInfo(pOperator, "StreamIntervalOperator", QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL, true, OP_NOT_OPENED,
                  pInfo, pTaskInfo);
4942 4943
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doStreamIntervalAgg, NULL,
                                         destroyStreamFinalIntervalOperatorInfo, optrDefaultBufFn, NULL);
4944

5
54liuyao 已提交
4945
  initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, &pInfo->twAggSup);
4946 4947 4948 4949 4950 4951 4952 4953
  code = appendDownstream(pOperator, &downstream, 1);
  if (code != TSDB_CODE_SUCCESS) {
    goto _error;
  }

  return pOperator;

_error:
4954
  destroyStreamFinalIntervalOperatorInfo(pInfo);
4955 4956 4957 4958
  taosMemoryFreeClear(pOperator);
  pTaskInfo->code = code;
  return NULL;
}