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

reset flush state

This commit is contained in:
54liuyao 2024-03-01 10:57:57 +08:00
parent 818b9d1564
commit 3febcb96a0
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;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true;
pPos->beFlushed = false;
*pKey = *pDestWinKey;
goto _end;
}
@ -167,6 +168,7 @@ int32_t getSessionWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey,
(*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true;
pPos->beFlushed = false;
*pKey = *pDestWinKey;
goto _end;
}
@ -380,6 +382,14 @@ static SStreamStateCur* seekKeyCurrentPrev_buff(SStreamFileState* pFileState, co
(*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) {
pCur = createSessionStateCursor(pFileState);
pCur->buffIndex = index;
@ -387,6 +397,7 @@ static SStreamStateCur* seekKeyCurrentPrev_buff(SStreamFileState* pFileState, co
*pIndex = index;
}
}
return pCur;
}
@ -666,6 +677,7 @@ int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, ch
(*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true;
pPos->beFlushed = false;
*key = *pDestWinKey;
goto _end;
}
@ -679,6 +691,7 @@ int32_t getStateWinResultBuff(SStreamFileState* pFileState, SSessionKey* key, ch
(*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true;
pPos->beFlushed = false;
*key = *pDestWinKey;
goto _end;
}
@ -771,6 +784,7 @@ int32_t getCountWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, C
(*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true;
pPos->beFlushed = false;
*pWinKey = *pDestWinKey;
goto _end;
}
@ -799,6 +813,7 @@ int32_t getCountWinResultBuff(SStreamFileState* pFileState, SSessionKey* pKey, C
(*pVal) = pPos;
SSessionKey* pDestWinKey = (SSessionKey*)pPos->pKey;
pPos->beUsed = true;
pPos->beFlushed = false;
*pWinKey = *pDestWinKey;
goto _end;
}