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

fix recover

This commit is contained in:
yihaoDeng 2023-04-14 14:05:59 +00:00
parent efa7b48c1b
commit 80f348de1d
2 changed files with 19 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
source/libs/stream/src/streamStateRocksdb.c
View File

@ -489,7 +489,8 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
int i = streamGetInit(funcname); \ int i = streamGetInit(funcname); \
if (i < 0) { \ if (i < 0) { \
qWarn("streamState failed to get cf name: %s", funcname); \ qWarn("streamState failed to get cf name: %s", funcname); \
return -1; \ code = -1; \
break; \
} \ } \
char toString[128] = {0}; \ char toString[128] = {0}; \
if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString); \ if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString); \
@ -515,7 +516,8 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
int i = streamGetInit(funcname); \ int i = streamGetInit(funcname); \
if (i < 0) { \ if (i < 0) { \
qWarn("streamState failed to get cf name: %s", funcname); \ qWarn("streamState failed to get cf name: %s", funcname); \
return -1; \ code = -1; \
break; \
} \ } \
char toString[128] = {0}; \ char toString[128] = {0}; \
if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString); \ if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString); \
@ -550,7 +552,8 @@ rocksdb_iterator_t* streamStateIterCreate(SStreamState* pState, const char* cfNa
int i = streamGetInit(funcname); \ int i = streamGetInit(funcname); \
if (i < 0) { \ if (i < 0) { \
qWarn("streamState failed to get cf name: %s", funcname); \ qWarn("streamState failed to get cf name: %s", funcname); \
return -1; \ code = -1; \
break; \
} \ } \
char toString[128] = {0}; \ char toString[128] = {0}; \
if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString); \ if (qDebugFlag & DEBUG_TRACE) ginitDict[i].toStrFunc((void*)key, toString); \
@ -970,36 +973,20 @@ int32_t streamStateAddIfNotExist_rocksdb(SStreamState* pState, const SWinKey* ke
} }
SStreamStateCur* streamStateSeekToLast_rocksdb(SStreamState* pState, const SWinKey* key) { SStreamStateCur* streamStateSeekToLast_rocksdb(SStreamState* pState, const SWinKey* key) {
qDebug("streamStateGetCur_rocksdb"); qDebug("streamStateGetCur_rocksdb");
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur)); int32_t code = 0;
const SStateKey maxStateKey = {.key = {.groupId = UINT64_MAX, .ts = INT64_MAX}, .opNum = INT64_MAX};
STREAM_STATE_PUT_ROCKSDB(pState, "state", &maxStateKey, "", 0);
char buf[128] = {0};
int32_t klen = stateKeyEncode((void*)&maxStateKey, buf);
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) return NULL; if (pCur == NULL) return NULL;
pCur->db = pState->pTdbState->rocksdb; pCur->db = pState->pTdbState->rocksdb;
pCur->iter = streamStateIterCreate(pState, "state", &pCur->snapshot, &pCur->readOpt); pCur->iter = streamStateIterCreate(pState, "state", &pCur->snapshot, &pCur->readOpt);
rocksdb_iter_seek(pCur->iter, buf, (size_t)klen);
rocksdb_iter_prev(pCur->iter);
SStateKey sKey = {.key = *key, .opNum = pState->number}; STREAM_STATE_DEL_ROCKSDB(pState, "state", &maxStateKey);
char buf[128] = {0};
int len = stateKeyEncode((void*)&sKey, buf);
rocksdb_iter_seek(pCur->iter, buf, len);
if (!rocksdb_iter_valid(pCur->iter)) {
rocksdb_iter_seek_to_last(pCur->iter);
} else {
rocksdb_iter_seek_to_last(pCur->iter);
}
return pCur;
// rocksdb_iter_seek(pCur->iter, buf, len);
// if (rocksdb_iter_valid(pCur->iter)) {
// SStateKey curKey;
// size_t kLen = 0;
// char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &kLen);
// stateKeyDecode((void*)&curKey, keyStr);
// if (stateKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) == 0) {
// pCur->number = pState->number;
// return pCur;
// }
// }
// streamStateFreeCur(pCur);
return pCur; return pCur;
} }
SStreamStateCur* streamStateGetCur_rocksdb(SStreamState* pState, const SWinKey* key) { SStreamStateCur* streamStateGetCur_rocksdb(SStreamState* pState, const SWinKey* key) {

15
source/libs/stream/src/tstreamFileState.c
View File

@ -327,7 +327,7 @@ int32_t flushSnapshot(SStreamFileState* pFileState, SStreamSnapshot* pSnapshot,
SListIter iter = {0}; SListIter iter = {0};
tdListInitIter(pSnapshot, &iter, TD_LIST_FORWARD); tdListInitIter(pSnapshot, &iter, TD_LIST_FORWARD);
const int32_t BATCH_LIMIT = 128; const int32_t BATCH_LIMIT = 256;
SListNode* pNode = NULL; SListNode* pNode = NULL;
void* batch = streamStateCreateBatch(); void* batch = streamStateCreateBatch();
@ -414,6 +414,7 @@ int32_t deleteExpiredCheckPoint(SStreamFileState* pFileState, TSKEY mark) {
sscanf(val, "%" PRId64 "", &ts); sscanf(val, "%" PRId64 "", &ts);
taosMemoryFree(val); taosMemoryFree(val);
if (ts < mark) { if (ts < mark) {
// statekey winkey.ts < mark
forceRemoveCheckpoint(pFileState, i); forceRemoveCheckpoint(pFileState, i);
break; break;
} else { } else {
@ -428,16 +429,8 @@ int32_t recoverSnapshot(SStreamFileState* pFileState) {
void* pStVal = NULL; void* pStVal = NULL;
int32_t len = 0; int32_t len = 0;
SWinKey key = {.groupId = 0, .ts = 0}; SWinKey key = {.groupId = 0, .ts = 0};
// SStreamStateCur* pCur = streamStateGetCur_rocksdb(pFileState->pFileStore, &key); SStreamStateCur* pCur = streamStateSeekToLast_rocksdb(pFileState->pFileStore, &key);
// if (!pCur) {
// return TSDB_CODE_FAILED;
// }
// code = streamStateSeekLast(pFileState->pFileStore, pCur);
// if (code != TSDB_CODE_SUCCESS) {
// return code;
// }
SStreamStateCur* pCur = streamStateSeekToLast_rocksdb(pState, &key);
if (pCur == NULL) { if (pCur == NULL) {
return -1; return -1;
} }