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

add checkpoint

This commit is contained in:
yihaoDeng 2023-06-20 11:44:51 +08:00
parent 5ccfae87d9
commit f1fcc1d2f1
1 changed files with 69 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
source/libs/stream/src/streamSnapshot.c
View File

@ -17,9 +17,17 @@
#include "rocksdb/c.h" #include "rocksdb/c.h"
#include "tcommon.h" #include "tcommon.h"
typedef struct SBackendFile {
SArray* pSst;
char* pCurrent;
char* pMainfest;
char* pOptions;
char* pCheckpointMeta;
} SBanckendFile;
struct SStreamSnapHandle { struct SStreamSnapHandle {
void* handle; void* handle;
SArray* fileList; SArray* fileList;
SBanckendFile* pBackendFile;
}; };
struct SStreamSnapReader { struct SStreamSnapReader {
@ -35,15 +43,70 @@ struct StreamSnapWriter {
int64_t ever; int64_t ever;
}; };
void streamSnapHandleInit(SStreamSnapHandle* handle) { const char* ROCKSDB_OPTIONS = "OPTIONS";
const char* ROCKSDB_MAINFEST = "MANIFEST";
const char* ROCKSDB_SST = "sst";
const char* ROCKSDB_CURRENT = "CURRENT";
const char* ROCKSDB_CHECKPOINT_META = "CHECKPOINT";
int32_t streamSnapHandleInit(SStreamSnapHandle* handle, char* path) {
// impl later // impl later
int32_t code = 0;
handle->fileList = taosArrayInit(32, sizeof(void*)); handle->fileList = taosArrayInit(32, sizeof(void*));
return;
TdDirPtr pDir = taosOpenDir(path);
if (NULL == pDir) {
goto _err;
}
SBanckendFile* pFile = taosMemoryCalloc(1, sizeof(SBanckendFile));
pFile->pSst = taosArrayInit(16, sizeof(void*));
TdDirEntryPtr pDirEntry;
while ((pDirEntry = taosReadDir(pDir)) != NULL) {
char* name = taosGetDirEntryName(pDirEntry);
if (strlen(name) >= strlen(ROCKSDB_CURRENT) && 0 == strncmp(name, ROCKSDB_CURRENT, strlen(ROCKSDB_CURRENT))) {
pFile->pCurrent = taosStrdup(name);
continue;
}
if (strlen(name) >= strlen(ROCKSDB_MAINFEST) && 0 == strncmp(name, ROCKSDB_MAINFEST, strlen(ROCKSDB_MAINFEST))) {
pFile->pMainfest = taosStrdup(name);
continue;
}
if (strlen(name) >= strlen(ROCKSDB_OPTIONS) && 0 == strncmp(name, ROCKSDB_OPTIONS, strlen(ROCKSDB_OPTIONS))) {
pFile->pMainfest = taosStrdup(name);
continue;
}
if (strlen(name) >= strlen(ROCKSDB_CHECKPOINT_META) &&
0 == strncmp(name, ROCKSDB_CHECKPOINT_META, strlen(ROCKSDB_CHECKPOINT_META))) {
pFile->pCheckpointMeta = taosStrdup(name);
continue;
}
if (strlen(name) >= strlen(ROCKSDB_SST) &&
0 == strncmp(name - strlen(ROCKSDB_SST), ROCKSDB_SST, strlen(ROCKSDB_SST))) {
char* sst = taosStrdup(name);
taosArrayPush(pFile->pSst, &sst);
}
}
taosCloseDir(&pDir);
handle->pBackendFile = pFile;
_err:
code = -1;
return code;
} }
void streamSnapHandleDestroy(SStreamSnapHandle* handle) { void streamSnapHandleDestroy(SStreamSnapHandle* handle) {
for (int i = 0; handle && i < taosArrayGetSize(handle->fileList); i++) { SBanckendFile* pFile = handle->pBackendFile;
taosMemoryFree(taosArrayGetP(handle->fileList, i)); taosMemoryFree(pFile->pCheckpointMeta);
taosMemoryFree(pFile->pCurrent);
taosMemoryFree(pFile->pMainfest);
taosMemoryFree(pFile->pOptions);
for (int i = 0; i < taosArrayGetSize(pFile->pSst); i++) {
char* sst = taosArrayGetP(pFile->pSst, i);
taosMemoryFree(sst);
} }
taosArrayDestroy(pFile->pSst);
taosMemoryFree(pFile);
return; return;
} }