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-07-17 06:47:23 +00:00
parent f0f2215f49
commit 394cc771e5
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1323,9 +1323,9 @@ bool streamStateIterSeekAndValid(rocksdb_iterator_t* iter, char* buf, size_t len
} }
return true; return true;
} }
rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* pChkptFileName, rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfKeyName,
rocksdb_snapshot_t** snapshot, rocksdb_readoptions_t** readOpt) { rocksdb_snapshot_t** snapshot, rocksdb_readoptions_t** readOpt) {
int idx = streamStateGetCfIdx(pState, pChkptFileName); int idx = streamStateGetCfIdx(pState, cfKeyName);
SBackendCfWrapper* wrapper = pState->pTdbState->pBackendCfWrapper; SBackendCfWrapper* wrapper = pState->pTdbState->pBackendCfWrapper;
if (snapshot != NULL) { if (snapshot != NULL) {
@ -2387,13 +2387,13 @@ int32_t streamStateGetBatchSize(void* pBatch) {
void streamStateClearBatch(void* pBatch) { rocksdb_writebatch_clear((rocksdb_writebatch_t*)pBatch); } void streamStateClearBatch(void* pBatch) { rocksdb_writebatch_clear((rocksdb_writebatch_t*)pBatch); }
void streamStateDestroyBatch(void* pBatch) { rocksdb_writebatch_destroy((rocksdb_writebatch_t*)pBatch); } void streamStateDestroyBatch(void* pBatch) { rocksdb_writebatch_destroy((rocksdb_writebatch_t*)pBatch); }
int32_t streamStatePutBatch(SStreamState* pState, const char* pChkptFileName, rocksdb_writebatch_t* pBatch, void* key, int32_t streamStatePutBatch(SStreamState* pState, const char* cfKeyName, rocksdb_writebatch_t* pBatch, void* key,
void* val, int32_t vlen, int64_t ttl) { void* val, int32_t vlen, int64_t ttl) {
SBackendCfWrapper* wrapper = pState->pTdbState->pBackendCfWrapper; SBackendCfWrapper* wrapper = pState->pTdbState->pBackendCfWrapper;
int i = streamStateGetCfIdx(pState, pChkptFileName); int i = streamStateGetCfIdx(pState, cfKeyName);
if (i < 0) { if (i < 0) {
qError("streamState failed to put to cf name:%s", pChkptFileName); qError("streamState failed to put to cf name:%s", cfKeyName);
return -1; return -1;
} }