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

fix(stream): make sure open operator only once for scan-history task.

This commit is contained in:
Haojun Liao 2023-11-30 16:10:09 +08:00
parent 999afd97e5
commit 30d78c6be1
2 changed files with 6 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -402,7 +402,8 @@ typedef struct SHistoryTaskInfo {
int32_t tickCount; int32_t tickCount;
int32_t retryTimes; int32_t retryTimes;
int32_t waitInterval; int32_t waitInterval;
int64_t haltVer; // offset in wal when halt the stream task int64_t haltVer; // offset in wal when halt the stream task
bool operatorOpen; // false by default
} SHistoryTaskInfo; } SHistoryTaskInfo;
typedef struct STaskOutputInfo { typedef struct STaskOutputInfo {

5
source/libs/stream/src/streamExec.c
View File

@ -256,7 +256,10 @@ SScanhistoryDataInfo streamScanHistoryData(SStreamTask* pTask, int64_t st) {
bool finished = false; bool finished = false;
const char* id = pTask->id.idStr; const char* id = pTask->id.idStr;
qSetStreamOpOpen(exec); if (!pTask->hTaskInfo.operatorOpen) {
qSetStreamOpOpen(exec);
pTask->hTaskInfo.operatorOpen = true;
}
while (1) { while (1) {
if (streamTaskShouldPause(pTask)) { if (streamTaskShouldPause(pTask)) {