#3543

Committed 29 Nov 2024 02:58AM UTC coverage: 60.842% (+0.02%) from 60.819%

Build # #3543

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #28973 from taosdata/merge/mainto3.0

merge: from main to 3.0

Run Details

120460 of 253224 branches covered (47.57%)

Branch coverage included in aggregate %.

706 of 908 new or added lines in 18 files covered. (77.75%)

2401 existing lines in 137 files now uncovered.

201633 of 276172 relevant lines covered (73.01%)

19045673.23 hits per line

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

78.4

/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"

typedef struct SEventWindowOperatorInfo {
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SExprSupp          scalarSup;
  SWindowRowsSup     winSup;
  int32_t            tsSlotId;  // primary timestamp column slot id
  STimeWindowAggSupp twAggSup;
  uint64_t           groupId;  // current group id, used to identify the data block from different groups
  SFilterInfo*       pStartCondInfo;
  SFilterInfo*       pEndCondInfo;
  bool               inWindow;
  SResultRow*        pRow;
  SSDataBlock*       pPreDataBlock;
  SOperatorInfo*     pOperator;
} SEventWindowOperatorInfo;

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;

  setOperatorInfo(pOperator, "EventWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_STATE, 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;

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

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

taosdata / TDengine / #3543

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