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

Merge pull request #24967 from taosdata/fix/TD-28859 

reset flush state
This commit is contained in:
Haojun Liao 2024-03-04 15:51:36 +08:00 committed by GitHub
parent 8418e5de21 01d555aabf
commit e7f35f5dcf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 15 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -156,6 +156,7 @@ int32_t getSessionWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey,
(*pVal) = pPos; (*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey; SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true; pPos->beUsed = true;
pPos->beFlushed = false;
*pKey = *pDestWinKey; *pKey = *pDestWinKey;
goto _end; goto _end;
} }
@ -167,6 +168,7 @@ int32_t getSessionWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey,
(*pVal) = pPos; (*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey; SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true; pPos->beUsed = true;
pPos->beFlushed = false;
*pKey = *pDestWinKey; *pKey = *pDestWinKey;
goto _end; goto _end;
} }
@ -380,6 +382,14 @@ static SStreamStateCur* seekKeyCurrentPrev_buff(SStreamFileState* pFileState, co
(*pWins) = pWinStates; (*pWins) = pWinStates;
} }
if (size > 0 && index == -1) {
SRowBuffPos* pPos = taosArrayGetP(pWinStates, 0);
SSessionKey* pWin = (SSessionKey*)pPos->pKey;
if (pWinKey->win.skey == pWin->win.skey) {
index = 0;
}
}
if (index >= 0) { if (index >= 0) {
pCur = createSessionStateCursor(pFileState); pCur = createSessionStateCursor(pFileState);
pCur->buffIndex = index; pCur->buffIndex = index;
@ -387,6 +397,7 @@ static SStreamStateCur* seekKeyCurrentPrev_buff(SStreamFileState* pFileState, co
*pIndex = index; *pIndex = index;
} }
} }
return pCur; return pCur;
} }
@ -666,6 +677,7 @@ int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, ch
(*pVal) = pPos; (*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey; SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true; pPos->beUsed = true;
pPos->beFlushed = false;
*key = *pDestWinKey; *key = *pDestWinKey;
goto _end; goto _end;
} }
@ -679,6 +691,7 @@ int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, ch
(*pVal) = pPos; (*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey; SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true; pPos->beUsed = true;
pPos->beFlushed = false;
*key = *pDestWinKey; *key = *pDestWinKey;
goto _end; goto _end;
} }
@ -771,6 +784,7 @@ int32_t getCountWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, C
(*pVal) = pPos; (*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey; SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true; pPos->beUsed = true;
pPos->beFlushed = false;
*pWinKey = *pDestWinKey; *pWinKey = *pDestWinKey;
goto _end; goto _end;
} }
@ -799,6 +813,7 @@ int32_t getCountWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, C
(*pVal) = pPos; (*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey; SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true; pPos->beUsed = true;
pPos->beFlushed = false;
*pWinKey = *pDestWinKey; *pWinKey = *pDestWinKey;
goto _end; goto _end;
} }