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

opt force window close memory

This commit is contained in:
54liuyao 2024-12-03 09:11:46 +08:00
parent 803ed61ca0
commit 2a8263efea
2 changed files with 19 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
source/libs/executor/src/streamfilloperator.c
View File

@ -987,7 +987,14 @@ _end:
} }
void resetStreamFillSup(SStreamFillSupporter* pFillSup) { void resetStreamFillSup(SStreamFillSupporter* pFillSup) {
tSimpleHashClear(pFillSup->pResMap); _hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
SSHashObj* pNewMap = tSimpleHashInit(16, hashFn);
if (pNewMap != NULL) {
tSimpleHashCleanup(pFillSup->pResMap);
pFillSup->pResMap = pNewMap;
} else {
tSimpleHashClear(pFillSup->pResMap);
}
pFillSup->hasDelete = false; pFillSup->hasDelete = false;
} }
void resetStreamFillInfo(SStreamFillOperatorInfo* pInfo) { void resetStreamFillInfo(SStreamFillOperatorInfo* pInfo) {
@ -1406,6 +1413,7 @@ static int32_t doStreamForceFillNext(SOperatorInfo* pOperator, SSDataBlock** ppR
} }
pInfo->stateStore.streamStateClearExpiredState(pInfo->pState); pInfo->stateStore.streamStateClearExpiredState(pInfo->pState);
resetStreamFillInfo(pInfo);
setStreamOperatorCompleted(pOperator); setStreamOperatorCompleted(pOperator);
(*ppRes) = NULL; (*ppRes) = NULL;
goto _end; goto _end;
@ -1477,6 +1485,7 @@ static int32_t doStreamForceFillNext(SOperatorInfo* pOperator, SSDataBlock** ppR
if ((*ppRes) == NULL) { if ((*ppRes) == NULL) {
pInfo->stateStore.streamStateClearExpiredState(pInfo->pState); pInfo->stateStore.streamStateClearExpiredState(pInfo->pState);
resetStreamFillInfo(pInfo);
setStreamOperatorCompleted(pOperator); setStreamOperatorCompleted(pOperator);
} }

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

@ -1228,6 +1228,8 @@ void setFillInfo(SStreamFileState* pFileState) {
} }
void clearExpiredState(SStreamFileState* pFileState) { void clearExpiredState(SStreamFileState* pFileState) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0;
SSHashObj* pSearchBuff = pFileState->searchBuff; SSHashObj* pSearchBuff = pFileState->searchBuff;
void* pIte = NULL; void* pIte = NULL;
int32_t iter = 0; int32_t iter = 0;
@ -1246,6 +1248,13 @@ void clearExpiredState(SStreamFileState* pFileState) {
} }
taosArrayRemoveBatch(pWinStates, 0, size - 1, NULL); taosArrayRemoveBatch(pWinStates, 0, size - 1, NULL);
} }
code = clearRowBuff(pFileState);
QUERY_CHECK_CODE(code, lino, _end);
_end:
if (code != TSDB_CODE_SUCCESS) {
qError("%s failed at line %d since %s", __func__, lino, tstrerror(code));
}
} }
int32_t getStateSearchRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, void** pVal, int32_t* pVLen, int32_t getStateSearchRowBuff(SStreamFileState* pFileState, const SWinKey* pKey, void** pVal, int32_t* pVLen,