#3653

Committed 14 Mar 2025 08:10AM UTC coverage: 22.565% (-41.0%) from 63.596%

Build # #3653

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat(keep): support keep on super table level. (#30097)

* Feat: support use keep while create super table.

* Test(keep): add test for create super table with keep option.

* Feat(keep): Add tmsg for create keep.

* Feat(keep): support alter table option keep.

* Fix(keep): Add baisc test for alter table option.

* Fix(keep): memory leek.

* Feat(keep): add keep to metaEntry&metaCache and fix earliestTs with stn keep.

* Test(keep): add some cases for select with stb keep.

* Fix: fix ci core while alter stb.

* Feat(keep): delete expired data in super table level.

* Feat: remove get stb keep while query.

* Fix : build error.

* Revert "Fix : build error."

This reverts commit 0ed66e4e8.

* Revert "Feat(keep): delete expired data in super table level."

This reverts commit 36330f6b4.

* Fix : build errors.

* Feat : support restart taosd.

* Fix : alter table comment problems.

* Test : add tests for super table keep.

* Fix: change sdb stb reserve size.

* Test: add more tests.

* Feat: Disable normal tables and sub tables from setting the keep parameter

* Fix: add more checks to avoid unknown address.

* Docs: Add docs for stable keep.

* Fix: some review changes.

* Fix: review errors.

Run Details

49248 of 302527 branches covered (16.28%)

Branch coverage included in aggregate %.

53 of 99 new or added lines in 12 files covered. (53.54%)

155872 existing lines in 443 files now uncovered.

87359 of 302857 relevant lines covered (28.84%)

570004.22 hits per line

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

0.0

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

taosdata / TDengine / #3653

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