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

session state recover

This commit is contained in:
liuyao 2023-10-13 11:32:35 +08:00
parent 9b3fa72fea
commit 429125be93
3 changed files with 25 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/libs/stream/tstreamFileState.h
View File

@ -79,6 +79,9 @@ int32_t getSessionFlushedBuff(SStreamFileState* pFileState, SSessionKey* pKey, v
int32_t deleteSessionWinStateBuffFn(void* pBuff, const void *key, size_t keyLen);
int32_t deleteSessionWinStateBuffByPosFn(SStreamFileState* pFileState, SRowBuffPos* pPos);
SRowBuffPos* createSessionWinBuff(SStreamFileState* pFileState, SSessionKey* pKey, void* p, int32_t* pVLen);
int32_t recoverSesssion(SStreamFileState* pFileState, int64_t ckId);
void sessionWinStateClear(SStreamFileState* pFileState);
void sessionWinStateCleanup(void* pBuff);

1
source/libs/executor/src/streamtimewindowoperator.c
View File

@ -3590,7 +3590,6 @@ void streamStateReloadState(SOperatorInfo* pOperator) {
for (int32_t i = 0; i < num; i++) {
SStateWindowInfo curInfo = {0};
SStateWindowInfo nextInfo = {0};
SStateWindowInfo dummy = {0};
qDebug("===stream=== reload state. try process result %" PRId64 ", %" PRIu64 ", index:%d", pSeKeyBuf[i].win.skey,
pSeKeyBuf[i].groupId, i);
getStateWindowInfoByKey(pAggSup, pSeKeyBuf + i, &curInfo, &nextInfo);

26
source/libs/stream/src/tstreamFileState.c
View File

@ -177,7 +177,10 @@ SStreamFileState* streamFileStateInit(int64_t memSize, uint32_t keySize, uint32_
// todo(liuyao) optimize
if (type == STREAM_STATE_BUFF_HASH) {
recoverSnapshot(pFileState, checkpointId);
} else {
recoverSesssion(pFileState, checkpointId);
}
return pFileState;
_error:
@ -642,12 +645,24 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
}
int32_t recoverSesssion(SStreamFileState* pFileState, int64_t ckId) {
int code = TSDB_CODE_SUCCESS;
int code = TSDB_CODE_SUCCESS;
if (pFileState->maxTs != INT64_MIN) {
int64_t mark = (INT64_MIN + pFileState->deleteMark >= pFileState->maxTs)
? INT64_MIN
: pFileState->maxTs - pFileState->deleteMark;
deleteExpiredCheckPoint(pFileState, mark);
}
SStreamStateCur* pCur = streamStateSessionSeekToLast_rocksdb(pFileState->pFileStore);
if (pCur == NULL) {
return -1;
}
int32_t recoverNum = TMIN(MIN_NUM_OF_ROW_BUFF, pFileState->maxRowCount);
while (code == TSDB_CODE_SUCCESS) {
if (pFileState->curRowCount >= recoverNum) {
break;
}
void* pVal = NULL;
int32_t vlen = 0;
SSessionKey key = {0};
@ -655,12 +670,14 @@ int32_t recoverSesssion(SStreamFileState* pFileState, int64_t ckId) {
if (code != 0) {
break;
}
taosMemoryFree(pVal);
SRowBuffPos* pPos = createSessionWinBuff(pFileState, &key, pVal, &vlen);
putSessionWinResultBuff(pFileState, pPos);
code = streamStateSessionCurPrev_rocksdb(pCur);
}
streamStateFreeCur(pCur);
return code;
}
int32_t recoverSnapshot(SStreamFileState* pFileState, int64_t ckId) {
int32_t code = TSDB_CODE_SUCCESS;
if (pFileState->maxTs != INT64_MIN) {
@ -674,11 +691,12 @@ int32_t recoverSnapshot(SStreamFileState* pFileState, int64_t ckId) {
if (pCur == NULL) {
return -1;
}
int32_t recoverNum = TMIN(MIN_NUM_OF_ROW_BUFF, pFileState->maxRowCount);
while (code == TSDB_CODE_SUCCESS) {
if (pFileState->curRowCount == pFileState->maxRowCount) {
if (pFileState->curRowCount >= recoverNum) {
break;
}
void* pVal = NULL;
int32_t vlen = 0;
SRowBuffPos* pNewPos = getNewRowPosForWrite(pFileState);