#3660

Committed 15 Mar 2025 09:06AM UTC coverage: 62.039% (-1.3%) from 63.314%

Build # #3660

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(stream): support stream processing for virtual tables (#30144)

* enh: add client processing

* enh: add mnode vtables processing

* enh: add mnode vtable processing

* enh: add normal child vtable support

* fix: compile issues

* fix: compile issues

* fix: create stream issues

* fix: multi stream scan issue

* fix: remove debug info

* fix: agg task and task level issues

* fix: correct task output type

* fix: split vtablescan from agg

* fix: memory leak issues

* fix: add limitations

* Update 09-error-code.md

* Update 09-error-code.md

* fix: remove usless case

* feat(stream): extract original table data in source scan task

Implemented functionality in the source task to extract data
corresponding to the virtual table from the original table using WAL.
The extracted data is then sent to the downstream merge task for further
processing.

* feat(stream): multi-way merge using loser tree in virtual merge task

Implemented multi-way merge in the merge task using a loser tree to
combine data from multiple original table into a single virtual table.
The merged virtual table data is then pushed downstream for further
processing.  Introduced memory limit handling during the merge process
with configurable behavior when the memory limit is reached.

* fix(test): remove useless cases

---------

Co-authored-by: dapan1121 <wpan@taosdata.com>
Co-authored-by: Pan Wei <72057773+dapan1121@users.noreply.github.com>

Run Details

154078 of 317582 branches covered (48.52%)

Branch coverage included in aggregate %.

313 of 2391 new or added lines in 34 files covered. (13.09%)

26134 existing lines in 205 files now uncovered.

240261 of 318051 relevant lines covered (75.54%)

16655189.27 hits per line

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

78.59

/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);

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->exprSupp.hasWindowOrGroup = true;

  SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;

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

  code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0);
  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);
  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->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);

  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, MAIN_SCAN, 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);
      QUERY_CHECK_CODE(code, lino, _end);
    }

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

    code = doFilter(pRes, pSup->pFilterInfo, 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], 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->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);

  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
31	int32_t createEventwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,	279,965✔
32	SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
33	QRY_PARAM_CHECK(pOptrInfo);	279,965!
34
35	int32_t code = TSDB_CODE_SUCCESS;	279,965✔
36	int32_t lino = 0;	279,965✔
37	SEventWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SEventWindowOperatorInfo));	279,965!
38	SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));	279,963!
39	if (pInfo == NULL \|\| pOperator == NULL) {	279,964!
40	code = terrno;	×
41	goto _error;	×
42	}
43
44	pOperator->exprSupp.hasWindowOrGroup = true;	279,964✔
45
46	SEventWinodwPhysiNode* pEventWindowNode = (SEventWinodwPhysiNode*)physiNode;	279,964✔
47
48	int32_t tsSlotId = ((SColumnNode*)pEventWindowNode->window.pTspk)->slotId;	279,964✔
49	code = filterInitFromNode((SNode*)pEventWindowNode->pStartCond, &pInfo->pStartCondInfo, 0);	279,964✔
50	QUERY_CHECK_CODE(code, lino, _error);	279,965!
51
52	code = filterInitFromNode((SNode*)pEventWindowNode->pEndCond, &pInfo->pEndCondInfo, 0);	279,965✔
53	QUERY_CHECK_CODE(code, lino, _error);	279,965!
54
55	if (pEventWindowNode->window.pExprs != NULL) {	279,965✔
56	int32_t numOfScalarExpr = 0;	262✔
57	SExprInfo* pScalarExprInfo = NULL;	262✔
58
59	code = createExprInfo(pEventWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);	262✔
60	QUERY_CHECK_CODE(code, lino, _error);	262!
61	code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);	262✔
62	QUERY_CHECK_CODE(code, lino, _error);	262!
63	}
64
65	code = filterInitFromNode((SNode*)pEventWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);	279,965✔
66	QUERY_CHECK_CODE(code, lino, _error);	279,965!
67
68	size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;	279,965✔
69
70	int32_t num = 0;	279,965✔
71	SExprInfo* pExprInfo = NULL;	279,965✔
72	code = createExprInfo(pEventWindowNode->window.pFuncs, NULL, &pExprInfo, &num);	279,965✔
73	QUERY_CHECK_CODE(code, lino, _error);	279,965!
74
75	initResultSizeInfo(&pOperator->resultInfo, 4096);	279,965✔
76
77	code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,	279,965✔
78	pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);	279,965✔
79	QUERY_CHECK_CODE(code, lino, _error);	279,964!
80
81	SSDataBlock* pResBlock = createDataBlockFromDescNode(pEventWindowNode->window.node.pOutputDataBlockDesc);	279,964✔
82	QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);	279,965!
83	initBasicInfo(&pInfo->binfo, pResBlock);	279,965✔
84
85	code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);	279,965✔
86	QUERY_CHECK_CODE(code, lino, _error);	279,963!
87
88	initResultRowInfo(&pInfo->binfo.resultRowInfo);	279,963✔
89	pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;	279,963✔
90	pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;	279,963✔
91
92	pInfo->twAggSup = (STimeWindowAggSupp){.waterMark = pEventWindowNode->window.watermark,	279,963✔
93	.calTrigger = pEventWindowNode->window.triggerType};	279,963✔
94
95	code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);	279,963✔
96	QUERY_CHECK_CODE(code, lino, _error);	279,963!
97
98	pInfo->tsSlotId = tsSlotId;	279,963✔
99	pInfo->pPreDataBlock = NULL;	279,963✔
100	pInfo->pOperator = pOperator;	279,963✔
101	pInfo->trueForLimit = pEventWindowNode->trueForLimit;	279,963✔
102
103	setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_EVENT, true, OP_NOT_OPENED, pInfo,	279,963✔
104	pTaskInfo);
105	pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, eventWindowAggregateNext, NULL, destroyEWindowOperatorInfo,	279,964✔
106	optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
107
108	code = appendDownstream(pOperator, &downstream, 1);	279,964✔
109	if (code != TSDB_CODE_SUCCESS) {	279,964!
110	goto _error;	×
111	}
112
113	*pOptrInfo = pOperator;	279,964✔
114	return TSDB_CODE_SUCCESS;	279,964✔
115
116	_error:	×
117	if (pInfo != NULL) {	×
118	destroyEWindowOperatorInfo(pInfo);	×
119	}
120
121	destroyOperatorAndDownstreams(pOperator, &downstream, 1);	×
122	pTaskInfo->code = code;	×
123	return code;	×
124	}
125
126	void cleanupResultInfoInEventWindow(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo) {	279,965✔
127	if (pInfo == NULL \|\| pInfo->pRow == NULL \|\| pOperator == NULL) {	279,965!
128	return;	80,518✔
129	}
130	SExprSupp* pSup = &pOperator->exprSupp;	199,447✔
131	for (int32_t j = 0; j < pSup->numOfExprs; ++j) {	496,438✔
132	pSup->pCtx[j].resultInfo = getResultEntryInfo(pInfo->pRow, j, pSup->rowEntryInfoOffset);	296,991✔
133	if (pSup->pCtx[j].fpSet.cleanup) {	296,991!
134	pSup->pCtx[j].fpSet.cleanup(&pSup->pCtx[j]);	×
135	}
136	}
137	}
138
139	void destroyEWindowOperatorInfo(void* param) {	279,965✔
140	SEventWindowOperatorInfo* pInfo = (SEventWindowOperatorInfo*)param;	279,965✔
141	if (pInfo == NULL) {	279,965!
142	return;	×
143	}
144
145	if (pInfo->pRow != NULL) {	279,965✔
146	taosMemoryFree(pInfo->pRow);	199,447!
147	}
148
149	if (pInfo->pStartCondInfo != NULL) {	279,965!
150	filterFreeInfo(pInfo->pStartCondInfo);	279,965✔
151	pInfo->pStartCondInfo = NULL;	279,965✔
152	}
153
154	if (pInfo->pEndCondInfo != NULL) {	279,965!
155	filterFreeInfo(pInfo->pEndCondInfo);	279,965✔
156	pInfo->pEndCondInfo = NULL;	279,965✔
157	}
158
159	cleanupBasicInfo(&pInfo->binfo);	279,965✔
160	colDataDestroy(&pInfo->twAggSup.timeWindowData);	279,965✔
161
162	cleanupResultInfoInEventWindow(pInfo->pOperator, pInfo);	279,965✔
163	pInfo->pOperator = NULL;	279,965✔
164	cleanupAggSup(&pInfo->aggSup);	279,965✔
165	cleanupExprSupp(&pInfo->scalarSup);	279,965✔
166	taosMemoryFreeClear(param);	279,964!
167	}
168
169	static int32_t eventWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {	479,409✔
170	int32_t code = TSDB_CODE_SUCCESS;	479,409✔
171	int32_t lino = 0;	479,409✔
172	SEventWindowOperatorInfo* pInfo = pOperator->info;	479,409✔
173	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	479,409✔
174
175	SExprSupp* pSup = &pOperator->exprSupp;	479,409✔
176	int32_t order = pInfo->binfo.inputTsOrder;	479,409✔
177
178	SSDataBlock* pRes = pInfo->binfo.pRes;	479,409✔
179
180	blockDataCleanup(pRes);	479,409✔
181
182	SOperatorInfo* downstream = pOperator->pDownstream[0];	479,412✔
183	while (1) {	245,194✔
184	SSDataBlock* pBlock = NULL;	724,606✔
185	if (pInfo->pPreDataBlock == NULL) {	724,606✔
186	pBlock = getNextBlockFromDownstream(pOperator, 0);	724,576✔
187	} else {
188	pBlock = pInfo->pPreDataBlock;	30✔
189	pInfo->pPreDataBlock = NULL;	30✔
190	}
191
192	if (pBlock == NULL) {	724,606✔
193	break;	476,401✔
194	}
195
196	pRes->info.scanFlag = pBlock->info.scanFlag;	248,205✔
197	code = setInputDataBlock(pSup, pBlock, order, MAIN_SCAN, true);	248,205✔
198	QUERY_CHECK_CODE(code, lino, _end);	248,205!
199
200	code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);	248,205✔
201	QUERY_CHECK_CODE(code, lino, _end);	248,205!
202
203	// there is an scalar expression that needs to be calculated right before apply the group aggregation.
204	if (pInfo->scalarSup.pExprInfo != NULL) {	248,205✔
205	code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,	264✔
206	pInfo->scalarSup.numOfExprs, NULL);
207	QUERY_CHECK_CODE(code, lino, _end);	264!
208	}
209
210	code = eventWindowAggImpl(pOperator, pInfo, pBlock);	248,205✔
211	QUERY_CHECK_CODE(code, lino, _end);	248,206!
212
213	code = doFilter(pRes, pSup->pFilterInfo, NULL);	248,206✔
214	QUERY_CHECK_CODE(code, lino, _end);	248,206!
215
216	if (pRes->info.rows >= pOperator->resultInfo.threshold \|\|	248,206✔
217	(pRes->info.id.groupId != pInfo->groupId && pRes->info.rows > 0)) {	245,203✔
218	(*ppRes) = pRes;	3,012✔
219	return code;	3,012✔
220	}
221	}
222
223	_end:	476,401✔
224	if (code != TSDB_CODE_SUCCESS) {	476,401!
225	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
226	pTaskInfo->code = code;	×
227	T_LONG_JMP(pTaskInfo->env, code);	×
228	}
229	(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;	476,401✔
230	return code;	476,401✔
231	}
232
233	static int32_t setSingleOutputTupleBufv1(SResultRowInfo* pResultRowInfo, STimeWindow* win, SResultRow** pResult,	75,991,047✔
234	SExprSupp* pExprSup, SAggSupporter* pAggSup) {
235	if (*pResult == NULL) {	75,991,047✔
236	SResultRow* p = taosMemoryCalloc(1, pAggSup->resultRowSize);	199,445!
237	if (!p) {	199,447!
238	return terrno;	×
239	}
240	pResultRowInfo->cur = (SResultRowPosition){.pageId = p->pageId, .offset = p->offset};	199,447✔
241	*pResult = p;	199,447✔
242	}
243
244	(pResult)->win = win;	75,991,049✔
245
246	return setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	75,991,049✔
247	}
248
249	static int32_t doEventWindowAggImpl(SEventWindowOperatorInfo* pInfo, SExprSupp* pSup, int32_t startIndex,	75,992,386✔
250	int32_t endIndex, const SSDataBlock* pBlock, int64_t* tsList,
251	SExecTaskInfo* pTaskInfo) {
252	int32_t code = TSDB_CODE_SUCCESS;	75,992,386✔
253	SWindowRowsSup* pRowSup = &pInfo->winSup;	75,992,386✔
254
255	int32_t numOfOutput = pSup->numOfExprs;	75,992,386✔
256	int32_t numOfRows = endIndex - startIndex + 1;	75,992,386✔
257
258	doKeepTuple(pRowSup, tsList[endIndex], pBlock->info.id.groupId);	75,992,386✔
259
260	code = setSingleOutputTupleBufv1(&pInfo->binfo.resultRowInfo, &pRowSup->win, &pInfo->pRow, pSup, &pInfo->aggSup);	75,992,021✔
261	if (code != TSDB_CODE_SUCCESS) { // null data, too many state code	75,986,499!
262	qError("failed to set single output tuple buffer, code:%d", code);	×
263	return code;	×
264	}
265
266	updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pRowSup->win, 0);	75,986,499✔
267	code = applyAggFunctionOnPartialTuples(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startIndex, numOfRows,	75,986,568✔
268	pBlock->info.rows, numOfOutput);	75,986,568✔
269	return code;	75,989,676✔
270	}
271
272	int32_t eventWindowAggImpl(SOperatorInfo* pOperator, SEventWindowOperatorInfo* pInfo, SSDataBlock* pBlock) {	248,205✔
273	int32_t code = TSDB_CODE_SUCCESS;	248,205✔
274	int32_t lino = 0;	248,205✔
275	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	248,205✔
276	SExprSupp* pSup = &pOperator->exprSupp;	248,205✔
277	SSDataBlock* pRes = pInfo->binfo.pRes;	248,205✔
278	int64_t gid = pBlock->info.id.groupId;	248,205✔
279	SColumnInfoData ps = NULL, pe = NULL;	248,205✔
280	SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);	248,205✔
281	QUERY_CHECK_NULL(pColInfoData, code, lino, _return, terrno);	248,206!
282	TSKEY* tsList = (TSKEY*)pColInfoData->pData;	248,206✔
283	SWindowRowsSup* pRowSup = &pInfo->winSup;	248,206✔
284	int32_t rowIndex = 0;	248,206✔
285	int64_t minWindowSize = getMinWindowSize(pOperator);	248,206✔
286
287	pRowSup->numOfRows = 0;	248,206✔
288	if (pInfo->groupId == 0) {	248,206✔
289	pInfo->groupId = gid;	248,145✔
290	} else if (pInfo->groupId != gid) {	61✔
291	// this is a new group, reset the info
292	pInfo->inWindow = false;	31✔
293	pInfo->groupId = gid;	31✔
294	pInfo->pPreDataBlock = pBlock;	31✔
295	goto _return;	31✔
296	}
297	pRes->info.id.groupId = pInfo->groupId;	248,175✔
298
299	SFilterColumnParam param1 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	248,175✔
300
301	code = filterSetDataFromSlotId(pInfo->pStartCondInfo, &param1);	248,174✔
302	QUERY_CHECK_CODE(code, lino, _return);	248,175!
303
304	int32_t status1 = 0;	248,175✔
305	code = filterExecute(pInfo->pStartCondInfo, pBlock, &ps, NULL, param1.numOfCols, &status1);	248,175✔
306	QUERY_CHECK_CODE(code, lino, _return);	248,175!
307
308	SFilterColumnParam param2 = {.numOfCols = taosArrayGetSize(pBlock->pDataBlock), .pDataBlock = pBlock->pDataBlock};	248,175✔
309	code = filterSetDataFromSlotId(pInfo->pEndCondInfo, &param2);	248,175✔
310	QUERY_CHECK_CODE(code, lino, _return);	248,175!
311
312	int32_t status2 = 0;	248,175✔
313	code = filterExecute(pInfo->pEndCondInfo, pBlock, &pe, NULL, param2.numOfCols, &status2);	248,175✔
314	QUERY_CHECK_CODE(code, lino, _return);	248,175!
315
316	int32_t startIndex = pInfo->inWindow ? 0 : -1;	248,175✔
317	while (rowIndex < pBlock->info.rows) {	152,206,303✔
318	if (pInfo->inWindow) { // let's find the first end value	152,004,120✔
319	for (rowIndex = startIndex; rowIndex < pBlock->info.rows; ++rowIndex) {	96,449,836✔
320	if (((bool*)pe->pData)[rowIndex]) {	96,405,276✔
321	break;	75,945,632✔
322	}
323	}
324
325	if (rowIndex < pBlock->info.rows) {	75,990,192✔
326	code = doEventWindowAggImpl(pInfo, pSup, startIndex, rowIndex, pBlock, tsList, pTaskInfo);	75,946,820✔
327	QUERY_CHECK_CODE(code, lino, _return);	75,945,146!
328	doUpdateNumOfRows(pSup->pCtx, pInfo->pRow, pSup->numOfExprs, pSup->rowEntryInfoOffset);	75,945,146✔
329
330	if (pRowSup->win.ekey - pRowSup->win.skey < minWindowSize) {	75,941,608✔
331	qDebug("skip small window, groupId: %" PRId64 ", windowSize: %" PRId64 ", minWindowSize: %" PRId64,	12!
332	pInfo->groupId, pRowSup->win.ekey - pRowSup->win.skey, minWindowSize);
333	} else {
334	// check buffer size
335	if (pRes->info.rows + pInfo->pRow->numOfRows >= pRes->info.capacity) {	75,941,596!
UNCOV 336	int32_t newSize = pRes->info.rows + pInfo->pRow->numOfRows;	×
UNCOV 337	code = blockDataEnsureCapacity(pRes, newSize);	×
UNCOV 338	QUERY_CHECK_CODE(code, lino, _return);	×
339	}
340
341	code = copyResultrowToDataBlock(pSup->pExprInfo, pSup->numOfExprs, pInfo->pRow, pSup->pCtx, pRes,	75,941,596✔
342	pSup->rowEntryInfoOffset, pTaskInfo);	75,941,596✔
343	QUERY_CHECK_CODE(code, lino, _return);	75,931,428!
344
345	pRes->info.rows += pInfo->pRow->numOfRows;	75,931,428✔
346	}
347	pInfo->pRow->numOfRows = 0;	75,932,255✔
348
349	pInfo->inWindow = false;	75,932,255✔
350	rowIndex += 1;	75,932,255✔
351	} else {
352	code = doEventWindowAggImpl(pInfo, pSup, startIndex, pBlock->info.rows - 1, pBlock, tsList, pTaskInfo);	43,372✔
353	QUERY_CHECK_CODE(code, lino, _return);	44,158!
354	}
355	} else { // find the first start value that is fulfill for the start condition
356	for (; rowIndex < pBlock->info.rows; ++rowIndex) {	97,107,331✔
357	if (((bool*)ps->pData)[rowIndex]) {	97,062,279✔
358	doKeepNewWindowStartInfo(pRowSup, tsList, rowIndex, gid);	75,968,876✔
359	pInfo->inWindow = true;	75,971,571✔
360	startIndex = rowIndex;	75,971,571✔
361	if (pInfo->pRow != NULL) {	75,971,571✔
362	clearResultRowInitFlag(pSup->pCtx, pSup->numOfExprs);	75,774,825✔
363	}
364	break;	75,981,398✔
365	}
366	}
367
368	if (pInfo->inWindow) {	76,026,450✔
369	continue; // try to find the end position	75,981,715✔
370	} else {
371	break; // no valid start position, quit	44,735✔
372	}
373	}
374	}
375
376	_return:	202,183✔
377
378	if (code != TSDB_CODE_SUCCESS) {	246,949!
379	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
380	}
381	colDataDestroy(ps);	246,949✔
382	taosMemoryFree(ps);	248,206!
383	colDataDestroy(pe);	248,206✔
384	taosMemoryFree(pe);	248,206!
385
386	return code;	248,206✔
387	}

taosdata / TDengine / #3660

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