#4852

Committed 14 Nov 2025 08:06AM UTC coverage: 63.812% (+0.06%) from 63.754%

Build # #4852

Build Type

push

travis-ci

Committed by

guanshengliang

Commit Message

Merge branch '3.0' into cover/3.0

Run Details

355 of 675 new or added lines in 18 files covered. (52.59%)

3142 existing lines in 128 files now uncovered.

149263 of 233910 relevant lines covered (63.81%)

118719500.98 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.99

/source/libs/executor/src/eventwindowoperator.c

/*
 * 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/>.
 */

#include "executorInt.h"
#include "filter.h"
#include "function.h"
#include "functionMgt.h"
#include "operator.h"
#include "querytask.h"
#include "tcommon.h"
#include "tcompare.h"
#include "tdatablock.h"
#include "ttime.h"

static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** pRes);
static void    destroyEWindowOperatorInfo(void* param);
static int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock);
void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo);

static int32_t resetEventWindowOperState(SOperatorInfo* pOper) {
  SEventWindowOperatorInfo* pEvent = pOper->info;
  SExecTaskInfo*           pTaskInfo = pOper->pTaskInfo;
  SEventWinodwPhysiNode* pPhynode = (SEventWinodwPhysiNode*)pOper->pPhyNode;
  pOper->status = OP_NOT_OPENED;

  resetBasicOperatorState(&pEvent->binfo);
  taosMemoryFreeClear(pEvent->pRow);

  pEvent->groupId = 0;
  pEvent->pPreDataBlock = NULL;
  pEvent->inWindow = false;
  pEvent->winSup.lastTs = INT64_MIN;

  colDataDestroy(&pEvent->twAggSup.timeWindowData);
  int32_t code = initExecTimeWindowInfo(&pEvent->twAggSup.timeWindowData, &pTaskInfo->window);
  cleanupResultInfoInEventWindow(pOper, pEvent);

  if (code == 0) {
    code = resetAggSup(&pOper->exprSupp, &pEvent->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,
                       sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, pTaskInfo->streamInfo.pState,
                       &pTaskInfo->storageAPI.functionStore);
  }
  if (code == 0) {
    code = resetExprSupp(&pEvent->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,
                         &pTaskInfo->storageAPI.functionStore);
  }
  return code;
}

int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,
                                             SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
  QRY_PARAM_CHECK(pOptrInfo);

  int32_t                   code = TSDB_CODE_SUCCESS;
  int32_t                   lino = 0;
  SEventWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SEventWindowOperatorInfo));
  SOperatorInfo*            pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
  if (pInfo == NULL || pOperator == NULL) {
    code = terrno;
    goto _error;
  }

  pOperator->pPhyNode = physiNode;
  pOperator->exprSupp.hasWindowOrGroup = true;

  SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;

  int32_t tsSlotId = ((SColumnNode*)pEventWindowNode->window.pTspk)->slotId;
  code = filterInitFromNode((SNode*)pEventWindowNode->pStartCond, &pInfo->pStartCondInfo, 0,
                            pTaskInfo->pStreamRuntimeInfo);
  QUERY_CHECK_CODE(code, lino, _error);

  code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0,
                            pTaskInfo->pStreamRuntimeInfo);
  QUERY_CHECK_CODE(code, lino, _error);

  if (pEventWindowNode->window.pExprs != NULL) {
    int32_t    numOfScalarExpr = 0;
    SExprInfo* pScalarExprInfo = NULL;

    code = createExprInfo(pEventWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);
    QUERY_CHECK_CODE(code, lino, _error);
    code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);
    QUERY_CHECK_CODE(code, lino, _error);
  }

  code = filterInitFromNode((SNode*)pEventWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,
                            pTaskInfo->pStreamRuntimeInfo);
  QUERY_CHECK_CODE(code, lino, _error);

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

  int32_t    num = 0;
  SExprInfo* pExprInfo = NULL;
  code = createExprInfo(pEventWindowNode->window.pFuncs, NULL, &pExprInfo, &num);
  QUERY_CHECK_CODE(code, lino, _error);

  initResultSizeInfo(&pOperator->resultInfo, 4096);

  code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,
                    pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);
  QUERY_CHECK_CODE(code, lino, _error);

  SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc);
  QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);
  initBasicInfo(&pInfo->binfo, pResBlock);

  code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);
  QUERY_CHECK_CODE(code, lino, _error);

  initResultRowInfo(&pInfo->binfo.resultRowInfo);
  pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;
  pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;
  pInfo->winSup.lastTs = INT64_MIN;

  pInfo->twAggSup = (STimeWindowAggSupp){.waterMark = pEventWindowNode->window.watermark,
                                         .calTrigger = pEventWindowNode->window.triggerType};

  code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
  QUERY_CHECK_CODE(code, lino, _error);

  pInfo->tsSlotId = tsSlotId;
  pInfo->pPreDataBlock = NULL;
  pInfo->pOperator = pOperator;
  pInfo->trueForLimit = pEventWindowNode->trueForLimit;

  setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT, true, OP_NOT_OPENED, pInfo,
                  pTaskInfo);
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, eventWindowAggregateNext, NULL, destroyEWindowOperatorInfo,
                                         optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);

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

  *pOptrInfo = pOperator;
  return TSDB_CODE_SUCCESS;

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

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);
  pTaskInfo->code = code;
  return code;
}

void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo) {
  if (pInfo == NULL || pInfo->pRow == NULL || pOperator == NULL) {
    return;
  }
  SExprSupp*       pSup = &pOperator->exprSupp;
  for (int32_t j = 0; j < pSup->numOfExprs; ++j) {
    pSup->pCtx[j].resultInfo = getResultEntryInfo(pInfo->pRow, j, pSup->rowEntryInfoOffset);
    if (pSup->pCtx[j].fpSet.cleanup) {
      pSup->pCtx[j].fpSet.cleanup(&pSup->pCtx[j]);
    }
  }
}

void destroyEWindowOperatorInfo(void* param) {
  SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;
  if (pInfo == NULL) {
    return;
  }

  if (pInfo->pRow != NULL) {
    taosMemoryFree(pInfo->pRow);
  }

  if (pInfo->pStartCondInfo != NULL) {
    filterFreeInfo(pInfo->pStartCondInfo);
    pInfo->pStartCondInfo = NULL;
  }

  if (pInfo->pEndCondInfo != NULL) {
    filterFreeInfo(pInfo->pEndCondInfo);
    pInfo->pEndCondInfo = NULL;
  }

  cleanupBasicInfo(&pInfo->binfo);
  colDataDestroy(&pInfo->twAggSup.timeWindowData);

  cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);
  pInfo->pOperator = NULL;
  cleanupAggSup(&pInfo->aggSup);
  cleanupExprSupp(&pInfo->scalarSup);
  taosMemoryFreeClear(param);
}

static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
  int32_t                   code = TSDB_CODE_SUCCESS;
  int32_t                   lino = 0;
  SEventWindowOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;

  SExprSupp* pSup = &pOperator->exprSupp;
  int32_t    order = pInfo->binfo.inputTsOrder;

  SSDataBlock* pRes = pInfo->binfo.pRes;

  blockDataCleanup(pRes);

  SOperatorInfo* downstream = pOperator->pDownstream[0];
  while (1) {
    SSDataBlock* pBlock = NULL;
    if (pInfo->pPreDataBlock == NULL) {
      pBlock = getNextBlockFromDownstream(pOperator, 0);
    } else {
      pBlock = pInfo->pPreDataBlock;
      pInfo->pPreDataBlock = NULL;
    }

    if (pBlock == NULL) {
      break;
    }

    pRes->info.scanFlag = pBlock->info.scanFlag;
    code = setInputDataBlock(pSup, pBlock, order, pBlock->info.scanFlag, true);
    QUERY_CHECK_CODE(code, lino, _end);

    code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);
    QUERY_CHECK_CODE(code, lino, _end);

    // there is an scalar expression that needs to be calculated right before apply the group aggregation.
    if (pInfo->scalarSup.pExprInfo != NULL) {
      code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,
                                   pInfo->scalarSup.numOfExprs, NULL, GET_STM_RTINFO(pOperator->pTaskInfo));
      QUERY_CHECK_CODE(code, lino, _end);
    }

    code = eventWindowAggImpl(pOperator, pInfo, pBlock);
    QUERY_CHECK_CODE(code, lino, _end);

    code = doFilter(pRes, pSup->pFilterInfo, NULL, NULL);
    QUERY_CHECK_CODE(code, lino, _end);

    if (pRes->info.rows >= pOperator->resultInfo.threshold ||
        (pRes->info.id.groupId != pInfo->groupId && pRes->info.rows > 0)) {
      (*ppRes) = pRes;
      return code;
    }
  }

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
    pTaskInfo->code = code;
    T_LONG_JMP(pTaskInfo->env, code);
  }
  (*ppRes) =  pRes->info.rows == 0 ? NULL : pRes;
  return code;
}

static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,
                                         SExprSupp* pExprSup, SAggSupporter* pAggSup) {
  if (*pResult == NULL) {
    SResultRow* p = taosMemoryCalloc(1, pAggSup->resultRowSize);
    if (!p) {
      return terrno;
    }
    pResultRowInfo->cur = (SResultRowPosition){.pageId = p->pageId, .offset = p->offset};
    *pResult = p;
  }

  (*pResult)->win = *win;

  return setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);
}

static int32_t doEventWindowAggImpl(SEventWindowOperatorInfo* pInfo, SExprSupp* pSup, int32_t startIndex,
                                    int32_t endIndex, const SSDataBlock* pBlock, int64_t* tsList,
                                    SExecTaskInfo* pTaskInfo) {
  int32_t code = TSDB_CODE_SUCCESS;
  SWindowRowsSup* pRowSup = &pInfo->winSup;

  int32_t numOfOutput = pSup->numOfExprs;
  int32_t numOfRows = endIndex - startIndex + 1;

  doKeepTuple(pRowSup, tsList[endIndex], endIndex, pBlock->info.id.groupId);

  code = setSingleOutputTupleBufv1(&pInfo->binfo.resultRowInfo, &pRowSup->win, &pInfo->pRow, pSup, &pInfo->aggSup);
  if (code != TSDB_CODE_SUCCESS) {  // null data, too many state code
    qError("failed to set single output tuple buffer, code:%d", code);
    return code;
  }

  updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, 0);
  code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startIndex, numOfRows,
                                         pBlock->info.rows, numOfOutput);
  return code;
}

int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {
  int32_t          code = TSDB_CODE_SUCCESS;
  int32_t          lino = 0;
  SExecTaskInfo*   pTaskInfo = pOperator->pTaskInfo;
  SExprSupp*       pSup = &pOperator->exprSupp;
  SSDataBlock*     pRes = pInfo->binfo.pRes;
  int64_t          gid = pBlock->info.id.groupId;
  SColumnInfoData *ps = NULL, *pe = NULL;
  SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
  QUERY_CHECK_NULL(pColInfoData, code, lino, _return, terrno);
  TSKEY*           tsList = (TSKEY*)pColInfoData->pData;
  SWindowRowsSup*  pRowSup = &pInfo->winSup;
  int32_t          rowIndex = 0;
  int64_t          minWindowSize = getMinWindowSize(pOperator);

  pRowSup->numOfRows = 0;
  if (pInfo->groupId == 0) {
    pInfo->groupId = gid;
  } else if (pInfo->groupId != gid) {
    // this is a new group, reset the info
    pInfo->inWindow = false;
    pInfo->groupId = gid;
    pInfo->winSup.lastTs = INT64_MIN;
    pInfo->pPreDataBlock = pBlock;
    goto _return;
  }
  pRes->info.id.groupId = pInfo->groupId;

  SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};

  code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &param1);
  QUERY_CHECK_CODE(code, lino, _return);

  int32_t status1 = 0;
  code = filterExecute(pInfo->pStartCondInfo, pBlock, &ps, NULL, param1.numOfCols, &status1);
  QUERY_CHECK_CODE(code, lino, _return);

  SFilterColumnParam param2 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};
  code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &param2);
  QUERY_CHECK_CODE(code, lino, _return);

  int32_t status2 = 0;
  code = filterExecute(pInfo->pEndCondInfo, pBlock, &pe, NULL, param2.numOfCols, &status2);
  QUERY_CHECK_CODE(code, lino, _return);

  for (int32_t i = 0; i < pBlock->info.rows; ++i) {
    if (pBlock->info.scanFlag != PRE_SCAN) {
      if (pInfo->winSup.lastTs == INT64_MIN) {
        pInfo->winSup.lastTs = tsList[i];
      } else {
        if (tsList[i] == pInfo->winSup.lastTs) {
          qError("duplicate timestamp found in event window operator, groupId: %" PRId64 ", timestamp: %" PRId64,
                 gid, tsList[i]);
          code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;
          QUERY_CHECK_CODE(code, lino, _return);
        }
      }
    }
  }
  int32_t startIndex = pInfo->inWindow ? 0 : -1;
  while (rowIndex < pBlock->info.rows) {
    if (pInfo->inWindow) {  // let's find the first end value
      for (rowIndex = startIndex; rowIndex < pBlock->info.rows; ++rowIndex) {
        if (((bool*)pe->pData)[rowIndex]) {
          break;
        }
      }

      if (rowIndex < pBlock->info.rows) {
        code = doEventWindowAggImpl(pInfo, pSup, startIndex, rowIndex, pBlock, tsList, pTaskInfo);
        QUERY_CHECK_CODE(code, lino, _return);
        doUpdateNumOfRows(pSup->pCtx, pInfo->pRow, pSup->numOfExprs, pSup->rowEntryInfoOffset);

        if (pRowSup->win.ekey - pRowSup->win.skey < minWindowSize) {
          qDebug("skip small window, groupId: %" PRId64 ", windowSize: %" PRId64 ", minWindowSize: %" PRId64,
                 pInfo->groupId, pRowSup->win.ekey - pRowSup->win.skey, minWindowSize);
        } else {
          // check buffer size
          if (pRes->info.rows + pInfo->pRow->numOfRows >= pRes->info.capacity) {
            int32_t newSize = pRes->info.rows + pInfo->pRow->numOfRows;
            code = blockDataEnsureCapacity(pRes, newSize);
            QUERY_CHECK_CODE(code, lino, _return);
          }

          code = copyResultrowToDataBlock(pSup->pExprInfo, pSup->numOfExprs, pInfo->pRow, pSup->pCtx, pRes,
                                          pSup->rowEntryInfoOffset, pTaskInfo);
          QUERY_CHECK_CODE(code, lino, _return);

          pRes->info.rows += pInfo->pRow->numOfRows;
        }
        pInfo->pRow->numOfRows = 0;

        pInfo->inWindow = false;
        rowIndex += 1;
      } else {
        code = doEventWindowAggImpl(pInfo, pSup, startIndex, pBlock->info.rows - 1, pBlock, tsList, pTaskInfo);
        QUERY_CHECK_CODE(code, lino, _return);
      }
    } else {  // find the first start value that is fulfill for the start condition
      for (; rowIndex < pBlock->info.rows; ++rowIndex) {
        if (((bool*)ps->pData)[rowIndex]) {
          doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);
          pInfo->inWindow = true;
          startIndex = rowIndex;
          if (pInfo->pRow != NULL) {
            clearResultRowInitFlag(pSup->pCtx, pSup->numOfExprs);
          }
          break;
        }
      }

      if (pInfo->inWindow) {
        continue;  // try to find the end position
      } else {
        break;  // no valid start position, quit
      }
    }
  }

_return:

  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  colDataDestroy(ps);
  taosMemoryFree(ps);
  colDataDestroy(pe);
  taosMemoryFree(pe);

  return code;
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#include "executorInt.h"
17	#include "filter.h"
18	#include "function.h"
19	#include "functionMgt.h"
20	#include "operator.h"
21	#include "querytask.h"
22	#include "tcommon.h"
23	#include "tcompare.h"
24	#include "tdatablock.h"
25	#include "ttime.h"
26
27	static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** pRes);
28	static void destroyEWindowOperatorInfo(void* param);
29	static int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock);
30	void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo);
31
32	static int32_t resetEventWindowOperState(SOperatorInfo* pOper) {	×
33	SEventWindowOperatorInfo* pEvent = pOper->info;	×
34	SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;	×
35	SEventWinodwPhysiNode* pPhynode = (SEventWinodwPhysiNode*)pOper->pPhyNode;	×
36	pOper->status = OP_NOT_OPENED;	×
37
38	resetBasicOperatorState(&pEvent->binfo);	×
39	taosMemoryFreeClear(pEvent->pRow);	×
40
41	pEvent->groupId = 0;	×
42	pEvent->pPreDataBlock = NULL;	×
43	pEvent->inWindow = false;	×
NEW 44	pEvent->winSup.lastTs = INT64_MIN;	×
45
UNCOV 46	colDataDestroy(&pEvent->twAggSup.timeWindowData);	×
47	int32_t code = initExecTimeWindowInfo(&pEvent->twAggSup.timeWindowData, &pTaskInfo->window);	×
48	cleanupResultInfoInEventWindow(pOper, pEvent);	×
49
UNCOV 50	if (code == 0) {	×
51	code = resetAggSup(&pOper->exprSupp, &pEvent->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,	×
52	sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, pTaskInfo->streamInfo.pState,	×
53	&pTaskInfo->storageAPI.functionStore);
54	}
UNCOV 55	if (code == 0) {	×
56	code = resetExprSupp(&pEvent->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,	×
57	&pTaskInfo->storageAPI.functionStore);
58	}
UNCOV 59	return code;	×
60	}
61
62	int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,	184,051✔
63	SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
64	QRY_PARAM_CHECK(pOptrInfo);	184,051✔
65
66	int32_t code = TSDB_CODE_SUCCESS;	184,051✔
67	int32_t lino = 0;	184,051✔
68	SEventWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SEventWindowOperatorInfo));	184,051✔
69	SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));	184,051✔
70	if (pInfo == NULL \|\| pOperator == NULL) {	184,051✔
UNCOV 71	code = terrno;	×
72	goto _error;	×
73	}
74
75	pOperator->pPhyNode = physiNode;	184,051✔
76	pOperator->exprSupp.hasWindowOrGroup = true;	184,051✔
77
78	SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;	184,051✔
79
80	int32_t tsSlotId = ((SColumnNode*)pEventWindowNode->window.pTspk)->slotId;	184,051✔
81	code = filterInitFromNode((SNode*)pEventWindowNode->pStartCond, &pInfo->pStartCondInfo, 0,	184,051✔
82	pTaskInfo->pStreamRuntimeInfo);	184,051✔
83	QUERY_CHECK_CODE(code, lino, _error);	184,051✔
84
85	code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0,	184,051✔
86	pTaskInfo->pStreamRuntimeInfo);	184,051✔
87	QUERY_CHECK_CODE(code, lino, _error);	184,051✔
88
89	if (pEventWindowNode->window.pExprs != NULL) {	184,051✔
90	int32_t numOfScalarExpr = 0;	611✔
91	SExprInfo* pScalarExprInfo = NULL;	611✔
92
93	code = createExprInfo(pEventWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);	611✔
94	QUERY_CHECK_CODE(code, lino, _error);	611✔
95	code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);	611✔
96	QUERY_CHECK_CODE(code, lino, _error);	611✔
97	}
98
99	code = filterInitFromNode((SNode*)pEventWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,	184,051✔
100	pTaskInfo->pStreamRuntimeInfo);	184,051✔
101	QUERY_CHECK_CODE(code, lino, _error);	184,051✔
102
103	size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;	184,051✔
104
105	int32_t num = 0;	184,051✔
106	SExprInfo* pExprInfo = NULL;	184,051✔
107	code = createExprInfo(pEventWindowNode->window.pFuncs, NULL, &pExprInfo, &num);	184,051✔
108	QUERY_CHECK_CODE(code, lino, _error);	184,051✔
109
110	initResultSizeInfo(&pOperator->resultInfo, 4096);	184,051✔
111
112	code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,	368,102✔
113	pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);	184,051✔
114	QUERY_CHECK_CODE(code, lino, _error);	184,051✔
115
116	SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc);	184,051✔
117	QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);	184,051✔
118	initBasicInfo(&pInfo->binfo, pResBlock);	184,051✔
119
120	code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);	184,051✔
121	QUERY_CHECK_CODE(code, lino, _error);	184,051✔
122
123	initResultRowInfo(&pInfo->binfo.resultRowInfo);	184,051✔
124	pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;	184,051✔
125	pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;	184,051✔
126	pInfo->winSup.lastTs = INT64_MIN;	184,051✔
127
128	pInfo->twAggSup = (STimeWindowAggSupp){.waterMark = pEventWindowNode->window.watermark,	184,051✔
129	.calTrigger = pEventWindowNode->window.triggerType};	184,051✔
130
131	code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);	184,051✔
132	QUERY_CHECK_CODE(code, lino, _error);	184,051✔
133
134	pInfo->tsSlotId = tsSlotId;	184,051✔
135	pInfo->pPreDataBlock = NULL;	184,051✔
136	pInfo->pOperator = pOperator;	184,051✔
137	pInfo->trueForLimit = pEventWindowNode->trueForLimit;	184,051✔
138
139	setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT, true, OP_NOT_OPENED, pInfo,	184,051✔
140	pTaskInfo);
141	pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, eventWindowAggregateNext, NULL, destroyEWindowOperatorInfo,	184,051✔
142	optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
143
144	setOperatorResetStateFn(pOperator, resetEventWindowOperState);	184,051✔
145	code = appendDownstream(pOperator, &downstream, 1);	184,051✔
146	if (code != TSDB_CODE_SUCCESS) {	184,051✔
UNCOV 147	goto _error;	×
148	}
149
150	*pOptrInfo = pOperator;	184,051✔
151	return TSDB_CODE_SUCCESS;	184,051✔
152
UNCOV 153	_error:	×
UNCOV 154	if (pInfo != NULL) {	×
155	destroyEWindowOperatorInfo(pInfo);	×
156	}
157
UNCOV 158	destroyOperatorAndDownstreams(pOperator, &downstream, 1);	×
UNCOV 159	pTaskInfo->code = code;	×
160	return code;	×
161	}
162
163	void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo) {	184,051✔
164	if (pInfo == NULL \|\| pInfo->pRow == NULL \|\| pOperator == NULL) {	184,051✔
165	return;	36,785✔
166	}
167	SExprSupp* pSup = &pOperator->exprSupp;	147,266✔
168	for (int32_t j = 0; j < pSup->numOfExprs; ++j) {	1,743,422✔
169	pSup->pCtx[j].resultInfo = getResultEntryInfo(pInfo->pRow, j, pSup->rowEntryInfoOffset);	1,596,156✔
170	if (pSup->pCtx[j].fpSet.cleanup) {	1,596,156✔
UNCOV 171	pSup->pCtx[j].fpSet.cleanup(&pSup->pCtx[j]);	×
172	}
173	}
174	}
175
176	void destroyEWindowOperatorInfo(void* param) {	184,051✔
177	SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;	184,051✔
178	if (pInfo == NULL) {	184,051✔
UNCOV 179	return;	×
180	}
181
182	if (pInfo->pRow != NULL) {	184,051✔
183	taosMemoryFree(pInfo->pRow);	147,266✔
184	}
185
186	if (pInfo->pStartCondInfo != NULL) {	184,051✔
187	filterFreeInfo(pInfo->pStartCondInfo);	184,051✔
188	pInfo->pStartCondInfo = NULL;	184,051✔
189	}
190
191	if (pInfo->pEndCondInfo != NULL) {	184,051✔
192	filterFreeInfo(pInfo->pEndCondInfo);	184,051✔
193	pInfo->pEndCondInfo = NULL;	184,051✔
194	}
195
196	cleanupBasicInfo(&pInfo->binfo);	184,051✔
197	colDataDestroy(&pInfo->twAggSup.timeWindowData);	184,051✔
198
199	cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);	184,051✔
200	pInfo->pOperator = NULL;	184,051✔
201	cleanupAggSup(&pInfo->aggSup);	184,051✔
202	cleanupExprSupp(&pInfo->scalarSup);	184,051✔
203	taosMemoryFreeClear(param);	184,051✔
204	}
205
206	static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {	529,516✔
207	int32_t code = TSDB_CODE_SUCCESS;	529,516✔
208	int32_t lino = 0;	529,516✔
209	SEventWindowOperatorInfo* pInfo = pOperator->info;	529,516✔
210	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	529,516✔
211
212	SExprSupp* pSup = &pOperator->exprSupp;	529,516✔
213	int32_t order = pInfo->binfo.inputTsOrder;	529,516✔
214
215	SSDataBlock* pRes = pInfo->binfo.pRes;	529,516✔
216
217	blockDataCleanup(pRes);	529,516✔
218
219	SOperatorInfo* downstream = pOperator->pDownstream[0];	529,516✔
220	while (1) {	917,009✔
221	SSDataBlock* pBlock = NULL;	1,446,525✔
222	if (pInfo->pPreDataBlock == NULL) {	1,446,525✔
223	pBlock = getNextBlockFromDownstream(pOperator, 0);	1,050,272✔
224	} else {
225	pBlock = pInfo->pPreDataBlock;	396,253✔
226	pInfo->pPreDataBlock = NULL;	396,253✔
227	}
228
229	if (pBlock == NULL) {	1,446,525✔
230	break;	129,768✔
231	}
232
233	pRes->info.scanFlag = pBlock->info.scanFlag;	1,316,757✔
234	code = setInputDataBlock(pSup, pBlock, order, pBlock->info.scanFlag, true);	1,316,757✔
235	QUERY_CHECK_CODE(code, lino, _end);	1,316,757✔
236
237	code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);	1,316,757✔
238	QUERY_CHECK_CODE(code, lino, _end);	1,316,757✔
239
240	// there is an scalar expression that needs to be calculated right before apply the group aggregation.
241	if (pInfo->scalarSup.pExprInfo != NULL) {	1,316,757✔
242	code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,	1,539✔
243	pInfo->scalarSup.numOfExprs, NULL, GET_STM_RTINFO(pOperator->pTaskInfo));	1,539✔
244	QUERY_CHECK_CODE(code, lino, _end);	1,539✔
245	}
246
247	code = eventWindowAggImpl(pOperator, pInfo, pBlock);	1,316,757✔
248	QUERY_CHECK_CODE(code, lino, _end);	1,316,757✔
249
250	code = doFilter(pRes, pSup->pFilterInfo, NULL, NULL);	1,308,100✔
251	QUERY_CHECK_CODE(code, lino, _end);	1,308,100✔
252
253	if (pRes->info.rows >= pOperator->resultInfo.threshold \|\|	1,308,100✔
254	(pRes->info.id.groupId != pInfo->groupId && pRes->info.rows > 0)) {	1,277,128✔
255	(*ppRes) = pRes;	391,091✔
256	return code;	391,091✔
257	}
258	}
259
260	_end:	138,425✔
261	if (code != TSDB_CODE_SUCCESS) {	138,425✔
262	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	8,657✔
263	pTaskInfo->code = code;	8,657✔
264	T_LONG_JMP(pTaskInfo->env, code);	8,657✔
265	}
266	(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;	129,768✔
267	return code;	129,768✔
268	}
269
270	static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,	218,261,237✔
271	SExprSupp* pExprSup, SAggSupporter* pAggSup) {
272	if (*pResult == NULL) {	218,261,237✔
273	SResultRow* p = taosMemoryCalloc(1, pAggSup->resultRowSize);	147,266✔
274	if (!p) {	147,266✔
UNCOV 275	return terrno;	×
276	}
277	pResultRowInfo->cur = (SResultRowPosition){.pageId = p->pageId, .offset = p->offset};	147,266✔
278	*pResult = p;	147,266✔
279	}
280
281	(pResult)->win = win;	218,261,237✔
282
283	return setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	218,261,237✔
284	}
285
286	static int32_t doEventWindowAggImpl(SEventWindowOperatorInfo* pInfo, SExprSupp* pSup, int32_t startIndex,	218,261,237✔
287	int32_t endIndex, const SSDataBlock* pBlock, int64_t* tsList,
288	SExecTaskInfo* pTaskInfo) {
289	int32_t code = TSDB_CODE_SUCCESS;	218,261,237✔
290	SWindowRowsSup* pRowSup = &pInfo->winSup;	218,261,237✔
291
292	int32_t numOfOutput = pSup->numOfExprs;	218,261,237✔
293	int32_t numOfRows = endIndex - startIndex + 1;	218,261,237✔
294
295	doKeepTuple(pRowSup, tsList[endIndex], endIndex, pBlock->info.id.groupId);	218,261,237✔
296
297	code = setSingleOutputTupleBufv1(&pInfo->binfo.resultRowInfo, &pRowSup->win, &pInfo->pRow, pSup, &pInfo->aggSup);	218,261,237✔
298	if (code != TSDB_CODE_SUCCESS) { // null data, too many state code	218,261,237✔
UNCOV 299	qError("failed to set single output tuple buffer, code:%d", code);	×
UNCOV 300	return code;	×
301	}
302
303	updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, 0);	218,261,237✔
304	code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startIndex, numOfRows,	218,261,237✔
305	pBlock->info.rows, numOfOutput);	218,261,237✔
306	return code;	218,261,237✔
307	}
308
309	int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {	1,316,757✔
310	int32_t code = TSDB_CODE_SUCCESS;	1,316,757✔
311	int32_t lino = 0;	1,316,757✔
312	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	1,316,757✔
313	SExprSupp* pSup = &pOperator->exprSupp;	1,316,757✔
314	SSDataBlock* pRes = pInfo->binfo.pRes;	1,316,757✔
315	int64_t gid = pBlock->info.id.groupId;	1,316,757✔
316	SColumnInfoData ps = NULL, pe = NULL;	1,316,757✔
317	SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);	1,316,757✔
318	QUERY_CHECK_NULL(pColInfoData, code, lino, _return, terrno);	1,316,757✔
319	TSKEY* tsList = (TSKEY*)pColInfoData->pData;	1,316,757✔
320	SWindowRowsSup* pRowSup = &pInfo->winSup;	1,316,757✔
321	int32_t rowIndex = 0;	1,316,757✔
322	int64_t minWindowSize = getMinWindowSize(pOperator);	1,316,757✔
323
324	pRowSup->numOfRows = 0;	1,316,757✔
325	if (pInfo->groupId == 0) {	1,316,757✔
326	pInfo->groupId = gid;	481,129✔
327	} else if (pInfo->groupId != gid) {	835,628✔
328	// this is a new group, reset the info
329	pInfo->inWindow = false;	428,701✔
330	pInfo->groupId = gid;	428,701✔
331	pInfo->winSup.lastTs = INT64_MIN;	428,701✔
332	pInfo->pPreDataBlock = pBlock;	428,701✔
333	goto _return;	428,701✔
334	}
335	pRes->info.id.groupId = pInfo->groupId;	888,056✔
336
337	SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	888,056✔
338
339	code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &param1);	888,056✔
340	QUERY_CHECK_CODE(code, lino, _return);	888,056✔
341
342	int32_t status1 = 0;	888,056✔
343	code = filterExecute(pInfo->pStartCondInfo, pBlock, &ps, NULL, param1.numOfCols, &status1);	888,056✔
344	QUERY_CHECK_CODE(code, lino, _return);	888,056✔
345
346	SFilterColumnParam param2 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	888,056✔
347	code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &param2);	888,056✔
348	QUERY_CHECK_CODE(code, lino, _return);	888,056✔
349
350	int32_t status2 = 0;	888,056✔
351	code = filterExecute(pInfo->pEndCondInfo, pBlock, &pe, NULL, param2.numOfCols, &status2);	888,056✔
352	QUERY_CHECK_CODE(code, lino, _return);	888,056✔
353
354	for (int32_t i = 0; i < pBlock->info.rows; ++i) {	1,636,464,176✔
355	if (pBlock->info.scanFlag != PRE_SCAN) {	1,635,584,777✔
356	if (pInfo->winSup.lastTs == INT64_MIN) {	1,635,584,777✔
357	pInfo->winSup.lastTs = tsList[i];	552,176✔
358	} else {
359	if (tsList[i] == pInfo->winSup.lastTs) {	1,635,032,601✔
360	qError("duplicate timestamp found in event window operator, groupId: %" PRId64 ", timestamp: %" PRId64,	8,657✔
361	gid, tsList[i]);
362	code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;	8,657✔
363	QUERY_CHECK_CODE(code, lino, _return);	8,657✔
364	}
365	}
366	}
367	}
368	int32_t startIndex = pInfo->inWindow ? 0 : -1;	879,399✔
369	while (rowIndex < pBlock->info.rows) {	437,180,325✔
370	if (pInfo->inWindow) { // let's find the first end value	436,515,819✔
371	for (rowIndex = startIndex; rowIndex < pBlock->info.rows; ++rowIndex) {	1,242,177,099✔
372	if (((bool*)pe->pData)[rowIndex]) {	1,241,623,107✔
373	break;	217,707,245✔
374	}
375	}
376
377	if (rowIndex < pBlock->info.rows) {	218,261,237✔
378	code = doEventWindowAggImpl(pInfo, pSup, startIndex, rowIndex, pBlock, tsList, pTaskInfo);	217,707,245✔
379	QUERY_CHECK_CODE(code, lino, _return);	217,707,245✔
380	doUpdateNumOfRows(pSup->pCtx, pInfo->pRow, pSup->numOfExprs, pSup->rowEntryInfoOffset);	217,707,245✔
381
382	if (pRowSup->win.ekey - pRowSup->win.skey < minWindowSize) {	217,707,245✔
UNCOV 383	qDebug("skip small window, groupId: %" PRId64 ", windowSize: %" PRId64 ", minWindowSize: %" PRId64,	×
384	pInfo->groupId, pRowSup->win.ekey - pRowSup->win.skey, minWindowSize);
385	} else {
386	// check buffer size
387	if (pRes->info.rows + pInfo->pRow->numOfRows >= pRes->info.capacity) {	217,707,245✔
388	int32_t newSize = pRes->info.rows + pInfo->pRow->numOfRows;	2,229,984✔
389	code = blockDataEnsureCapacity(pRes, newSize);	2,229,984✔
390	QUERY_CHECK_CODE(code, lino, _return);	2,229,984✔
391	}
392
393	code = copyResultrowToDataBlock(pSup->pExprInfo, pSup->numOfExprs, pInfo->pRow, pSup->pCtx, pRes,	217,707,245✔
394	pSup->rowEntryInfoOffset, pTaskInfo);	217,707,245✔
395	QUERY_CHECK_CODE(code, lino, _return);	217,707,245✔
396
397	pRes->info.rows += pInfo->pRow->numOfRows;	217,707,245✔
398	}
399	pInfo->pRow->numOfRows = 0;	217,707,245✔
400
401	pInfo->inWindow = false;	217,707,245✔
402	rowIndex += 1;	217,707,245✔
403	} else {
404	code = doEventWindowAggImpl(pInfo, pSup, startIndex, pBlock->info.rows - 1, pBlock, tsList, pTaskInfo);	553,992✔
405	QUERY_CHECK_CODE(code, lino, _return);	553,992✔
406	}
407	} else { // find the first start value that is fulfill for the start condition
408	for (; rowIndex < pBlock->info.rows; ++rowIndex) {	612,198,938✔
409	if (((bool*)ps->pData)[rowIndex]) {	611,984,045✔
410	doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);	218,039,689✔
411	pInfo->inWindow = true;	218,039,689✔
412	startIndex = rowIndex;	218,039,689✔
413	if (pInfo->pRow != NULL) {	218,039,689✔
414	clearResultRowInitFlag(pSup->pCtx, pSup->numOfExprs);	217,892,423✔
415	}
416	break;	218,039,689✔
417	}
418	}
419
420	if (pInfo->inWindow) {	218,254,582✔
421	continue; // try to find the end position	218,039,689✔
422	} else {
423	break; // no valid start position, quit	214,893✔
424	}
425	}
426	}
427
428	_return:	1,316,757✔
429
430	if (code != TSDB_CODE_SUCCESS) {	1,316,757✔
431	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	8,657✔
432	}
433	colDataDestroy(ps);	1,316,757✔
434	taosMemoryFree(ps);	1,316,757✔
435	colDataDestroy(pe);	1,316,757✔
436	taosMemoryFree(pe);	1,316,757✔
437
438	return code;	1,316,757✔
439	}

taosdata / TDengine / #4852

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous