mirrors/dragonfly
1
0
Fork 0
mirror of https://github.com/dragonflydb/dragonfly.git synced 2025-05-11 18:35:46 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

feat(rdb_saver): Support big value serialization for stream (#4376)

Browse source 
fixes dragonflydb#4317

Signed-off-by: Stepan Bagritsevich <stefan@dragonflydb.io>
This commit is contained in:
Stepan Bagritsevich 2024-12-26 19:15:35 +04:00 • committed by GitHub
parent fb8234ce43
commit 0065c27f27
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 17 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

30
src/server/rdb_save.cc
View file

@ -534,7 +534,9 @@ error_code RdbSerializer::SaveStreamObject(const PrimeValue& pv) {
stream* s = (stream*)pv.RObjPtr(); stream* s = (stream*)pv.RObjPtr();
rax* rax = s->rax_tree; rax* rax = s->rax_tree;
RETURN_ON_ERR(SaveLen(raxSize(rax))); const size_t rax_size = raxSize(rax);
RETURN_ON_ERR(SaveLen(rax_size));
/* Serialize all the listpacks inside the radix tree as they are, /* Serialize all the listpacks inside the radix tree as they are,
* when loading back, we'll use the first entry of each listpack * when loading back, we'll use the first entry of each listpack
@ -542,22 +544,22 @@ error_code RdbSerializer::SaveStreamObject(const PrimeValue& pv) {
raxIterator ri; raxIterator ri;
raxStart(&ri, rax); raxStart(&ri, rax);
raxSeek(&ri, "^", NULL, 0); raxSeek(&ri, "^", NULL, 0);
while (raxNext(&ri)) {
auto stop_listpacks_rax = absl::MakeCleanup([&] { raxStop(&ri); });
for (size_t i = 0; raxNext(&ri); i++) {
uint8_t* lp = (uint8_t*)ri.data; uint8_t* lp = (uint8_t*)ri.data;
size_t lp_bytes = lpBytes(lp); size_t lp_bytes = lpBytes(lp);
error_code ec = SaveString((uint8_t*)ri.key, ri.key_len);
if (ec) { RETURN_ON_ERR(SaveString((uint8_t*)ri.key, ri.key_len));
raxStop(&ri); RETURN_ON_ERR(SaveString(lp, lp_bytes));
return ec;
const FlushState flush_state =
(i + 1 < rax_size) ? FlushState::kFlushMidEntry : FlushState::kFlushEndEntry;
FlushIfNeeded(flush_state);
} }
ec = SaveString(lp, lp_bytes); std::move(stop_listpacks_rax).Invoke();
if (ec) {
raxStop(&ri);
return ec;
}
}
raxStop(&ri);
/* Save the number of elements inside the stream. We cannot obtain /* Save the number of elements inside the stream. We cannot obtain
* this easily later, since our macro nodes should be checked for * this easily later, since our macro nodes should be checked for
@ -597,7 +599,7 @@ error_code RdbSerializer::SaveStreamObject(const PrimeValue& pv) {
raxStart(&ri, s->cgroups); raxStart(&ri, s->cgroups);
raxSeek(&ri, "^", NULL, 0); raxSeek(&ri, "^", NULL, 0);
auto cleanup = absl::MakeCleanup([&] { raxStop(&ri); }); auto stop_cgroups_rax = absl::MakeCleanup([&] { raxStop(&ri); });
while (raxNext(&ri)) { while (raxNext(&ri)) {
streamCG* cg = (streamCG*)ri.data; streamCG* cg = (streamCG*)ri.data;

1
tests/dragonfly/snapshot_test.py
View file

@ -574,6 +574,7 @@ async def test_tiered_entries_throttle(async_client: aioredis.Redis):
("SET"), ("SET"),
("ZSET"), ("ZSET"),
("LIST"), ("LIST"),
("STREAM"),
], ],
) )
@pytest.mark.slow @pytest.mark.slow