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

Merge pull request #26723 from taosdata/fix/TD-31071 

fix(stream):fix issue
This commit is contained in:
Haojun Liao 2024-07-23 11:38:21 +08:00 committed by GitHub
parent 3f79e5b7e3 f589cc6894
commit 12ffdef6b0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 22 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -482,7 +482,8 @@ void doStreamCountSaveCheckpoint(SOperatorInfo* pOperator) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
QUERY_CHECK_CODE(code, lino, _end); QUERY_CHECK_CODE(code, lino, _end);
} }
len = doStreamCountEncodeOpState(&pBuf, len, pOperator, true); void* pTmpBuf = pBuf;
len = doStreamCountEncodeOpState(&pTmpBuf, len, pOperator, true);
pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_COUNT_OP_CHECKPOINT_NAME, pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_COUNT_OP_CHECKPOINT_NAME,
strlen(STREAM_COUNT_OP_CHECKPOINT_NAME), pBuf, len); strlen(STREAM_COUNT_OP_CHECKPOINT_NAME), pBuf, len);
saveStreamOperatorStateComplete(&pInfo->basic); saveStreamOperatorStateComplete(&pInfo->basic);

4
source/libs/executor/src/streameventwindowoperator.c
View File

@ -542,6 +542,10 @@ void doStreamEventSaveCheckpoint(SOperatorInfo* pOperator) {
if (needSaveStreamOperatorInfo(&pInfo->basic)) { if (needSaveStreamOperatorInfo(&pInfo->basic)) {
int32_t len = doStreamEventEncodeOpState(NULL, 0, pOperator); int32_t len = doStreamEventEncodeOpState(NULL, 0, pOperator);
void* buf = taosMemoryCalloc(1, len); void* buf = taosMemoryCalloc(1, len);
if (!buf) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
return;
}
void* pBuf = buf; void* pBuf = buf;
len = doStreamEventEncodeOpState(&pBuf, len, pOperator); len = doStreamEventEncodeOpState(&pBuf, len, pOperator);
pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_EVENT_OP_CHECKPOINT_NAME, pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_EVENT_OP_CHECKPOINT_NAME,

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

@ -1410,6 +1410,10 @@ void doStreamIntervalSaveCheckpoint(SOperatorInfo* pOperator) {
if (needSaveStreamOperatorInfo(&pInfo->basic)) { if (needSaveStreamOperatorInfo(&pInfo->basic)) {
int32_t len = doStreamIntervalEncodeOpState(NULL, 0, pOperator); int32_t len = doStreamIntervalEncodeOpState(NULL, 0, pOperator);
void* buf = taosMemoryCalloc(1, len); void* buf = taosMemoryCalloc(1, len);
if (!buf) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
return;
}
void* pBuf = buf; void* pBuf = buf;
len = doStreamIntervalEncodeOpState(&pBuf, len, pOperator); len = doStreamIntervalEncodeOpState(&pBuf, len, pOperator);
pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_INTERVAL_OP_CHECKPOINT_NAME, pInfo->stateStore.streamStateSaveInfo(pInfo->pState, STREAM_INTERVAL_OP_CHECKPOINT_NAME,
@ -3193,6 +3197,10 @@ void doStreamSessionSaveCheckpoint(SOperatorInfo* pOperator) {
if (needSaveStreamOperatorInfo(&pInfo->basic)) { if (needSaveStreamOperatorInfo(&pInfo->basic)) {
int32_t len = doStreamSessionEncodeOpState(NULL, 0, pOperator, true); int32_t len = doStreamSessionEncodeOpState(NULL, 0, pOperator, true);
void* buf = taosMemoryCalloc(1, len); void* buf = taosMemoryCalloc(1, len);
if (!buf) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
return;
}
void* pBuf = buf; void* pBuf = buf;
len = doStreamSessionEncodeOpState(&pBuf, len, pOperator, true); len = doStreamSessionEncodeOpState(&pBuf, len, pOperator, true);
pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_SESSION_OP_CHECKPOINT_NAME, pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_SESSION_OP_CHECKPOINT_NAME,
@ -4401,6 +4409,10 @@ void doStreamStateSaveCheckpoint(SOperatorInfo* pOperator) {
if (needSaveStreamOperatorInfo(&pInfo->basic)) { if (needSaveStreamOperatorInfo(&pInfo->basic)) {
int32_t len = doStreamStateEncodeOpState(NULL, 0, pOperator, true); int32_t len = doStreamStateEncodeOpState(NULL, 0, pOperator, true);
void* buf = taosMemoryCalloc(1, len); void* buf = taosMemoryCalloc(1, len);
if (!buf) {
qError("%s failed at line %d since %s", __func__, __LINE__, tstrerror(TSDB_CODE_OUT_OF_MEMORY));
return;
}
void* pBuf = buf; void* pBuf = buf;
len = doStreamStateEncodeOpState(&pBuf, len, pOperator, true); len = doStreamStateEncodeOpState(&pBuf, len, pOperator, true);
pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_STATE_OP_CHECKPOINT_NAME, pInfo->streamAggSup.stateStore.streamStateSaveInfo(pInfo->streamAggSup.pState, STREAM_STATE_OP_CHECKPOINT_NAME,