#3621

Committed 22 Feb 2025 11:44AM UTC coverage: 2.037% (-61.5%) from 63.573%

Build # #3621

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

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

merge: from main to 3.0 branch

Run Details

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

0 of 174 new or added lines in 18 files covered. (0.0%)

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 hits per line

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

0.0

/source/libs/executor/src/countwindowoperator.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 SCountWindowResult {
  int32_t    winRows;
  SResultRow row;
} SCountWindowResult;

typedef struct SCountWindowSupp {
  SArray* pWinStates;
  int32_t stateIndex;
  int32_t curStateIndex;
} SCountWindowSupp;

typedef struct SCountWindowOperatorInfo {
  SOptrBasicInfo     binfo;
  SAggSupporter      aggSup;
  SExprSupp          scalarSup;
  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
  SResultRow*        pRow;
  int32_t            windowCount;
  int32_t            windowSliding;
  SCountWindowSupp   countSup;
  SSDataBlock*       pPreDataBlock;
  int32_t            preStateIndex;
} SCountWindowOperatorInfo;

void destroyCountWindowOperatorInfo(void* param) {
  SCountWindowOperatorInfo* pInfo = (SCountWindowOperatorInfo*)param;
  if (pInfo == NULL) {
    return;
  }
  cleanupBasicInfo(&pInfo->binfo);
  colDataDestroy(&pInfo->twAggSup.timeWindowData);

  cleanupAggSup(&pInfo->aggSup);
  cleanupExprSupp(&pInfo->scalarSup);
  taosArrayDestroy(pInfo->countSup.pWinStates);
  taosMemoryFreeClear(param);
}

static int32_t countWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes);

static void clearWinStateBuff(SCountWindowResult* pBuff) { pBuff->winRows = 0; }

static SCountWindowResult* getCountWinStateInfo(SCountWindowSupp* pCountSup) {
  SCountWindowResult* pBuffInfo = taosArrayGet(pCountSup->pWinStates, pCountSup->stateIndex);
  pCountSup->curStateIndex = pCountSup->stateIndex;
  if (!pBuffInfo) {
    terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
    return NULL;
  }
  int32_t size = taosArrayGetSize(pCountSup->pWinStates);
  if (size == 0) {
    terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));
    return NULL;
  }
  pCountSup->stateIndex = (pCountSup->stateIndex + 1) % size;
  return pBuffInfo;
}

static int32_t setCountWindowOutputBuff(SExprSupp* pExprSup, SCountWindowSupp* pCountSup, SResultRow** pResult,
                                        SCountWindowResult** ppResBuff) {
  int32_t             code = TSDB_CODE_SUCCESS;
  int32_t             lino = 0;
  SCountWindowResult* pBuff = getCountWinStateInfo(pCountSup);
  QUERY_CHECK_NULL(pBuff, code, lino, _end, terrno);
  (*pResult) = &pBuff->row;
  code = setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);
  (*ppResBuff) = pBuff;

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
  }
  return code;
}

static int32_t updateCountWindowInfo(int32_t start, int32_t blockRows, int32_t countWinRows, int32_t* pCurrentRows) {
  int32_t rows = TMIN(countWinRows - (*pCurrentRows), blockRows - start);
  (*pCurrentRows) += rows;
  return rows;
}

void doCountWindowAggImpl(SOperatorInfo* pOperator, SSDataBlock* pBlock) {
  int32_t                   code = TSDB_CODE_SUCCESS;
  int32_t                   lino = 0;
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
  SExprSupp*                pExprSup = &pOperator->exprSupp;
  SCountWindowOperatorInfo* pInfo = pOperator->info;
  SSDataBlock*              pRes = pInfo->binfo.pRes;
  SColumnInfoData*          pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);
  QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno);
  TSKEY* tsCols = (TSKEY*)pColInfoData->pData;
  int32_t numOfBuff = taosArrayGetSize(pInfo->countSup.pWinStates);
  if (numOfBuff == 0) {
    code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;
    qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));
    T_LONG_JMP(pTaskInfo->env, code);
  }
  pInfo->countSup.stateIndex = (pInfo->preStateIndex + 1) % numOfBuff;

  int32_t newSize = pRes->info.rows + pBlock->info.rows / pInfo->windowSliding + 1;
  if (newSize > pRes->info.capacity) {
    code = blockDataEnsureCapacity(pRes, newSize);
    QUERY_CHECK_CODE(code, lino, _end);
  }

  for (int32_t i = 0; i < pBlock->info.rows;) {
    SCountWindowResult* pBuffInfo = NULL;
    code = setCountWindowOutputBuff(pExprSup, &pInfo->countSup, &pInfo->pRow, &pBuffInfo);
    if (code != TSDB_CODE_SUCCESS) {
      qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
      T_LONG_JMP(pTaskInfo->env, code);
    }
    int32_t prevRows = pBuffInfo->winRows;
    int32_t num = updateCountWindowInfo(i, pBlock->info.rows, pInfo->windowCount, &pBuffInfo->winRows);
    int32_t step = num;
    if (prevRows == 0) {
      pInfo->pRow->win.skey = tsCols[i];
    }
    pInfo->pRow->win.ekey = tsCols[num + i - 1];

    updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->pRow->win, 0);
    code = applyAggFunctionOnPartialTuples(pTaskInfo, pExprSup->pCtx, &pInfo->twAggSup.timeWindowData, i, num,
                                           pBlock->info.rows, pExprSup->numOfExprs);
    QUERY_CHECK_CODE(code, lino, _end);
    if (pInfo->windowCount != pInfo->windowSliding) {
      if (prevRows <= pInfo->windowSliding) {
        if (pBuffInfo->winRows > pInfo->windowSliding) {
          step = pInfo->windowSliding - prevRows;
        } else {
          step = pInfo->windowSliding;
        }
      } else {
        step = 0;
      }
    }
    if (pBuffInfo->winRows == pInfo->windowCount) {
      doUpdateNumOfRows(pExprSup->pCtx, pInfo->pRow, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);
      code = copyResultrowToDataBlock(pExprSup->pExprInfo, pExprSup->numOfExprs, pInfo->pRow, pExprSup->pCtx, pRes,
                                      pExprSup->rowEntryInfoOffset, pTaskInfo);
      QUERY_CHECK_CODE(code, lino, _end);
      pRes->info.rows += pInfo->pRow->numOfRows;
      clearWinStateBuff(pBuffInfo);
      pInfo->preStateIndex = pInfo->countSup.curStateIndex;
      clearResultRowInitFlag(pExprSup->pCtx, pExprSup->numOfExprs);
    }
    i += step;
  }

  code = doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);
  QUERY_CHECK_CODE(code, lino, _end);

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

static void buildCountResult(SExprSupp* pExprSup, SCountWindowSupp* pCountSup, SExecTaskInfo* pTaskInfo,
                             SFilterInfo* pFilterInfo, int32_t preStateIndex, SSDataBlock* pBlock) {
  SResultRow* pResultRow = NULL;
  int32_t     code = TSDB_CODE_SUCCESS;
  int32_t     lino = 0;
  int32_t     numOfBuff = taosArrayGetSize(pCountSup->pWinStates);
  int32_t     newSize = pBlock->info.rows + numOfBuff;
  if (newSize > pBlock->info.capacity) {
    code = blockDataEnsureCapacity(pBlock, newSize);
    QUERY_CHECK_CODE(code, lino, _end);
  }
  pCountSup->stateIndex = (preStateIndex + 1) % numOfBuff;
  for (int32_t i = 0; i < numOfBuff; i++) {
    SCountWindowResult* pBuff = NULL;
    code = setCountWindowOutputBuff(pExprSup, pCountSup, &pResultRow, &pBuff);
    QUERY_CHECK_CODE(code, lino, _end);
    if (pBuff->winRows == 0) {
      continue;
    }
    doUpdateNumOfRows(pExprSup->pCtx, pResultRow, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);
    code = copyResultrowToDataBlock(pExprSup->pExprInfo, pExprSup->numOfExprs, pResultRow, pExprSup->pCtx, pBlock,
                                    pExprSup->rowEntryInfoOffset, pTaskInfo);
    QUERY_CHECK_CODE(code, lino, _end);
    pBlock->info.rows += pResultRow->numOfRows;
    clearWinStateBuff(pBuff);
    clearResultRowInitFlag(pExprSup->pCtx, pExprSup->numOfExprs);
  }
  code = doFilter(pBlock, pFilterInfo, NULL);
  QUERY_CHECK_CODE(code, lino, _end);

_end:
  if (code != TSDB_CODE_SUCCESS) {
    qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
    T_LONG_JMP(pTaskInfo->env, code);
  }
}

static int32_t countWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {
  int32_t                   code = TSDB_CODE_SUCCESS;
  int32_t                   lino = 0;
  SCountWindowOperatorInfo* pInfo = pOperator->info;
  SExecTaskInfo*            pTaskInfo = pOperator->pTaskInfo;
  SExprSupp*                pExprSup = &pOperator->exprSupp;
  int32_t                   order = pInfo->binfo.inputTsOrder;
  SSDataBlock*              pRes = pInfo->binfo.pRes;

  blockDataCleanup(pRes);

  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(pExprSup, 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);
    }

    if (pInfo->groupId == 0) {
      pInfo->groupId = pBlock->info.id.groupId;
    } else if (pInfo->groupId != pBlock->info.id.groupId) {
      pInfo->pPreDataBlock = pBlock;
      pRes->info.id.groupId = pInfo->groupId;
      buildCountResult(pExprSup, &pInfo->countSup, pTaskInfo, pOperator->exprSupp.pFilterInfo, pInfo->preStateIndex, pRes);
      pInfo->groupId = pBlock->info.id.groupId;
      if (pRes->info.rows > 0) {
        (*ppRes) = pRes;
        return code;
      }
    }

    doCountWindowAggImpl(pOperator, pBlock);
    if (pRes->info.rows >= pOperator->resultInfo.threshold) {
      pRes->info.id.groupId = pInfo->groupId;
      (*ppRes) = pRes;
      return code;
    }
  }

  pRes->info.id.groupId = pInfo->groupId;
  buildCountResult(pExprSup, &pInfo->countSup, pTaskInfo, pOperator->exprSupp.pFilterInfo, pInfo->preStateIndex, pRes);

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

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

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

  pOperator->exprSupp.hasWindowOrGroup = true;

  SCountWinodwPhysiNode* pCountWindowNode = (SCountWinodwPhysiNode*)physiNode;

  pInfo->tsSlotId = ((SColumnNode*)pCountWindowNode->window.pTspk)->slotId;

  if (pCountWindowNode->window.pExprs != NULL) {
    int32_t    numOfScalarExpr = 0;
    SExprInfo* pScalarExprInfo = NULL;
    code = createExprInfo(pCountWindowNode->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);
  }

  size_t     keyBufSize = 0;
  int32_t    num = 0;
  SExprInfo* pExprInfo = NULL;
  code = createExprInfo(pCountWindowNode->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(pCountWindowNode->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->windowCount = pCountWindowNode->windowCount;
  pInfo->windowSliding = pCountWindowNode->windowSliding;
  // sizeof(SCountWindowResult)
  int32_t itemSize = sizeof(int32_t) + pInfo->aggSup.resultRowSize;
  int32_t numOfItem = 1;
  if (pInfo->windowCount != pInfo->windowSliding) {
    numOfItem = pInfo->windowCount / pInfo->windowSliding + 1;
  }

  pInfo->countSup.pWinStates = taosArrayInit_s(itemSize, numOfItem);
  if (!pInfo->countSup.pWinStates) {
    goto _error;
  }

  pInfo->countSup.stateIndex = 0;
  pInfo->pPreDataBlock = NULL;
  pInfo->preStateIndex = 0;

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

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

  setOperatorInfo(pOperator, "CountWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_COUNT, true, OP_NOT_OPENED, pInfo,
                  pTaskInfo);
  pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, countWindowAggregateNext, NULL, destroyCountWindowOperatorInfo,
                                         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) {
    destroyCountWindowOperatorInfo(pInfo);
  }

  destroyOperatorAndDownstreams(pOperator, &downstream, 1);
  pTaskInfo->code = code;
  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 SCountWindowResult {
28	int32_t winRows;
29	SResultRow row;
30	} SCountWindowResult;
31
32	typedef struct SCountWindowSupp {
33	SArray* pWinStates;
34	int32_t stateIndex;
35	int32_t curStateIndex;
36	} SCountWindowSupp;
37
38	typedef struct SCountWindowOperatorInfo {
39	SOptrBasicInfo binfo;
40	SAggSupporter aggSup;
41	SExprSupp scalarSup;
42	int32_t tsSlotId; // primary timestamp column slot id
43	STimeWindowAggSupp twAggSup;
44	uint64_t groupId; // current group id, used to identify the data block from different groups
45	SResultRow* pRow;
46	int32_t windowCount;
47	int32_t windowSliding;
48	SCountWindowSupp countSup;
49	SSDataBlock* pPreDataBlock;
50	int32_t preStateIndex;
51	} SCountWindowOperatorInfo;
52
UNCOV 53	void destroyCountWindowOperatorInfo(void* param) {	×
UNCOV 54	SCountWindowOperatorInfo* pInfo = (SCountWindowOperatorInfo*)param;	×
UNCOV 55	if (pInfo == NULL) {	×
56	return;	×
57	}
UNCOV 58	cleanupBasicInfo(&pInfo->binfo);	×
UNCOV 59	colDataDestroy(&pInfo->twAggSup.timeWindowData);	×
60
UNCOV 61	cleanupAggSup(&pInfo->aggSup);	×
UNCOV 62	cleanupExprSupp(&pInfo->scalarSup);	×
UNCOV 63	taosArrayDestroy(pInfo->countSup.pWinStates);	×
UNCOV 64	taosMemoryFreeClear(param);	×
65	}
66
67	static int32_t countWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes);
68
UNCOV 69	static void clearWinStateBuff(SCountWindowResult* pBuff) { pBuff->winRows = 0; }	×
70
UNCOV 71	static SCountWindowResult* getCountWinStateInfo(SCountWindowSupp* pCountSup) {	×
UNCOV 72	SCountWindowResult* pBuffInfo = taosArrayGet(pCountSup->pWinStates, pCountSup->stateIndex);	×
UNCOV 73	pCountSup->curStateIndex = pCountSup->stateIndex;	×
UNCOV 74	if (!pBuffInfo) {	×
75	terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
76	qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));	×
77	return NULL;	×
78	}
UNCOV 79	int32_t size = taosArrayGetSize(pCountSup->pWinStates);	×
UNCOV 80	if (size == 0) {	×
81	terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
82	qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));	×
83	return NULL;	×
84	}
UNCOV 85	pCountSup->stateIndex = (pCountSup->stateIndex + 1) % size;	×
UNCOV 86	return pBuffInfo;	×
87	}
88
UNCOV 89	static int32_t setCountWindowOutputBuff(SExprSupp* pExprSup, SCountWindowSupp* pCountSup, SResultRow** pResult,	×
90	SCountWindowResult** ppResBuff) {
UNCOV 91	int32_t code = TSDB_CODE_SUCCESS;	×
UNCOV 92	int32_t lino = 0;	×
UNCOV 93	SCountWindowResult* pBuff = getCountWinStateInfo(pCountSup);	×
UNCOV 94	QUERY_CHECK_NULL(pBuff, code, lino, _end, terrno);	×
UNCOV 95	(*pResult) = &pBuff->row;	×
UNCOV 96	code = setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	×
UNCOV 97	(*ppResBuff) = pBuff;	×
98
UNCOV 99	_end:	×
UNCOV 100	if (code != TSDB_CODE_SUCCESS) {	×
101	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
102	}
UNCOV 103	return code;	×
104	}
105
UNCOV 106	static int32_t updateCountWindowInfo(int32_t start, int32_t blockRows, int32_t countWinRows, int32_t* pCurrentRows) {	×
UNCOV 107	int32_t rows = TMIN(countWinRows - (*pCurrentRows), blockRows - start);	×
UNCOV 108	(*pCurrentRows) += rows;	×
UNCOV 109	return rows;	×
110	}
111
UNCOV 112	void doCountWindowAggImpl(SOperatorInfo* pOperator, SSDataBlock* pBlock) {	×
UNCOV 113	int32_t code = TSDB_CODE_SUCCESS;	×
UNCOV 114	int32_t lino = 0;	×
UNCOV 115	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	×
UNCOV 116	SExprSupp* pExprSup = &pOperator->exprSupp;	×
UNCOV 117	SCountWindowOperatorInfo* pInfo = pOperator->info;	×
UNCOV 118	SSDataBlock* pRes = pInfo->binfo.pRes;	×
UNCOV 119	SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);	×
UNCOV 120	QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno);	×
UNCOV 121	TSKEY* tsCols = (TSKEY*)pColInfoData->pData;	×
UNCOV 122	int32_t numOfBuff = taosArrayGetSize(pInfo->countSup.pWinStates);	×
UNCOV 123	if (numOfBuff == 0) {	×
124	code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
125	qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
126	T_LONG_JMP(pTaskInfo->env, code);	×
127	}
UNCOV 128	pInfo->countSup.stateIndex = (pInfo->preStateIndex + 1) % numOfBuff;	×
129
UNCOV 130	int32_t newSize = pRes->info.rows + pBlock->info.rows / pInfo->windowSliding + 1;	×
UNCOV 131	if (newSize > pRes->info.capacity) {	×
132	code = blockDataEnsureCapacity(pRes, newSize);	×
133	QUERY_CHECK_CODE(code, lino, _end);	×
134	}
135
UNCOV 136	for (int32_t i = 0; i < pBlock->info.rows;) {	×
UNCOV 137	SCountWindowResult* pBuffInfo = NULL;	×
UNCOV 138	code = setCountWindowOutputBuff(pExprSup, &pInfo->countSup, &pInfo->pRow, &pBuffInfo);	×
UNCOV 139	if (code != TSDB_CODE_SUCCESS) {	×
140	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
141	T_LONG_JMP(pTaskInfo->env, code);	×
142	}
UNCOV 143	int32_t prevRows = pBuffInfo->winRows;	×
UNCOV 144	int32_t num = updateCountWindowInfo(i, pBlock->info.rows, pInfo->windowCount, &pBuffInfo->winRows);	×
UNCOV 145	int32_t step = num;	×
UNCOV 146	if (prevRows == 0) {	×
UNCOV 147	pInfo->pRow->win.skey = tsCols[i];	×
148	}
UNCOV 149	pInfo->pRow->win.ekey = tsCols[num + i - 1];	×
150
UNCOV 151	updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->pRow->win, 0);	×
UNCOV 152	code = applyAggFunctionOnPartialTuples(pTaskInfo, pExprSup->pCtx, &pInfo->twAggSup.timeWindowData, i, num,	×
UNCOV 153	pBlock->info.rows, pExprSup->numOfExprs);	×
UNCOV 154	QUERY_CHECK_CODE(code, lino, _end);	×
UNCOV 155	if (pInfo->windowCount != pInfo->windowSliding) {	×
UNCOV 156	if (prevRows <= pInfo->windowSliding) {	×
UNCOV 157	if (pBuffInfo->winRows > pInfo->windowSliding) {	×
UNCOV 158	step = pInfo->windowSliding - prevRows;	×
159	} else {
UNCOV 160	step = pInfo->windowSliding;	×
161	}
162	} else {
163	step = 0;	×
164	}
165	}
UNCOV 166	if (pBuffInfo->winRows == pInfo->windowCount) {	×
UNCOV 167	doUpdateNumOfRows(pExprSup->pCtx, pInfo->pRow, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	×
UNCOV 168	code = copyResultrowToDataBlock(pExprSup->pExprInfo, pExprSup->numOfExprs, pInfo->pRow, pExprSup->pCtx, pRes,	×
UNCOV 169	pExprSup->rowEntryInfoOffset, pTaskInfo);	×
UNCOV 170	QUERY_CHECK_CODE(code, lino, _end);	×
UNCOV 171	pRes->info.rows += pInfo->pRow->numOfRows;	×
UNCOV 172	clearWinStateBuff(pBuffInfo);	×
UNCOV 173	pInfo->preStateIndex = pInfo->countSup.curStateIndex;	×
UNCOV 174	clearResultRowInitFlag(pExprSup->pCtx, pExprSup->numOfExprs);	×
175	}
UNCOV 176	i += step;	×
177	}
178
UNCOV 179	code = doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL);	×
UNCOV 180	QUERY_CHECK_CODE(code, lino, _end);	×
181
UNCOV 182	_end:	×
UNCOV 183	if (code != TSDB_CODE_SUCCESS) {	×
184	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
185	pTaskInfo->code = code;	×
186	T_LONG_JMP(pTaskInfo->env, code);	×
187	}
UNCOV 188	}	×
189
UNCOV 190	static void buildCountResult(SExprSupp* pExprSup, SCountWindowSupp* pCountSup, SExecTaskInfo* pTaskInfo,	×
191	SFilterInfo* pFilterInfo, int32_t preStateIndex, SSDataBlock* pBlock) {
UNCOV 192	SResultRow* pResultRow = NULL;	×
UNCOV 193	int32_t code = TSDB_CODE_SUCCESS;	×
UNCOV 194	int32_t lino = 0;	×
UNCOV 195	int32_t numOfBuff = taosArrayGetSize(pCountSup->pWinStates);	×
UNCOV 196	int32_t newSize = pBlock->info.rows + numOfBuff;	×
UNCOV 197	if (newSize > pBlock->info.capacity) {	×
198	code = blockDataEnsureCapacity(pBlock, newSize);	×
199	QUERY_CHECK_CODE(code, lino, _end);	×
200	}
UNCOV 201	pCountSup->stateIndex = (preStateIndex + 1) % numOfBuff;	×
UNCOV 202	for (int32_t i = 0; i < numOfBuff; i++) {	×
UNCOV 203	SCountWindowResult* pBuff = NULL;	×
UNCOV 204	code = setCountWindowOutputBuff(pExprSup, pCountSup, &pResultRow, &pBuff);	×
UNCOV 205	QUERY_CHECK_CODE(code, lino, _end);	×
UNCOV 206	if (pBuff->winRows == 0) {	×
UNCOV 207	continue;	×
208	}
UNCOV 209	doUpdateNumOfRows(pExprSup->pCtx, pResultRow, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	×
UNCOV 210	code = copyResultrowToDataBlock(pExprSup->pExprInfo, pExprSup->numOfExprs, pResultRow, pExprSup->pCtx, pBlock,	×
UNCOV 211	pExprSup->rowEntryInfoOffset, pTaskInfo);	×
UNCOV 212	QUERY_CHECK_CODE(code, lino, _end);	×
UNCOV 213	pBlock->info.rows += pResultRow->numOfRows;	×
UNCOV 214	clearWinStateBuff(pBuff);	×
UNCOV 215	clearResultRowInitFlag(pExprSup->pCtx, pExprSup->numOfExprs);	×
216	}
UNCOV 217	code = doFilter(pBlock, pFilterInfo, NULL);	×
UNCOV 218	QUERY_CHECK_CODE(code, lino, _end);	×
219
UNCOV 220	_end:	×
UNCOV 221	if (code != TSDB_CODE_SUCCESS) {	×
222	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
223	T_LONG_JMP(pTaskInfo->env, code);	×
224	}
UNCOV 225	}	×
226
UNCOV 227	static int32_t countWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {	×
UNCOV 228	int32_t code = TSDB_CODE_SUCCESS;	×
UNCOV 229	int32_t lino = 0;	×
UNCOV 230	SCountWindowOperatorInfo* pInfo = pOperator->info;	×
UNCOV 231	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	×
UNCOV 232	SExprSupp* pExprSup = &pOperator->exprSupp;	×
UNCOV 233	int32_t order = pInfo->binfo.inputTsOrder;	×
UNCOV 234	SSDataBlock* pRes = pInfo->binfo.pRes;	×
235
UNCOV 236	blockDataCleanup(pRes);	×
237
UNCOV 238	while (1) {	×
UNCOV 239	SSDataBlock* pBlock = NULL;	×
UNCOV 240	if (pInfo->pPreDataBlock == NULL) {	×
UNCOV 241	pBlock = getNextBlockFromDownstream(pOperator, 0);	×
242	} else {
UNCOV 243	pBlock = pInfo->pPreDataBlock;	×
UNCOV 244	pInfo->pPreDataBlock = NULL;	×
245	}
246
UNCOV 247	if (pBlock == NULL) {	×
UNCOV 248	break;	×
249	}
250
UNCOV 251	pRes->info.scanFlag = pBlock->info.scanFlag;	×
UNCOV 252	code = setInputDataBlock(pExprSup, pBlock, order, MAIN_SCAN, true);	×
UNCOV 253	QUERY_CHECK_CODE(code, lino, _end);	×
254
UNCOV 255	code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);	×
UNCOV 256	QUERY_CHECK_CODE(code, lino, _end);	×
257
258	// there is an scalar expression that needs to be calculated right before apply the group aggregation.
UNCOV 259	if (pInfo->scalarSup.pExprInfo != NULL) {	×
260	code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,	×
261	pInfo->scalarSup.numOfExprs, NULL);
262	QUERY_CHECK_CODE(code, lino, _end);	×
263	}
264
UNCOV 265	if (pInfo->groupId == 0) {	×
UNCOV 266	pInfo->groupId = pBlock->info.id.groupId;	×
UNCOV 267	} else if (pInfo->groupId != pBlock->info.id.groupId) {	×
UNCOV 268	pInfo->pPreDataBlock = pBlock;	×
UNCOV 269	pRes->info.id.groupId = pInfo->groupId;	×
UNCOV 270	buildCountResult(pExprSup, &pInfo->countSup, pTaskInfo, pOperator->exprSupp.pFilterInfo, pInfo->preStateIndex, pRes);	×
UNCOV 271	pInfo->groupId = pBlock->info.id.groupId;	×
UNCOV 272	if (pRes->info.rows > 0) {	×
UNCOV 273	(*ppRes) = pRes;	×
UNCOV 274	return code;	×
275	}
276	}
277
UNCOV 278	doCountWindowAggImpl(pOperator, pBlock);	×
UNCOV 279	if (pRes->info.rows >= pOperator->resultInfo.threshold) {	×
280	pRes->info.id.groupId = pInfo->groupId;	×
281	(*ppRes) = pRes;	×
282	return code;	×
283	}
284	}
285
UNCOV 286	pRes->info.id.groupId = pInfo->groupId;	×
UNCOV 287	buildCountResult(pExprSup, &pInfo->countSup, pTaskInfo, pOperator->exprSupp.pFilterInfo, pInfo->preStateIndex, pRes);	×
288
UNCOV 289	_end:	×
UNCOV 290	if (code != TSDB_CODE_SUCCESS) {	×
291	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
292	pTaskInfo->code = code;	×
293	T_LONG_JMP(pTaskInfo->env, code);	×
294	}
UNCOV 295	(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;	×
UNCOV 296	return code;	×
297	}
298
UNCOV 299	int32_t createCountwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,	×
300	SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
UNCOV 301	QRY_PARAM_CHECK(pOptrInfo);	×
302
UNCOV 303	int32_t code = TSDB_CODE_SUCCESS;	×
UNCOV 304	int32_t lino = 0;	×
UNCOV 305	SCountWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SCountWindowOperatorInfo));	×
UNCOV 306	SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));	×
UNCOV 307	if (pInfo == NULL \|\| pOperator == NULL) {	×
308	code = terrno;	×
309	goto _error;	×
310	}
311
UNCOV 312	pOperator->exprSupp.hasWindowOrGroup = true;	×
313
UNCOV 314	SCountWinodwPhysiNode* pCountWindowNode = (SCountWinodwPhysiNode*)physiNode;	×
315
UNCOV 316	pInfo->tsSlotId = ((SColumnNode*)pCountWindowNode->window.pTspk)->slotId;	×
317
UNCOV 318	if (pCountWindowNode->window.pExprs != NULL) {	×
319	int32_t numOfScalarExpr = 0;	×
320	SExprInfo* pScalarExprInfo = NULL;	×
321	code = createExprInfo(pCountWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);	×
322	QUERY_CHECK_CODE(code, lino, _error);	×
323	code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);	×
324	QUERY_CHECK_CODE(code, lino, _error);	×
325	}
326
UNCOV 327	size_t keyBufSize = 0;	×
UNCOV 328	int32_t num = 0;	×
UNCOV 329	SExprInfo* pExprInfo = NULL;	×
UNCOV 330	code = createExprInfo(pCountWindowNode->window.pFuncs, NULL, &pExprInfo, &num);	×
UNCOV 331	QUERY_CHECK_CODE(code, lino, _error);	×
332
UNCOV 333	initResultSizeInfo(&pOperator->resultInfo, 4096);	×
334
UNCOV 335	code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,	×
UNCOV 336	pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);	×
UNCOV 337	QUERY_CHECK_CODE(code, lino, _error);	×
338
UNCOV 339	SSDataBlock* pResBlock = createDataBlockFromDescNode(pCountWindowNode->window.node.pOutputDataBlockDesc);	×
UNCOV 340	QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);	×
UNCOV 341	initBasicInfo(&pInfo->binfo, pResBlock);	×
342
UNCOV 343	code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);	×
UNCOV 344	QUERY_CHECK_CODE(code, lino, _error);	×
345
UNCOV 346	initResultRowInfo(&pInfo->binfo.resultRowInfo);	×
UNCOV 347	pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;	×
UNCOV 348	pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;	×
UNCOV 349	pInfo->windowCount = pCountWindowNode->windowCount;	×
UNCOV 350	pInfo->windowSliding = pCountWindowNode->windowSliding;	×
351	// sizeof(SCountWindowResult)
UNCOV 352	int32_t itemSize = sizeof(int32_t) + pInfo->aggSup.resultRowSize;	×
UNCOV 353	int32_t numOfItem = 1;	×
UNCOV 354	if (pInfo->windowCount != pInfo->windowSliding) {	×
UNCOV 355	numOfItem = pInfo->windowCount / pInfo->windowSliding + 1;	×
356	}
357
UNCOV 358	pInfo->countSup.pWinStates = taosArrayInit_s(itemSize, numOfItem);	×
UNCOV 359	if (!pInfo->countSup.pWinStates) {	×
360	goto _error;	×
361	}
362
UNCOV 363	pInfo->countSup.stateIndex = 0;	×
UNCOV 364	pInfo->pPreDataBlock = NULL;	×
UNCOV 365	pInfo->preStateIndex = 0;	×
366
UNCOV 367	code = filterInitFromNode((SNode*)pCountWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0);	×
UNCOV 368	QUERY_CHECK_CODE(code, lino, _error);	×
369
UNCOV 370	code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);	×
UNCOV 371	QUERY_CHECK_CODE(code, lino, _error);	×
372
UNCOV 373	setOperatorInfo(pOperator, "CountWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_COUNT, true, OP_NOT_OPENED, pInfo,	×
374	pTaskInfo);
UNCOV 375	pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, countWindowAggregateNext, NULL, destroyCountWindowOperatorInfo,	×
376	optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
377
UNCOV 378	code = appendDownstream(pOperator, &downstream, 1);	×
UNCOV 379	if (code != TSDB_CODE_SUCCESS) {	×
380	goto _error;	×
381	}
382
UNCOV 383	*pOptrInfo = pOperator;	×
UNCOV 384	return TSDB_CODE_SUCCESS;	×
385
386	_error:	×
387	if (pInfo != NULL) {	×
388	destroyCountWindowOperatorInfo(pInfo);	×
389	}
390
391	destroyOperatorAndDownstreams(pOperator, &downstream, 1);	×
392	pTaskInfo->code = code;	×
393	return code;	×
394	}

taosdata / TDengine / #3621

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