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

reset close window state

This commit is contained in:
54liuyao 2024-02-25 19:56:38 +08:00
parent 27219f70af
commit 9d560c494d
1 changed files with 2 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -290,6 +290,7 @@ static void doStreamCountAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSDataBl
curWin.winInfo.pStatePos->beUpdated = true; curWin.winInfo.pStatePos->beUpdated = true;
SSessionKey key = {0}; SSessionKey key = {0};
getSessionHashKey(&curWin.winInfo.sessionWin, &key); getSessionHashKey(&curWin.winInfo.sessionWin, &key);
qDebug("===stream=== window close %" PRId64, key.win.skey);
tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo, sizeof(SResultWindowInfo)); tSimpleHashPut(pAggSup->pResultRows, &key, sizeof(SSessionKey), &curWin.winInfo, sizeof(SResultWindowInfo));
} }
@ -345,7 +346,7 @@ int32_t doStreamCountEncodeOpState(void** buf, int32_t len, SOperatorInfo* pOper
size_t keyLen = 0; size_t keyLen = 0;
int32_t iter = 0; int32_t iter = 0;
while ((pIte = tSimpleHashIterate(pInfo->streamAggSup.pResultRows, pIte, &iter)) != NULL) { while ((pIte = tSimpleHashIterate(pInfo->streamAggSup.pResultRows, pIte, &iter)) != NULL) {
void* key = taosHashGetKey(pIte, &keyLen); void* key = tSimpleHashGetKey(pIte, &keyLen);
tlen += encodeSSessionKey(buf, key); tlen += encodeSSessionKey(buf, key);
tlen += encodeSResultWindowInfo(buf, pIte, pInfo->streamAggSup.resultRowSize); tlen += encodeSResultWindowInfo(buf, pIte, pInfo->streamAggSup.resultRowSize);
} }