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

fix stream backend convert

This commit is contained in:
yihaoDeng 2023-11-16 14:41:04 +08:00
parent cee33ffb95
commit f5ef2e727f
3 changed files with 26 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/libs/stream/inc/streamBackendRocksdb.h
View File

@ -128,6 +128,7 @@ typedef struct {
TdThreadRwlock rwLock;
} SBkdMgt;
bool streamBackendDataIsExist(const char* path, int64_t chkpId, int32_t vgId);
void* streamBackendInit(const char* path, int64_t chkpId, int32_t vgId);
void streamBackendCleanup(void* arg);
void streamBackendHandleCleanup(void* arg);

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

@ -308,6 +308,16 @@ int32_t rebuildDirFromChkp2(const char* path, char* key, int64_t chkpId, char**
return 0;
}
bool streamBackendDataIsExist(const char* path, int64_t chkpId, int32_t vgId) {
bool exist = true;
char* state = taosMemoryCalloc(1, strlen(path) + 32);
sprintf(state, "%s%s%s", path, TD_DIRSEP, "state");
if (!taosDirExist(state)) {
exist = false;
}
taosMemoryFree(state);
return exist;
}
void* streamBackendInit(const char* streamPath, int64_t chkpId, int32_t vgId) {
char* backendPath = NULL;
int32_t code = rebuildDirFromCheckpoint(streamPath, chkpId, &backendPath);

13
source/libs/stream/src/streamMeta.c
View File

@ -190,6 +190,11 @@ int32_t streamMetaCheckBackendCompatible(SStreamMeta* pMeta) {
int32_t streamMetaCvtDbFormat(SStreamMeta* pMeta) {
int32_t code = 0;
int64_t chkpId = streamMetaGetLatestCheckpointId(pMeta);
bool exist = streamBackendDataIsExist(pMeta->path, chkpId, pMeta->vgId);
if (exist == false) {
return code;
}
SBackendWrapper* pBackend = streamBackendInit(pMeta->path, chkpId, pMeta->vgId);
void* pIter = taosHashIterate(pBackend->cfInst, NULL);
@ -206,6 +211,14 @@ int32_t streamMetaCvtDbFormat(SStreamMeta* pMeta) {
_EXIT:
streamBackendCleanup((void*)pBackend);
if (code == 0) {
char* state = taosMemoryCalloc(1, strlen(pMeta->path) + 32);
sprintf(state, "%s%s%s", pMeta->path, TD_DIRSEP, "state");
taosRemoveDir(state);
taosMemoryFree(state);
}
return code;
}
int32_t streamMetaMayCvtDbFormat(SStreamMeta* pMeta) {