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

rebuild sliding window

This commit is contained in:
54liuyao 2024-02-23 09:05:30 +08:00
parent a7b1ee9d59
commit c7a50acd63
2 changed files with 4 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
source/libs/executor/src/streamcountwindowoperator.c
View File

@ -99,6 +99,9 @@ void setCountOutputBuf(SStreamAggSupporter* pAggSup, TSKEY ts, uint64_t groupId,
(void**)&pCurWin->winInfo.pStatePos, &size); (void**)&pCurWin->winInfo.pStatePos, &size);
} }
} }
if (ts < pCurWin->winInfo.sessionWin.win.ekey) {
pBuffInfo->rebuildWindow = true;
}
} else { } else {
code = pAggSup->stateStore.streamStateCountWinAddIfNotExist( code = pAggSup->stateStore.streamStateCountWinAddIfNotExist(
pAggSup->pState, &pCurWin->winInfo.sessionWin, pAggSup->windowCount, (void**)&pCurWin->winInfo.pStatePos, &size); pAggSup->pState, &pCurWin->winInfo.sessionWin, pAggSup->windowCount, (void**)&pCurWin->winInfo.pStatePos, &size);

1
source/libs/stream/src/streamSessionState.c
View File

@ -488,6 +488,7 @@ SStreamStateCur* countWinStateSeekKeyPrev(SStreamFileState* pFileState, const SS
void* pFileStore = getStateFileStore(pFileState); void* pFileStore = getStateFileStore(pFileState);
SStreamStateCur* pCur = streamStateSessionSeekKeyPrev_rocksdb(pFileStore, pWinKey); SStreamStateCur* pCur = streamStateSessionSeekKeyPrev_rocksdb(pFileStore, pWinKey);
if (pCur) { if (pCur) {
pCur->pStreamFileState = pFileState;
SSessionKey key = {0}; SSessionKey key = {0};
void* pVal = NULL; void* pVal = NULL;
int len = 0; int len = 0;