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-25 14:35:20 +08:00
parent 2f656f3854
commit e36de5edf7
2 changed files with 56 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
include/libs/stream/streamSnapshot.h
View File

@ -17,7 +17,7 @@
#include "tcommon.h"
typedef struct SStreamSnapReader SStreamSnapReader;
typedef struct StreamSnapWriter StreamSnapWriter;
typedef struct SStreamSnapWriter SStreamSnapWriter;
typedef struct SStreamSnapHandle SStreamSnapHandle;
typedef struct SStreamSnapBlockHdr SStreamSnapBlockHdr;
@ -27,8 +27,8 @@ int32_t streamSnapReaderClose(SStreamSnapReader* pReader);
int32_t streamSnapRead(SStreamSnapReader* pReader, uint8_t** ppData, int64_t* size);
// SMetaSnapWriter ========================================
int32_t streamSnapWriterOpen(void* pMeta, int64_t sver, int64_t ever, StreamSnapWriter** ppWriter);
int32_t streamSnapWrite(StreamSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
int32_t streamSnapWriterClose(StreamSnapWriter** ppWriter, int8_t rollback);
int32_t streamSnapWriterOpen(void* pMeta, int64_t sver, int64_t ever, SStreamSnapWriter** ppWriter);
int32_t streamSnapWrite(SStreamSnapWriter* pWriter, uint8_t* pData, uint32_t nData);
int32_t streamSnapWriterClose(SStreamSnapWriter** ppWriter, int8_t rollback);
#endif

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

@ -62,10 +62,11 @@ struct SStreamSnapReader {
int64_t ever;
SStreamSnapHandle handle;
};
struct StreamSnapWriter {
void* pMeta;
int64_t sver;
int64_t ever;
struct SStreamSnapWriter {
void* pMeta;
int64_t sver;
int64_t ever;
SStreamSnapHandle handle;
};
const char* ROCKSDB_OPTIONS = "OPTIONS";
const char* ROCKSDB_MAINFEST = "MANIFEST";
@ -241,12 +242,53 @@ int32_t streamSnapRead(SStreamSnapReader* pReader, uint8_t** ppData, int64_t* si
return 0;
}
// SMetaSnapWriter ========================================
int32_t streamSnapWriterOpen(void* pMeta, int64_t sver, int64_t ever, StreamSnapWriter** ppWriter) {
int32_t streamSnapWriterOpen(void* pMeta, int64_t sver, int64_t ever, SStreamSnapWriter** ppWriter) {
// impl later
SStreamSnapWriter* pWriter = taosMemoryCalloc(1, sizeof(SStreamSnapWriter));
if (pWriter == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
const char* path = NULL;
// if (streamSnapHandleInit(&pWriter->handle, (char*)path) < 0) {
// return -1;
// }
SBanckendFile* pFile = taosMemoryCalloc(1, sizeof(SBanckendFile));
pFile->path = taosStrdup(path);
pFile->pSst = taosArrayInit(16, sizeof(void*));
SArray* list = taosArrayInit(64, sizeof(SBackendFileItem));
// SBackendFileItem item;
// // current
// item.name = (char*)ROCKSDB_CURRENT;
// item.type = ROCKSDB_CURRENT_TYPE;
// taosArrayPush(list, &item);
// // mainfest
// item.name = pFile->pMainfest;
// item.type = ROCKSDB_MAINFEST_TYPE;
// taosArrayPush(list, &item);
// // options
// item.name = pFile->pOptions;
// item.type = ROCKSDB_OPTIONS_TYPE;
// taosArrayPush(list, &item);
// // sst
// for (int i = 0; i < taosArrayGetSize(pFile->pSst); i++) {
// char* sst = taosArrayGetP(pFile->pSst, i);
// item.name = sst;
// item.type = ROCKSDB_SST_TYPE;
// taosArrayPush(list, &item);
// }
// // meta
// item.name = pFile->pCheckpointMeta;
// item.type = ROCKSDB_CHECKPOINT_META_TYPE;
// taosArrayPush(list, &item);
*ppWriter = pWriter;
return 0;
}
int32_t streamSnapWrite(SStreamSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
// impl later
return 0;
}
int32_t streamSnapWrite(StreamSnapWriter* pWriter, uint8_t* pData, uint32_t nData) {
// impl later
return 0;
}
int32_t streamSnapWriterClose(StreamSnapWriter** ppWriter, int8_t rollback) { return 0; }
int32_t streamSnapWriterClose(SStreamSnapWriter** ppWriter, int8_t rollback) { return 0; }