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

support reopen stream state

This commit is contained in:
yihaoDeng 2023-08-09 09:19:03 +00:00
parent f3cf907d5a
commit 6fdcd82a33
1 changed files with 7 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -34,6 +34,8 @@ typedef struct {
SArray* pAdd; SArray* pAdd;
SArray* pDel; SArray* pDel;
int8_t update;
} SBackendManager; } SBackendManager;
typedef struct SCompactFilteFactory { typedef struct SCompactFilteFactory {
@ -155,6 +157,7 @@ SBackendManager* backendManagerCreate(char* path) {
p->pAdd = taosArrayInit(64, sizeof(void*)); p->pAdd = taosArrayInit(64, sizeof(void*));
p->pDel = taosArrayInit(64, sizeof(void*)); p->pDel = taosArrayInit(64, sizeof(void*));
p->update = 0;
return p; return p;
} }
void backendManagerDestroy(SBackendManager* bm) { void backendManagerDestroy(SBackendManager* bm) {
@ -254,12 +257,16 @@ int32_t backendManagerGetDelta(SBackendManager* bm, int64_t chkpId, SArray* list
} }
pIter = taosHashIterate(pTable, pIter); pIter = taosHashIterate(pTable, pIter);
} }
bm->update = 1;
} else { } else {
int32_t code = compareHashTable(bm->pSSTable, pTable, bm->pAdd, bm->pDel); int32_t code = compareHashTable(bm->pSSTable, pTable, bm->pAdd, bm->pDel);
bm->curChkpId = chkpId; bm->curChkpId = chkpId;
taosHashCleanup(pTable); taosHashCleanup(pTable);
if (taosArrayGetSize(bm->pAdd) == 0 && taosArrayGetSize(bm->pDel) == 0) {
bm->update = 0;
}
} }
return 0; return 0;
} }