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

fix recover error

This commit is contained in:
yihaoDeng 2023-07-13 16:06:23 +08:00
parent b7403cd171
commit 18f9882834
1 changed files with 15 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1491,9 +1491,13 @@ int32_t streamStateGetKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey, cons
return -1; return -1;
} }
size_t vlen = 0; if (pVLen != NULL) {
if (pVal != NULL) *pVal = (char*)rocksdb_iter_value(pCur->iter, &vlen); size_t vlen = 0;
if (pVLen != NULL) *pVLen = vlen;
const char* valStr = rocksdb_iter_value(pCur->iter, &vlen);
*pVLen = decodeValueFunc((void*)valStr, vlen, NULL, (char**)&pVal);
}
*pKey = pKtmp->key; *pKey = pKtmp->key;
return 0; return 0;
} }
@ -1600,18 +1604,14 @@ SStreamStateCur* streamStateGetCur_rocksdb(SStreamState* pState, const SWinKey*
rocksdb_iter_seek(pCur->iter, buf, len); rocksdb_iter_seek(pCur->iter, buf, len);
if (rocksdb_iter_valid(pCur->iter) && !iterValueIsStale(pCur->iter)) { if (rocksdb_iter_valid(pCur->iter) && !iterValueIsStale(pCur->iter)) {
size_t vlen; SStateKey curKey;
char* val = (char*)rocksdb_iter_value(pCur->iter, &vlen); size_t kLen = 0;
if (!streamStateValueIsStale(val)) { char* keyStr = (char*)rocksdb_iter_key(pCur->iter, &kLen);
SStateKey curKey; stateKeyDecode((void*)&curKey, keyStr);
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) { if (stateKeyCmpr(&sKey, sizeof(sKey), &curKey, sizeof(curKey)) == 0) {
pCur->number = pState->number; pCur->number = pState->number;
return pCur; return pCur;
}
} }
} }
streamStateFreeCur(pCur); streamStateFreeCur(pCur);
@ -1900,8 +1900,7 @@ int32_t streamStateFillGetKVByCur_rocksdb(SStreamStateCur* pCur, SWinKey* pKey,
winKeyDecode(&winKey, keyStr); winKeyDecode(&winKey, keyStr);
const char* valStr = rocksdb_iter_value(pCur->iter, &vlen); const char* valStr = rocksdb_iter_value(pCur->iter, &vlen);
// char* dst = NULL; int32_t len = decodeValueFunc((void*)valStr, vlen, NULL, (char**)pVal);
int32_t len = decodeValueFunc((void*)valStr, vlen, NULL, (char**)pVal);
if (len < 0) { if (len < 0) {
return -1; return -1;
} }