#4940

Committed 27 Jan 2026 10:23AM UTC coverage: 66.832% (-0.1%) from 66.931%

Build # #4940

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

fix: asan invalid write issue (#34400)

Run Details

7 of 8 new or added lines in 2 files covered. (87.5%)

822 existing lines in 141 files now uncovered.

204293 of 305680 relevant lines covered (66.83%)

124534723.83 hits per line

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

88.26

/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;
  TSKEY   lastTs; // this ts is used to record the last timestamp, so that we can know whether the new row's ts is duplicated
} 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; i++) {
    if (pBlock->info.scanFlag != PRE_SCAN) {
      if (pInfo->countSup.lastTs == INT64_MIN) {
        pInfo->countSup.lastTs = tsCols[i];
      } else {
        if (tsCols[i] == pInfo->countSup.lastTs) {
          qError("duplicate timestamp found in count window operator" PRId64 ", timestamp: %" PRId64, tsCols[i]);
          code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;
          QUERY_CHECK_CODE(code, lino, _end);
        } else {
          pInfo->countSup.lastTs = tsCols[i];
        }
      }
    }
  }

  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, 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, 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, GET_STM_RTINFO(pOperator->pTaskInfo));
      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;
      pInfo->countSup.lastTs = INT64_MIN;
      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;
}

static int32_t resetCountWindowOperatorState(SOperatorInfo* pOper) {
  SCountWindowOperatorInfo* pCount = pOper->info;
  SExecTaskInfo*           pTaskInfo = pOper->pTaskInfo;
  SCountWindowPhysiNode* pPhynode = (SCountWindowPhysiNode*)pOper->pPhyNode;
  pOper->status = OP_NOT_OPENED;
  
  resetBasicOperatorState(&pCount->binfo);
  pCount->groupId = 0;
  pCount->countSup.stateIndex = 0;
  pCount->countSup.lastTs = INT64_MIN;
  pCount->pPreDataBlock = NULL;
  pCount->preStateIndex = 0;
  pCount->pRow = NULL;

  colDataDestroy(&pCount->twAggSup.timeWindowData);
  int32_t code = initExecTimeWindowInfo(&pCount->twAggSup.timeWindowData, &pTaskInfo->window);
  
  if (code == 0) {
    code = resetAggSup(&pOper->exprSupp, &pCount->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,
                        sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, pTaskInfo->streamInfo.pState,
                        &pTaskInfo->storageAPI.functionStore);
  }
  if (code == 0) {
    code = resetExprSupp(&pCount->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,
                          &pTaskInfo->storageAPI.functionStore);
  }
  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->pPhyNode = physiNode;
  pOperator->exprSupp.hasWindowOrGroup = true;
  pOperator->exprSupp.hasWindow = true;

  SCountWindowPhysiNode* pCountWindowNode = (SCountWindowPhysiNode*)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;
  pInfo->countSup.lastTs = INT64_MIN;

  code = filterInitFromNode((SNode*)pCountWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,
                            pTaskInfo->pStreamRuntimeInfo);
  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);
  setOperatorResetStateFn(pOperator, resetCountWindowOperatorState);                                    
  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	TSKEY lastTs; // this ts is used to record the last timestamp, so that we can know whether the new row's ts is duplicated
37	} SCountWindowSupp;
38
39	typedef struct SCountWindowOperatorInfo {
40	SOptrBasicInfo binfo;
41	SAggSupporter aggSup;
42	SExprSupp scalarSup;
43	int32_t tsSlotId; // primary timestamp column slot id
44	STimeWindowAggSupp twAggSup;
45	uint64_t groupId; // current group id, used to identify the data block from different groups
46	SResultRow* pRow;
47	int32_t windowCount;
48	int32_t windowSliding;
49	SCountWindowSupp countSup;
50	SSDataBlock* pPreDataBlock;
51	int32_t preStateIndex;
52	} SCountWindowOperatorInfo;
53
54	void destroyCountWindowOperatorInfo(void* param) {	774,533✔
55	SCountWindowOperatorInfo* pInfo = (SCountWindowOperatorInfo*)param;	774,533✔
56	if (pInfo == NULL) {	774,533✔
57	return;	×
58	}
59	cleanupBasicInfo(&pInfo->binfo);	774,533✔
60	colDataDestroy(&pInfo->twAggSup.timeWindowData);	774,999✔
61
62	cleanupAggSup(&pInfo->aggSup);	774,999✔
63	cleanupExprSupp(&pInfo->scalarSup);	774,999✔
64	taosArrayDestroy(pInfo->countSup.pWinStates);	774,999✔
65	taosMemoryFreeClear(param);	774,999✔
66	}
67
68	static int32_t countWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes);
69
70	static void clearWinStateBuff(SCountWindowResult* pBuff) { pBuff->winRows = 0; }	93,241,358✔
71
72	static SCountWindowResult* getCountWinStateInfo(SCountWindowSupp* pCountSup) {	95,399,491✔
73	SCountWindowResult* pBuffInfo = taosArrayGet(pCountSup->pWinStates, pCountSup->stateIndex);	95,399,491✔
74	pCountSup->curStateIndex = pCountSup->stateIndex;	95,399,863✔
75	if (!pBuffInfo) {	95,399,491✔
76	terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
77	qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));	×
78	return NULL;	×
79	}
80	int32_t size = taosArrayGetSize(pCountSup->pWinStates);	95,399,491✔
81	if (size == 0) {	95,399,491✔
82	terrno = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
83	qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(terrno));	×
84	return NULL;	×
85	}
86	pCountSup->stateIndex = (pCountSup->stateIndex + 1) % size;	95,399,491✔
87	return pBuffInfo;	95,399,025✔
88	}
89
90	static int32_t setCountWindowOutputBuff(SExprSupp* pExprSup, SCountWindowSupp* pCountSup, SResultRow** pResult,	95,399,491✔
91	SCountWindowResult** ppResBuff) {
92	int32_t code = TSDB_CODE_SUCCESS;	95,399,491✔
93	int32_t lino = 0;	95,399,491✔
94	SCountWindowResult* pBuff = getCountWinStateInfo(pCountSup);	95,399,491✔
95	QUERY_CHECK_NULL(pBuff, code, lino, _end, terrno);	95,399,397✔
96	(*pResult) = &pBuff->row;	95,399,397✔
97	code = setResultRowInitCtx(*pResult, pExprSup->pCtx, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	95,399,491✔
98	(*ppResBuff) = pBuff;	95,399,476✔
99
100	_end:	95,399,476✔
101	if (code != TSDB_CODE_SUCCESS) {	95,399,476✔
102	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
103	}
104	return code;	95,399,491✔
105	}
106
107	static int32_t updateCountWindowInfo(int32_t start, int32_t blockRows, int32_t countWinRows, int32_t* pCurrentRows) {	94,491,906✔
108	int32_t rows = TMIN(countWinRows - (*pCurrentRows), blockRows - start);	94,491,906✔
109	(*pCurrentRows) += rows;	94,491,906✔
110	return rows;	94,492,293✔
111	}
112
113	void doCountWindowAggImpl(SOperatorInfo* pOperator, SSDataBlock* pBlock) {	1,936,316✔
114	int32_t code = TSDB_CODE_SUCCESS;	1,936,316✔
115	int32_t lino = 0;	1,936,316✔
116	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	1,936,316✔
117	SExprSupp* pExprSup = &pOperator->exprSupp;	1,936,316✔
118	SCountWindowOperatorInfo* pInfo = pOperator->info;	1,936,316✔
119	SSDataBlock* pRes = pInfo->binfo.pRes;	1,936,316✔
120	SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, pInfo->tsSlotId);	1,936,316✔
121	QUERY_CHECK_NULL(pColInfoData, code, lino, _end, terrno);	1,936,316✔
122	TSKEY* tsCols = (TSKEY*)pColInfoData->pData;	1,936,316✔
123	int32_t numOfBuff = taosArrayGetSize(pInfo->countSup.pWinStates);	1,936,316✔
124	if (numOfBuff == 0) {	1,936,316✔
125	code = TSDB_CODE_QRY_EXECUTOR_INTERNAL_ERROR;	×
126	qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(code));	×
127	T_LONG_JMP(pTaskInfo->env, code);	×
128	}
129	pInfo->countSup.stateIndex = (pInfo->preStateIndex + 1) % numOfBuff;	1,936,316✔
130
131	int32_t newSize = pRes->info.rows + pBlock->info.rows / pInfo->windowSliding + 1;	1,936,316✔
132	if (newSize > pRes->info.capacity) {	1,936,316✔
133	code = blockDataEnsureCapacity(pRes, newSize);	×
134	QUERY_CHECK_CODE(code, lino, _end);	×
135	}
136
137	for (int32_t i = 0; i < pBlock->info.rows; i++) {	1,477,452,980✔
138	if (pBlock->info.scanFlag != PRE_SCAN) {	1,475,577,214✔
139	if (pInfo->countSup.lastTs == INT64_MIN) {	1,475,571,796✔
140	pInfo->countSup.lastTs = tsCols[i];	480,361✔
141	} else {
142	if (tsCols[i] == pInfo->countSup.lastTs) {	1,475,095,692✔
143	qError("duplicate timestamp found in count window operator" PRId64 ", timestamp: %" PRId64, tsCols[i]);	52,810✔
144	code = TSDB_CODE_QRY_WINDOW_DUP_TIMESTAMP;	49,714✔
145	QUERY_CHECK_CODE(code, lino, _end);	49,714✔
146	} else {
147	pInfo->countSup.lastTs = tsCols[i];	1,475,042,882✔
148	}
149	}
150	}
151	}
152
153	for (int32_t i = 0; i < pBlock->info.rows;) {	96,375,799✔
154	SCountWindowResult* pBuffInfo = NULL;	94,492,293✔
155	code = setCountWindowOutputBuff(pExprSup, &pInfo->countSup, &pInfo->pRow, &pBuffInfo);	94,492,293✔
156	if (code != TSDB_CODE_SUCCESS) {	94,492,293✔
157	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
158	T_LONG_JMP(pTaskInfo->env, code);	×
159	}
160	int32_t prevRows = pBuffInfo->winRows;	94,492,293✔
161	int32_t num = updateCountWindowInfo(i, pBlock->info.rows, pInfo->windowCount, &pBuffInfo->winRows);	94,491,906✔
162	int32_t step = num;	94,492,293✔
163	if (prevRows == 0) {	94,492,293✔
164	pInfo->pRow->win.skey = tsCols[i];	93,241,358✔
165	}
166	pInfo->pRow->win.ekey = tsCols[num + i - 1];	94,492,293✔
167
168	updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pInfo->pRow->win, 0);	94,491,827✔
169	code = applyAggFunctionOnPartialTuples(pTaskInfo, pExprSup->pCtx, &pInfo->twAggSup.timeWindowData, i, num,	188,739,154✔
170	pBlock->info.rows, pExprSup->numOfExprs);	94,491,906✔
171	QUERY_CHECK_CODE(code, lino, _end);	94,492,293✔
172	if (pInfo->windowCount != pInfo->windowSliding) {	94,492,293✔
173	if (prevRows <= pInfo->windowSliding) {	40,548✔
174	if (pBuffInfo->winRows > pInfo->windowSliding) {	40,548✔
175	step = pInfo->windowSliding - prevRows;	33,480✔
176	} else {
177	step = pInfo->windowSliding;	7,068✔
178	}
179	} else {
180	step = 0;	×
181	}
182	}
183	if (pBuffInfo->winRows == pInfo->windowCount) {	94,492,293✔
184	doUpdateNumOfRows(pExprSup->pCtx, pInfo->pRow, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	92,842,949✔
185	code = copyResultrowToDataBlock(pExprSup->pExprInfo, pExprSup->numOfExprs, pInfo->pRow, pExprSup->pCtx, pRes,	92,842,949✔
186	pExprSup->rowEntryInfoOffset, pTaskInfo);	92,842,949✔
187	QUERY_CHECK_CODE(code, lino, _end);	92,842,949✔
188	pRes->info.rows += pInfo->pRow->numOfRows;	92,842,949✔
189	clearWinStateBuff(pBuffInfo);	92,842,949✔
190	pInfo->preStateIndex = pInfo->countSup.curStateIndex;	92,842,949✔
191	clearResultRowInitFlag(pExprSup->pCtx, pExprSup->numOfExprs);	92,842,949✔
192	}
193	i += step;	94,492,293✔
194	}
195
196	code = doFilter(pRes, pOperator->exprSupp.pFilterInfo, NULL, NULL);	1,883,506✔
197	QUERY_CHECK_CODE(code, lino, _end);	1,883,506✔
198
199	_end:	1,883,506✔
200	if (code != TSDB_CODE_SUCCESS) {	1,936,316✔
201	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	52,810✔
202	pTaskInfo->code = code;	52,810✔
203	T_LONG_JMP(pTaskInfo->env, code);	52,810✔
204	}
205	}	1,883,506✔
206
207	static void buildCountResult(SExprSupp* pExprSup, SCountWindowSupp* pCountSup, SExecTaskInfo* pTaskInfo,	858,094✔
208	SFilterInfo* pFilterInfo, int32_t preStateIndex, SSDataBlock* pBlock) {
209	SResultRow* pResultRow = NULL;	858,094✔
210	int32_t code = TSDB_CODE_SUCCESS;	858,094✔
211	int32_t lino = 0;	858,094✔
212	int32_t numOfBuff = taosArrayGetSize(pCountSup->pWinStates);	858,094✔
213	int32_t newSize = pBlock->info.rows + numOfBuff;	858,094✔
214	if (newSize > pBlock->info.capacity) {	858,094✔
215	code = blockDataEnsureCapacity(pBlock, newSize);	×
216	QUERY_CHECK_CODE(code, lino, _end);	×
217	}
218	pCountSup->stateIndex = (preStateIndex + 1) % numOfBuff;	858,094✔
219	for (int32_t i = 0; i < numOfBuff; i++) {	1,765,664✔
220	SCountWindowResult* pBuff = NULL;	907,198✔
221	code = setCountWindowOutputBuff(pExprSup, pCountSup, &pResultRow, &pBuff);	907,198✔
222	QUERY_CHECK_CODE(code, lino, _end);	907,198✔
223	if (pBuff->winRows == 0) {	907,198✔
224	continue;	509,161✔
225	}
226	doUpdateNumOfRows(pExprSup->pCtx, pResultRow, pExprSup->numOfExprs, pExprSup->rowEntryInfoOffset);	398,037✔
227	code = copyResultrowToDataBlock(pExprSup->pExprInfo, pExprSup->numOfExprs, pResultRow, pExprSup->pCtx, pBlock,	398,409✔
228	pExprSup->rowEntryInfoOffset, pTaskInfo);	398,409✔
229	QUERY_CHECK_CODE(code, lino, _end);	398,409✔
230	pBlock->info.rows += pResultRow->numOfRows;	398,409✔
231	clearWinStateBuff(pBuff);	398,409✔
232	clearResultRowInitFlag(pExprSup->pCtx, pExprSup->numOfExprs);	398,409✔
233	}
234	code = doFilter(pBlock, pFilterInfo, NULL, NULL);	858,466✔
235	QUERY_CHECK_CODE(code, lino, _end);	858,094✔
236
237	_end:	858,094✔
238	if (code != TSDB_CODE_SUCCESS) {	858,094✔
239	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	×
240	T_LONG_JMP(pTaskInfo->env, code);	×
241	}
242	}	858,094✔
243
244	static int32_t countWindowAggregateNext(SOperatorInfo* pOperator, SSDataBlock** ppRes) {	1,130,222✔
245	int32_t code = TSDB_CODE_SUCCESS;	1,130,222✔
246	int32_t lino = 0;	1,130,222✔
247	SCountWindowOperatorInfo* pInfo = pOperator->info;	1,130,222✔
248	SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;	1,130,222✔
249	SExprSupp* pExprSup = &pOperator->exprSupp;	1,130,222✔
250	int32_t order = pInfo->binfo.inputTsOrder;	1,130,222✔
251	SSDataBlock* pRes = pInfo->binfo.pRes;	1,130,222✔
252
253	blockDataCleanup(pRes);	1,130,222✔
254
255	while (1) {	1,883,506✔
256	SSDataBlock* pBlock = NULL;	3,013,728✔
257	if (pInfo->pPreDataBlock == NULL) {	3,013,728✔
258	pBlock = getNextBlockFromDownstream(pOperator, 0);	2,916,362✔
259	} else {
260	pBlock = pInfo->pPreDataBlock;	97,366✔
261	pInfo->pPreDataBlock = NULL;	97,366✔
262	}
263
264	if (pBlock == NULL) {	3,012,890✔
265	break;	756,148✔
266	}
267
268	pRes->info.scanFlag = pBlock->info.scanFlag;	2,256,742✔
269	code = setInputDataBlock(pExprSup, pBlock, order, MAIN_SCAN, true);	2,257,208✔
270	QUERY_CHECK_CODE(code, lino, _end);	2,257,208✔
271
272	code = blockDataUpdateTsWindow(pBlock, pInfo->tsSlotId);	2,257,208✔
273	QUERY_CHECK_CODE(code, lino, _end);	2,257,208✔
274
275	// there is an scalar expression that needs to be calculated right before apply the group aggregation.
276	if (pInfo->scalarSup.pExprInfo != NULL) {	2,257,208✔
277	code = projectApplyFunctions(pInfo->scalarSup.pExprInfo, pBlock, pBlock, pInfo->scalarSup.pCtx,	473,674✔
278	pInfo->scalarSup.numOfExprs, NULL, GET_STM_RTINFO(pOperator->pTaskInfo));	473,674✔
279	QUERY_CHECK_CODE(code, lino, _end);	473,674✔
280	}
281
282	if (pInfo->groupId == 0) {	2,038,262✔
283	pInfo->groupId = pBlock->info.id.groupId;	1,197,750✔
284	} else if (pInfo->groupId != pBlock->info.id.groupId) {	840,512✔
285	pInfo->pPreDataBlock = pBlock;	101,946✔
286	pRes->info.id.groupId = pInfo->groupId;	101,946✔
287	buildCountResult(pExprSup, &pInfo->countSup, pTaskInfo, pOperator->exprSupp.pFilterInfo, pInfo->preStateIndex, pRes);	101,946✔
288	pInfo->groupId = pBlock->info.id.groupId;	101,946✔
289	pInfo->countSup.lastTs = INT64_MIN;	101,946✔
290	if (pRes->info.rows > 0) {	101,946✔
291	(*ppRes) = pRes;	101,946✔
292	return code;	101,946✔
293	}
294	}
295
296	doCountWindowAggImpl(pOperator, pBlock);	1,936,316✔
297	if (pRes->info.rows >= pOperator->resultInfo.threshold) {	1,883,506✔
298	pRes->info.id.groupId = pInfo->groupId;	×
299	(*ppRes) = pRes;	×
300	return code;	×
301	}
302	}
303
304	pRes->info.id.groupId = pInfo->groupId;	756,148✔
305	buildCountResult(pExprSup, &pInfo->countSup, pTaskInfo, pOperator->exprSupp.pFilterInfo, pInfo->preStateIndex, pRes);	756,520✔
306
307	_end:	975,094✔
308	if (code != TSDB_CODE_SUCCESS) {	975,094✔
309	qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));	218,946✔
310	pTaskInfo->code = code;	218,946✔
311	T_LONG_JMP(pTaskInfo->env, code);	218,946✔
312	}
313	(*ppRes) = pRes->info.rows == 0 ? NULL : pRes;	756,148✔
314	return code;	756,520✔
315	}
316
317	static int32_t resetCountWindowOperatorState(SOperatorInfo* pOper) {	1,446✔
318	SCountWindowOperatorInfo* pCount = pOper->info;	1,446✔
319	SExecTaskInfo* pTaskInfo = pOper->pTaskInfo;	1,446✔
320	SCountWindowPhysiNode* pPhynode = (SCountWindowPhysiNode*)pOper->pPhyNode;	1,446✔
321	pOper->status = OP_NOT_OPENED;	1,446✔
322
323	resetBasicOperatorState(&pCount->binfo);	1,446✔
324	pCount->groupId = 0;	1,446✔
325	pCount->countSup.stateIndex = 0;	1,446✔
326	pCount->countSup.lastTs = INT64_MIN;	1,446✔
327	pCount->pPreDataBlock = NULL;	1,446✔
328	pCount->preStateIndex = 0;	1,446✔
329	pCount->pRow = NULL;	1,446✔
330
331	colDataDestroy(&pCount->twAggSup.timeWindowData);	1,446✔
332	int32_t code = initExecTimeWindowInfo(&pCount->twAggSup.timeWindowData, &pTaskInfo->window);	1,446✔
333
334	if (code == 0) {	1,446✔
335	code = resetAggSup(&pOper->exprSupp, &pCount->aggSup, pTaskInfo, pPhynode->window.pFuncs, NULL,	2,892✔
336	sizeof(int64_t) * 2 + POINTER_BYTES, pTaskInfo->id.str, pTaskInfo->streamInfo.pState,	1,446✔
337	&pTaskInfo->storageAPI.functionStore);
338	}
339	if (code == 0) {	1,446✔
340	code = resetExprSupp(&pCount->scalarSup, pTaskInfo, pPhynode->window.pExprs, NULL,	1,446✔
341	&pTaskInfo->storageAPI.functionStore);
342	}
343	return code;	1,446✔
344	}
345
346	int32_t createCountwindowOperatorInfo(SOperatorInfo* downstream, SPhysiNode* physiNode,	774,999✔
347	SExecTaskInfo* pTaskInfo, SOperatorInfo** pOptrInfo) {
348	QRY_PARAM_CHECK(pOptrInfo);	774,999✔
349
350	int32_t code = TSDB_CODE_SUCCESS;	774,999✔
351	int32_t lino = 0;	774,999✔
352	SCountWindowOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SCountWindowOperatorInfo));	774,999✔
353	SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));	774,612✔
354	if (pInfo == NULL \|\| pOperator == NULL) {	774,999✔
UNCOV 355	code = terrno;	×
356	goto _error;	×
357	}
358
359	pOperator->pPhyNode = physiNode;	774,999✔
360	pOperator->exprSupp.hasWindowOrGroup = true;	774,999✔
361	pOperator->exprSupp.hasWindow = true;	774,999✔
362
363	SCountWindowPhysiNode* pCountWindowNode = (SCountWindowPhysiNode*)physiNode;	774,999✔
364
365	pInfo->tsSlotId = ((SColumnNode*)pCountWindowNode->window.pTspk)->slotId;	774,999✔
366
367	if (pCountWindowNode->window.pExprs != NULL) {	774,999✔
368	int32_t numOfScalarExpr = 0;	619,820✔
369	SExprInfo* pScalarExprInfo = NULL;	619,820✔
370	code = createExprInfo(pCountWindowNode->window.pExprs, NULL, &pScalarExprInfo, &numOfScalarExpr);	619,820✔
371	QUERY_CHECK_CODE(code, lino, _error);	619,820✔
372	code = initExprSupp(&pInfo->scalarSup, pScalarExprInfo, numOfScalarExpr, &pTaskInfo->storageAPI.functionStore);	619,820✔
373	QUERY_CHECK_CODE(code, lino, _error);	619,820✔
374	}
375
376	size_t keyBufSize = 0;	774,999✔
377	int32_t num = 0;	774,999✔
378	SExprInfo* pExprInfo = NULL;	774,999✔
379	code = createExprInfo(pCountWindowNode->window.pFuncs, NULL, &pExprInfo, &num);	774,999✔
380	QUERY_CHECK_CODE(code, lino, _error);	774,999✔
381
382	initResultSizeInfo(&pOperator->resultInfo, 4096);	774,999✔
383
384	code = initAggSup(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str,	973,758✔
385	pTaskInfo->streamInfo.pState, &pTaskInfo->storageAPI.functionStore);	774,999✔
386	QUERY_CHECK_CODE(code, lino, _error);	774,999✔
387
388	SSDataBlock* pResBlock = createDataBlockFromDescNode(pCountWindowNode->window.node.pOutputDataBlockDesc);	774,999✔
389	QUERY_CHECK_NULL(pResBlock, code, lino, _error, terrno);	774,999✔
390	initBasicInfo(&pInfo->binfo, pResBlock);	774,999✔
391
392	code = blockDataEnsureCapacity(pResBlock, pOperator->resultInfo.capacity);	774,627✔
393	QUERY_CHECK_CODE(code, lino, _error);	774,999✔
394
395	initResultRowInfo(&pInfo->binfo.resultRowInfo);	774,999✔
396	pInfo->binfo.inputTsOrder = physiNode->inputTsOrder;	774,999✔
397	pInfo->binfo.outputTsOrder = physiNode->outputTsOrder;	774,999✔
398	pInfo->windowCount = pCountWindowNode->windowCount;	774,999✔
399	pInfo->windowSliding = pCountWindowNode->windowSliding;	774,999✔
400	// sizeof(SCountWindowResult)
401	int32_t itemSize = sizeof(int32_t) + pInfo->aggSup.resultRowSize;	774,999✔
402	int32_t numOfItem = 1;	774,999✔
403	if (pInfo->windowCount != pInfo->windowSliding) {	774,999✔
404	numOfItem = pInfo->windowCount / pInfo->windowSliding + 1;	15,624✔
405	}
406
407	pInfo->countSup.pWinStates = taosArrayInit_s(itemSize, numOfItem);	774,999✔
408	if (!pInfo->countSup.pWinStates) {	774,999✔
409	goto _error;	×
410	}
411
412	pInfo->countSup.stateIndex = 0;	774,999✔
413	pInfo->pPreDataBlock = NULL;	774,999✔
414	pInfo->preStateIndex = 0;	774,999✔
415	pInfo->countSup.lastTs = INT64_MIN;	774,999✔
416
417	code = filterInitFromNode((SNode*)pCountWindowNode->window.node.pConditions, &pOperator->exprSupp.pFilterInfo, 0,	774,999✔
418	pTaskInfo->pStreamRuntimeInfo);	774,999✔
419	QUERY_CHECK_CODE(code, lino, _error);	774,999✔
420
421	code = initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);	774,999✔
422	QUERY_CHECK_CODE(code, lino, _error);	774,999✔
423
424	setOperatorInfo(pOperator, "CountWindowOperator", QUERY_NODE_PHYSICAL_PLAN_MERGE_COUNT, true, OP_NOT_OPENED, pInfo,	774,999✔
425	pTaskInfo);
426	pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, countWindowAggregateNext, NULL, destroyCountWindowOperatorInfo,	774,999✔
427	optrDefaultBufFn, NULL, optrDefaultGetNextExtFn, NULL);
428	setOperatorResetStateFn(pOperator, resetCountWindowOperatorState);	774,999✔
429	code = appendDownstream(pOperator, &downstream, 1);	774,999✔
430	if (code != TSDB_CODE_SUCCESS) {	774,999✔
431	goto _error;	×
432	}
433
434	*pOptrInfo = pOperator;	774,999✔
435	return TSDB_CODE_SUCCESS;	774,999✔
436
437	_error:	×
438	if (pInfo != NULL) {	×
439	destroyCountWindowOperatorInfo(pInfo);	×
440	}
441
442	destroyOperatorAndDownstreams(pOperator, &downstream, 1);	×
443	pTaskInfo->code = code;	×
444	return code;	×
445	}

taosdata / TDengine / #4940

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