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-17 15:39:03 +08:00
parent 1dbdb32dd7
commit ccc74ef558
1 changed files with 12 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -321,8 +321,9 @@ int32_t rebuildFromRemoteChkp(char* key, char* chkpPath, int64_t chkpId, char* d
int32_t len = strlen(defaultPath) + 32; int32_t len = strlen(defaultPath) + 32;
char* tmp = taosMemoryCalloc(1, len); char* tmp = taosMemoryCalloc(1, len);
sprintf(tmp, "%s%s%s", defaultPath, TD_DIRSEP, "_tmp"); sprintf(tmp, "%s%s", defaultPath, "_tmp");
if (taosIsDir(tmp)) taosRemoveDir(tmp); if (taosIsDir(tmp)) taosRemoveDir(tmp);
if (taosIsDir(defaultPath)) taosRenameFile(defaultPath, tmp);
SArray* list = taosArrayInit(2, sizeof(void*)); SArray* list = taosArrayInit(2, sizeof(void*));
code = remoteChkp_readMetaData(chkpPath, list); code = remoteChkp_readMetaData(chkpPath, list);
@ -332,9 +333,6 @@ int32_t rebuildFromRemoteChkp(char* key, char* chkpPath, int64_t chkpId, char* d
taosArrayDestroyP(list, taosMemoryFree); taosArrayDestroyP(list, taosMemoryFree);
if (code == 0) { if (code == 0) {
if (taosIsDir(defaultPath)) {
taosRenameFile(defaultPath, tmp);
}
taosMkDir(defaultPath); taosMkDir(defaultPath);
code = copyFiles(chkpPath, defaultPath); code = copyFiles(chkpPath, defaultPath);
} }
@ -342,7 +340,11 @@ int32_t rebuildFromRemoteChkp(char* key, char* chkpPath, int64_t chkpId, char* d
if (code != 0) { if (code != 0) {
if (taosIsDir(defaultPath)) taosRemoveDir(defaultPath); if (taosIsDir(defaultPath)) taosRemoveDir(defaultPath);
if (taosIsDir(tmp)) taosRenameFile(tmp, defaultPath); if (taosIsDir(tmp)) taosRenameFile(tmp, defaultPath);
} else {
taosRemoveDir(tmp);
} }
taosMemoryFree(tmp);
return code; return code;
} }
@ -350,15 +352,15 @@ int32_t rebuildFromLocalChkp(char* key, char* chkpPath, int64_t chkpId, char* de
int32_t code = -1; int32_t code = -1;
int32_t len = strlen(defaultPath) + 32; int32_t len = strlen(defaultPath) + 32;
char* tmp = taosMemoryCalloc(1, len); char* tmp = taosMemoryCalloc(1, len);
sprintf(tmp, "%s%s%s", defaultPath, TD_DIRSEP, "_tmp"); sprintf(tmp, "%s%s", defaultPath, "_tmp");
if (taosIsDir(tmp)) taosRemoveDir(tmp);
if (taosIsDir(defaultPath)) taosRenameFile(defaultPath, tmp);
if (taosIsDir(chkpPath) && isValidCheckpoint(chkpPath)) { if (taosIsDir(chkpPath) && isValidCheckpoint(chkpPath)) {
if (taosIsDir(tmp)) { if (taosIsDir(tmp)) {
taosRemoveDir(tmp); taosRemoveDir(tmp);
} }
if (taosIsDir(defaultPath)) {
taosRenameFile(defaultPath, tmp);
}
taosMkDir(defaultPath); taosMkDir(defaultPath);
code = copyFiles(chkpPath, defaultPath); code = copyFiles(chkpPath, defaultPath);
if (code != 0) { if (code != 0) {
@ -370,6 +372,8 @@ int32_t rebuildFromLocalChkp(char* key, char* chkpPath, int64_t chkpId, char* de
if (code != 0) { if (code != 0) {
if (taosIsDir(defaultPath)) taosRemoveDir(defaultPath); if (taosIsDir(defaultPath)) taosRemoveDir(defaultPath);
if (taosIsDir(tmp)) taosRenameFile(tmp, defaultPath); if (taosIsDir(tmp)) taosRenameFile(tmp, defaultPath);
} else {
taosRemoveDir(tmp);
} }
taosMemoryFree(tmp); taosMemoryFree(tmp);