xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: reset prefetched datablock after it is used

This commit is contained in:
shenglian zhou 2022-06-22 10:47:23 +08:00
parent 557d69d14a
commit 580a22d327
2 changed files with 7 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
source/libs/executor/src/timewindowoperator.c
View File

@ -3722,7 +3722,7 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
win.ekey = win.ekey =
taosTimeAdd(win.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1; taosTimeAdd(win.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, iaInfo->interval.precision) - 1;
// TODO: remove the hash table usage (groupid + winkey => result row position) // TODO: remove the hash table (groupid + winkey => result row position)
int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo); pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
if (ret != TSDB_CODE_SUCCESS || pResult == NULL) { if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
@ -3748,9 +3748,10 @@ static void doMergeIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo*
currTs = tsCols[currPos]; currTs = tsCols[currPos];
currWin.skey = currTs; currWin.skey = currTs;
currWin.ekey = taosTimeAdd(currWin.skey, iaInfo->interval.interval, iaInfo->interval.intervalUnit, currWin.ekey = taosTimeAdd(currWin.skey,
iaInfo->interval.precision) - iaInfo->interval.interval,
1; iaInfo->interval.intervalUnit,
iaInfo->interval.precision) - 1;
startPos = currPos; startPos = currPos;
ret = setTimeWindowOutputBuf(pResultRowInfo, &currWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, ret = setTimeWindowOutputBuf(pResultRowInfo, &currWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo); pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &iaInfo->aggSup, pTaskInfo);
@ -3790,6 +3791,7 @@ static SSDataBlock* doMergeIntervalAgg(SOperatorInfo* pOperator) {
} else { } else {
pBlock = miaInfo->prefetchedBlock; pBlock = miaInfo->prefetchedBlock;
miaInfo->groupId = pBlock->info.groupId; miaInfo->groupId = pBlock->info.groupId;
miaInfo->prefetchedBlock = NULL;
} }
if (pBlock == NULL) { if (pBlock == NULL) {