This is an automated email from the ASF dual-hosted git repository.
git-hulk pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 15e047ee1 perf(stream): trim with a single DeleteRange instead of
per-entry Delete (#3592)
15e047ee1 is described below
commit 15e047ee129b4363204382ebb04a7214bdee417d
Author: Alexander Bocharoff <[email protected]>
AuthorDate: Tue Aug 18 03:58:28 2026 -0700
perf(stream): trim with a single DeleteRange instead of per-entry Delete
(#3592)
### Problem
`Stream::trim()` — the shared path behind `XTRIM` and `XADD` with
`MAXLEN`/`MINID` — removes trimmed entries one at a time: a per-entry
`WriteBatch::Delete()` in a loop, committed as a single atomic batch.
This makes trimming **O(entries removed)**:
- Removing a large number of entries builds a correspondingly large
write batch of point tombstones and commits it in one atomic write.
- Those point tombstones are then scanned one-by-one by every reader,
and re-scanned by each subsequent trim's `Seek`.
For a stream that is trimmed continuously (each trim removes many
entries), this is expensive both on the write and on subsequent
reads/trims.
### Change
Replace the per-entry `Delete()` loop with a single
`WriteBatch::DeleteRange()` over the trimmed span — the same approach
already used for bulk deletion in `DestroyGroup()`.
- The entries are still scanned to count them, so `metadata` (`size`,
`first_entry_id`, `max_deleted_entry_id`) is maintained exactly as
before and `XLEN` stays exact. Only the deletion mechanism changes: the
write batch stays O(1) (one range tombstone) instead of O(entries
removed).
- Group, consumer and PEL subkeys are encoded with a `UINT64_MAX` prefix
and sort after every entry key, so they always stay outside the deleted
range. The exclusive upper bound is the first surviving entry for a
partial trim, or the next-version prefix when the whole stream is
trimmed.
### Tests
Adds cases for the behaviour specific to the range delete, alongside the
existing trim suite:
- consumer group + PEL survival across a full-entry trim,
- entries added after a trim-to-empty remaining visible (range-tombstone
sequence-number semantics),
- repeated `MINID` trims interleaved with adds keeping exact survivors
and first/max-deleted bookkeeping.
---------
Co-authored-by: hulk <[email protected]>
---
src/cluster/batch_sender.cc | 10 ++
src/cluster/batch_sender.h | 1 +
src/cluster/slot_migrate.cc | 12 +-
src/storage/batch_extractor.cc | 39 ++++-
src/storage/iterator.cc | 11 ++
src/types/redis_stream.cc | 17 +-
tests/cppunit/types/stream_test.cc | 171 +++++++++++++++++++++
.../integration/slotmigrate/slotmigrate_test.go | 38 +++++
tests/gocase/unit/server/poll_updates_test.go | 27 ++++
tests/gocase/unit/type/stream/stream_test.go | 79 ++++++++++
10 files changed, 394 insertions(+), 11 deletions(-)
diff --git a/src/cluster/batch_sender.cc b/src/cluster/batch_sender.cc
index e92221ee6..e0617600d 100644
--- a/src/cluster/batch_sender.cc
+++ b/src/cluster/batch_sender.cc
@@ -52,6 +52,16 @@ Status BatchSender::Delete(rocksdb::ColumnFamilyHandle *cf,
const rocksdb::Slice
entries_num_++;
return Status::OK();
}
+Status BatchSender::DeleteRange(rocksdb::ColumnFamilyHandle *cf, const
rocksdb::Slice &begin,
+ const rocksdb::Slice &end) {
+ auto s = write_batch_.DeleteRange(cf, begin, end);
+ if (!s.ok()) {
+ return {Status::NotOK, fmt::format("failed to delete range from migration
batch, {}", s.ToString())};
+ }
+ pending_entries_++;
+ entries_num_++;
+ return Status::OK();
+}
Status BatchSender::PutLogData(const rocksdb::Slice &blob) {
auto s = write_batch_.PutLogData(blob);
diff --git a/src/cluster/batch_sender.h b/src/cluster/batch_sender.h
index 07c4e9fdb..39d3e6bce 100644
--- a/src/cluster/batch_sender.h
+++ b/src/cluster/batch_sender.h
@@ -39,6 +39,7 @@ class BatchSender {
Status Put(rocksdb::ColumnFamilyHandle *cf, const rocksdb::Slice &key, const
rocksdb::Slice &value);
Status Delete(rocksdb::ColumnFamilyHandle *cf, const rocksdb::Slice &key);
+ Status DeleteRange(rocksdb::ColumnFamilyHandle *cf, const rocksdb::Slice
&begin, const rocksdb::Slice &end);
Status PutLogData(const rocksdb::Slice &blob);
void SetPrefixLogData(const std::string &prefix_logdata);
Status Send();
diff --git a/src/cluster/slot_migrate.cc b/src/cluster/slot_migrate.cc
index 02e0534c2..28907306b 100644
--- a/src/cluster/slot_migrate.cc
+++ b/src/cluster/slot_migrate.cc
@@ -731,8 +731,16 @@ Status
SlotMigrator::migrateIncrementalDataByRawKV(uint64_t end_seq, BatchSender
break;
}
case engine::WALItem::Type::kTypeDeleteRange: {
- // Do nothing in DeleteRange due to it might cross multiple slots.
It's only used in
- // FLUSHDB/FLUSHALL commands for now and maybe we can disable them
while migrating.
+ // The WAL extractor only surfaces range deletes confined to a single
in-range slot
+ // (e.g. a stream trim), so forwarding is safe; multi-slot ranges
(FLUSHDB/FLUSHALL) are
+ // dropped before they reach here.
+ if (item.column_family_id > kMaxColumnFamilyID) {
+ INFO("[migrate] Invalid delete-range column family id: {}",
item.column_family_id);
+ continue;
+ }
+
GET_OR_RET(batch_sender->DeleteRange(storage_->GetCFHandle(static_cast<ColumnFamilyID>(item.column_family_id)),
+ item.key, item.value));
+ break;
}
default:
break;
diff --git a/src/storage/batch_extractor.cc b/src/storage/batch_extractor.cc
index 8dfd4258e..e545b4258 100644
--- a/src/storage/batch_extractor.cc
+++ b/src/storage/batch_extractor.cc
@@ -408,10 +408,41 @@ rocksdb::Status WriteBatchExtractor::DeleteCF(uint32_t
column_family_id, const S
return rocksdb::Status::OK();
}
-rocksdb::Status WriteBatchExtractor::DeleteRangeCF([[maybe_unused]] uint32_t
column_family_id,
- [[maybe_unused]] const
Slice &begin_key,
- [[maybe_unused]] const
Slice &end_key) {
- // Do nothing with DeleteRange operations
+rocksdb::Status WriteBatchExtractor::DeleteRangeCF(uint32_t column_family_id,
const Slice &begin_key,
+ const Slice &end_key) {
+ // The only range delete over user entries is a stream trim
(redis_stream.cc). Reconstruct the
+ // equivalent command so RESP update consumers stay in sync, mirroring how a
per-entry stream
+ // delete becomes XDEL. Ranges on other column families (FLUSHDB/FLUSHALL)
are left alone.
+ if (column_family_id != static_cast<uint32_t>(ColumnFamilyID::Stream)) {
+ return rocksdb::Status::OK();
+ }
+
+ InternalKey begin_ikey(begin_key, is_slot_id_encoded_);
+ Slice begin_sub = begin_ikey.GetSubKey();
+ // Group/consumer/PEL subkeys are UINT64_MAX-prefixed and sort above every
entry; a trim range
+ // begins at a real entry id, so a UINT64_MAX prefix marks an XGROUP DESTROY
range to skip.
+ uint64_t begin_marker = 0;
+ GetFixed64(&begin_sub, &begin_marker);
+ if (begin_marker == UINT64_MAX) {
+ return rocksdb::Status::OK();
+ }
+
+ std::string user_key = begin_ikey.GetKey().ToString();
+ InternalKey end_ikey(end_key, is_slot_id_encoded_);
+ std::vector<std::string> command_args;
+ if (end_ikey.GetVersion() != begin_ikey.GetVersion()) {
+ // Full trim: the end bound is the next-version prefix, so every entry was
removed.
+ command_args = {"XTRIM", user_key, "MAXLEN", "0"};
+ } else {
+ // Partial trim: the end bound is the first surviving entry, so trim
everything below it.
+ Slice end_sub = end_ikey.GetSubKey();
+ redis::StreamEntryID first_survivor;
+ GetFixed64(&end_sub, &first_survivor.ms);
+ GetFixed64(&end_sub, &first_survivor.seq);
+ command_args = {"XTRIM", user_key, "MINID", first_survivor.ToString()};
+ }
+
+ resp_commands_[""].emplace_back(redis::ArrayOfBulkStrings(command_args));
return rocksdb::Status::OK();
}
diff --git a/src/storage/iterator.cc b/src/storage/iterator.cc
index e360ffe0c..826a25a0c 100644
--- a/src/storage/iterator.cc
+++ b/src/storage/iterator.cc
@@ -188,6 +188,17 @@ rocksdb::Status WALBatchExtractor::DeleteCF(uint32_t
column_family_id, const roc
rocksdb::Status WALBatchExtractor::DeleteRangeCF(uint32_t column_family_id,
const rocksdb::Slice &begin_key,
const rocksdb::Slice
&end_key) {
+ if (slot_range_.IsValid()) {
+ // A range delete can be migrated only when it is confined to a single
in-range slot. A stream
+ // trim (and XGROUP DESTROY) deletes one key's subkeys, so begin/end share
that key's slot;
+ // a FLUSHDB/FLUSHALL range spans the whole column family across slots and
cannot be applied
+ // slot-scoped, so drop it as before.
+ auto begin_slot_id = ExtractSlotId(begin_key);
+ auto end_slot_id = ExtractSlotId(end_key);
+ if (begin_slot_id != end_slot_id || !slot_range_.Contains(begin_slot_id)) {
+ return rocksdb::Status::OK();
+ }
+ }
items_.emplace_back(WALItem::Type::kTypeDeleteRange, column_family_id,
begin_key.ToString(), end_key.ToString());
return rocksdb::Status::OK();
}
diff --git a/src/types/redis_stream.cc b/src/types/redis_stream.cc
index 78ec5cc02..66e05bbda 100644
--- a/src/types/redis_stream.cc
+++ b/src/types/redis_stream.cc
@@ -1679,11 +1679,6 @@ rocksdb::Status Stream::trim(engine::Context &ctx, const
std::string &ns_key, co
break;
}
- auto s = batch->Delete(stream_cf_handle_, iter->key());
- if (!s.ok()) {
- return s;
- }
-
delete_cnt += 1;
metadata->size -= 1;
last_deleted = iter->key().ToString();
@@ -1699,6 +1694,18 @@ rocksdb::Status Stream::trim(engine::Context &ctx, const
std::string &ns_key, co
}
}
+ // Remove the counted run with one range tombstone rather than the per-entry
Delete() above:
+ // the write batch stays O(1) instead of O(entries removed), and readers
skip the span via the
+ // range-del aggregator instead of stepping over (and re-scanning) N point
tombstones. The scan
+ // stopped on the first surviving entry, so its key is the exclusive upper
bound; a full trim
+ // uses the version upper bound. Group/consumer/PEL subkeys sort above
entries, so either bound
+ // keeps them outside the deleted range.
+ if (delete_cnt > 0) {
+ std::string range_end = iter->Valid() ? iter->key().ToString() :
next_version_prefix_key;
+ auto s = batch->DeleteRange(stream_cf_handle_, start_key, range_end);
+ if (!s.ok()) return s;
+ }
+
if (metadata->size == 0) {
metadata->first_entry_id.Clear();
metadata->last_entry_id.Clear();
diff --git a/tests/cppunit/types/stream_test.cc
b/tests/cppunit/types/stream_test.cc
index 96c22c543..9307095cc 100644
--- a/tests/cppunit/types/stream_test.cc
+++ b/tests/cppunit/types/stream_test.cc
@@ -2017,6 +2017,177 @@ TEST_F(RedisStreamTest,
TrimWithMinIdGreaterThanLastEntryID) {
EXPECT_EQ(length, 0);
}
+// Trimming every entry away must leave the consumer group, its delivery
cursor and its
+// pending (PEL) entries intact: the range-delete used by trim stops at the
first
+// group/consumer/PEL subkey, which sorts above every entry key.
+TEST_F(RedisStreamTest, TrimAllEntriesPreservesConsumerGroupAndPel) {
+ redis::StreamAddOptions add_options;
+ redis::StreamEntryID id1, id2, id3;
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy("1-0");
+ auto s = stream_->Add(*ctx_, name_, add_options, {"k1", "v1"}, &id1);
+ EXPECT_TRUE(s.ok());
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy("2-0");
+ s = stream_->Add(*ctx_, name_, add_options, {"k2", "v2"}, &id2);
+ EXPECT_TRUE(s.ok());
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy("3-0");
+ s = stream_->Add(*ctx_, name_, add_options, {"k3", "v3"}, &id3);
+ EXPECT_TRUE(s.ok());
+
+ std::string group_name = "g1";
+ std::string consumer_name = "c1";
+ redis::StreamXGroupCreateOptions create_options = {false, 0, "0-0"};
+ s = stream_->CreateGroup(*ctx_, name_, create_options, group_name);
+ EXPECT_TRUE(s.ok());
+
+ // Deliver all three entries into the group's PEL.
+ redis::StreamRangeOptions range_options;
+ range_options.start = redis::StreamEntryID::Minimum();
+ range_options.end = redis::StreamEntryID::Maximum();
+ range_options.count = 10;
+ range_options.with_count = true;
+ range_options.exclude_start = true;
+ std::vector<redis::StreamEntry> delivered;
+ s = stream_->RangeWithPending(*ctx_, name_, range_options, &delivered,
group_name, consumer_name, false, true);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(delivered.size(), 3);
+
+ redis::StreamTrimOptions options;
+ options.strategy = redis::StreamTrimStrategy::MinID;
+ options.min_id = redis::StreamEntryID{4, 0};
+ uint64_t trimmed = 0;
+ s = stream_->Trim(*ctx_, name_, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 3);
+
+ uint64_t length = 0;
+ s = stream_->Len(*ctx_, name_, redis::StreamLenOptions{}, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 0);
+
+ std::vector<std::pair<std::string, redis::StreamConsumerGroupMetadata>>
group_metadata;
+ s = stream_->GetGroupInfo(*ctx_, name_, group_metadata);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(group_metadata.size(), 1);
+ EXPECT_EQ(group_metadata[0].first, group_name);
+ EXPECT_EQ(group_metadata[0].second.pending_number, 3);
+ EXPECT_EQ(group_metadata[0].second.last_delivered_id.ToString(),
id3.ToString());
+
+ // pending_number above is only the count cached in group metadata; read the
PEL subkeys
+ // themselves back to prove the range tombstone stopped short of them and
left every
+ // delivered id in place.
+ redis::StreamPendingOptions pending_options;
+ pending_options.stream_name = name_;
+ pending_options.group_name = group_name;
+ pending_options.start_id = redis::StreamEntryID::Minimum();
+ pending_options.end_id = redis::StreamEntryID::Maximum();
+ pending_options.with_count = true; // extended form: populate ext_results
from the PEL subkeys
+ pending_options.count = 10;
+ redis::StreamGetPendingEntryResult pending_infos;
+ std::vector<redis::StreamNACK> ext_results;
+ s = stream_->GetPendingEntries(*ctx_, pending_options, pending_infos,
ext_results);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(ext_results.size(), 3);
+ EXPECT_EQ(ext_results[0].id.ToString(), id1.ToString());
+ EXPECT_EQ(ext_results[1].id.ToString(), id2.ToString());
+ EXPECT_EQ(ext_results[2].id.ToString(), id3.ToString());
+}
+
+// Entries added after a trim-to-empty carry a higher sequence number than the
range
+// tombstone, so they must stay visible even though their ids fall inside the
deleted range.
+TEST_F(RedisStreamTest, TrimToEmptyThenAddedEntriesRemainVisible) {
+ redis::StreamAddOptions add_options;
+ redis::StreamEntryID id;
+ for (const char *ts : {"1-0", "2-0", "3-0"}) {
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy(ts);
+ auto s = stream_->Add(*ctx_, name_, add_options, {"k", "v"}, &id);
+ EXPECT_TRUE(s.ok());
+ }
+
+ redis::StreamTrimOptions options;
+ options.strategy = redis::StreamTrimStrategy::MinID;
+ options.min_id = redis::StreamEntryID{100, 0};
+ uint64_t trimmed = 0;
+ auto s = stream_->Trim(*ctx_, name_, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 3);
+
+ redis::StreamEntryID id4, id5;
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy("10-0");
+ s = stream_->Add(*ctx_, name_, add_options, {"k4", "v4"}, &id4);
+ EXPECT_TRUE(s.ok());
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy("11-0");
+ s = stream_->Add(*ctx_, name_, add_options, {"k5", "v5"}, &id5);
+ EXPECT_TRUE(s.ok());
+
+ uint64_t length = 0;
+ s = stream_->Len(*ctx_, name_, redis::StreamLenOptions{}, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 2);
+
+ redis::StreamRangeOptions range_options;
+ range_options.start = redis::StreamEntryID::Minimum();
+ range_options.end = redis::StreamEntryID::Maximum();
+ std::vector<redis::StreamEntry> entries;
+ s = stream_->Range(*ctx_, name_, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, id4.ToString());
+ CheckStreamEntryValues(entries[0].values, {"k4", "v4"});
+ EXPECT_EQ(entries[1].key, id5.ToString());
+ CheckStreamEntryValues(entries[1].values, {"k5", "v5"});
+}
+
+// Repeated forward MINID trims interleaved with adds must leave exactly the
surviving
+// entries and correct first/max-deleted bookkeeping.
+TEST_F(RedisStreamTest, RepeatedMinIdTrimKeepsExactSurvivors) {
+ redis::StreamAddOptions add_options;
+ redis::StreamEntryID id;
+ for (const char *ts : {"1-0", "2-0", "3-0", "4-0", "5-0"}) {
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy(ts);
+ auto s = stream_->Add(*ctx_, name_, add_options, {"k", "v"}, &id);
+ EXPECT_TRUE(s.ok());
+ }
+
+ redis::StreamTrimOptions options;
+ options.strategy = redis::StreamTrimStrategy::MinID;
+ options.min_id = redis::StreamEntryID{3, 0};
+ uint64_t trimmed = 0;
+ auto s = stream_->Trim(*ctx_, name_, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 2);
+
+ add_options.next_id_strategy = *ParseNextStreamEntryIDStrategy("6-0");
+ s = stream_->Add(*ctx_, name_, add_options, {"k6", "v6"}, &id);
+ EXPECT_TRUE(s.ok());
+
+ options.min_id = redis::StreamEntryID{5, 0};
+ trimmed = 0;
+ s = stream_->Trim(*ctx_, name_, options, &trimmed);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(trimmed, 2);
+
+ uint64_t length = 0;
+ s = stream_->Len(*ctx_, name_, redis::StreamLenOptions{}, &length);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(length, 2);
+
+ redis::StreamRangeOptions range_options;
+ range_options.start = redis::StreamEntryID::Minimum();
+ range_options.end = redis::StreamEntryID::Maximum();
+ std::vector<redis::StreamEntry> entries;
+ s = stream_->Range(*ctx_, name_, range_options, &entries);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(entries.size(), 2);
+ EXPECT_EQ(entries[0].key, (redis::StreamEntryID{5, 0}).ToString());
+ EXPECT_EQ(entries[1].key, (redis::StreamEntryID{6, 0}).ToString());
+
+ redis::StreamInfo info;
+ s = stream_->GetStreamInfo(*ctx_, name_, false, 0, &info);
+ EXPECT_TRUE(s.ok());
+ EXPECT_EQ(info.recorded_first_entry_id.ToString(), (redis::StreamEntryID{5,
0}).ToString());
+ EXPECT_EQ(info.max_deleted_entry_id.ToString(), (redis::StreamEntryID{4,
0}).ToString());
+}
+
TEST_F(RedisStreamTest, StreamInfoOnNonExistingStream) {
redis::StreamInfo info;
auto s = stream_->GetStreamInfo(*ctx_, name_, false, 0, &info);
diff --git a/tests/gocase/integration/slotmigrate/slotmigrate_test.go
b/tests/gocase/integration/slotmigrate/slotmigrate_test.go
index 6e430ce63..6c82bcf14 100644
--- a/tests/gocase/integration/slotmigrate/slotmigrate_test.go
+++ b/tests/gocase/integration/slotmigrate/slotmigrate_test.go
@@ -1118,6 +1118,44 @@ func TestSlotMigrateDataType(t *testing.T) {
require.EqualValues(t, []string{"element985", "element986",
"element987", "element988", "element989"},
rdb1.LRange(ctx, srcListName, -5, -1).Val())
})
+
+ t.Run("MIGRATE - XTRIM via parsing WAL logs", func(t *testing.T) {
+ testSlot += 1
+
+ streamName := fmt.Sprintf("stream_{%s}",
util.SlotTable[testSlot])
+ fillerListName := fmt.Sprintf("list_filler_{%s}",
util.SlotTable[testSlot])
+ require.NoError(t, rdb0.Del(ctx, streamName).Err())
+ require.NoError(t, rdb0.Del(ctx, fillerListName).Err())
+
+ // Seed the stream so its entries are part of the migration
snapshot.
+ for i := 1; i <= 10; i++ {
+ require.NoError(t, rdb0.XAdd(ctx, &redis.XAddArgs{
+ Stream: streamName, ID: fmt.Sprintf("%d-0", i),
Values: map[string]interface{}{"f": "v"},
+ }).Err())
+ }
+
+ // Slow down the migration so the XTRIM below is synced via the
WAL, where the trim is a
+ // single range tombstone. If that range delete were dropped
(as it was before), the
+ // destination would keep the trimmed entries while its
metadata claimed they were gone.
+ require.NoError(t, rdb0.ConfigSet(ctx, "migrate-batch-size-kb",
"1").Err())
+ require.NoError(t, rdb0.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
+ fillerValue := strings.Repeat("a", 512)
+ for i := 0; i < 10000; i++ {
+ require.NoError(t, rdb0.LPush(ctx, fillerListName,
fillerValue).Err())
+ }
+
+ require.Equal(t, "OK", rdb0.Do(ctx, "clusterx", "migrate",
testSlot, id1).Val())
+ requireMigrateState(t, rdb0, testSlot,
SlotMigrationStateStarted)
+
+ require.NoError(t, rdb0.XTrimMinID(ctx, streamName,
"6-0").Err())
+ waitForMigrateStateInDuration(t, rdb0, testSlot,
SlotMigrationStateSuccess, time.Minute)
+
+ require.EqualValues(t, 5, rdb1.XLen(ctx, streamName).Val())
+ items := rdb1.XRange(ctx, streamName, "-", "+").Val()
+ require.Len(t, items, 5)
+ require.EqualValues(t, "6-0", items[0].ID)
+ require.EqualValues(t, "10-0", items[4].ID)
+ })
}
func TestSlotMigrateDestNotSupportApplyBatch(t *testing.T) {
diff --git a/tests/gocase/unit/server/poll_updates_test.go
b/tests/gocase/unit/server/poll_updates_test.go
index 9a607e202..287989e20 100644
--- a/tests/gocase/unit/server/poll_updates_test.go
+++ b/tests/gocase/unit/server/poll_updates_test.go
@@ -290,6 +290,33 @@ func TestPollUpdates_WithRESPFormat(t *testing.T) {
}, pollUpdates.Updates)
})
+ t.Run("Stream trim", func(t *testing.T) {
+ // Trim removes entries with one range tombstone; POLLUPDATES
must still reconstruct an
+ // equivalent command. A partial trim becomes XTRIM MINID
<first surviving id>; a trim that
+ // empties the stream becomes XTRIM MAXLEN 0.
+ for _, id := range []string{"1-0", "2-0", "3-0"} {
+ require.NoError(t, rdb0.XAdd(ctx, &redis.XAddArgs{
+ Stream: "trimstream", ID: id, Values:
map[string]interface{}{"f": "v"},
+ }).Err())
+ }
+ require.NoError(t, rdb0.XTrimMinID(ctx, "trimstream",
"2-0").Err())
+ require.NoError(t, rdb0.XTrimMaxLen(ctx, "trimstream", 0).Err())
+ result, err := rdb0.Do(ctx, "POLLUPDATES", pollUpdates.NextSeq,
"MAX", 10, "FORMAT", "RESP").Result()
+ require.NoError(t, err)
+
+ pollUpdates = parsePollUpdatesResult(t, result.(map[any]any),
true)
+ require.Len(t, pollUpdates.Updates, 1)
+ require.EqualValues(t, []any{RESPFormat{
+ Commands: [][]string{
+ {"XADD", "trimstream", "1-0", "f", "v"},
+ {"XADD", "trimstream", "2-0", "f", "v"},
+ {"XADD", "trimstream", "3-0", "f", "v"},
+ {"XTRIM", "trimstream", "MINID", "2-0"},
+ {"XTRIM", "trimstream", "MAXLEN", "0"},
+ }},
+ }, pollUpdates.Updates)
+ })
+
t.Run("JSON type", func(t *testing.T) {
require.NoError(t, rdb0.JSONSet(ctx, "json", "$", `{"field":
"value"}`).Err())
require.NoError(t, rdb0.JSONDel(ctx, "json", "$.field").Err())
diff --git a/tests/gocase/unit/type/stream/stream_test.go
b/tests/gocase/unit/type/stream/stream_test.go
index 1be57ede2..7e3b65751 100644
--- a/tests/gocase/unit/type/stream/stream_test.go
+++ b/tests/gocase/unit/type/stream/stream_test.go
@@ -650,6 +650,85 @@ var streamTests = func(t *testing.T, configs
util.KvrocksServerConfigs) {
require.EqualValues(t, "5-0", items[2].ID)
})
+ t.Run("XTRIM removing every entry keeps the consumer group and its
PEL", func(t *testing.T) {
+ require.NoError(t, rdb.Del(ctx, "mystream").Err())
+ for i := 1; i <= 3; i++ {
+ require.NoError(t, rdb.XAdd(ctx,
&redis.XAddArgs{Stream: "mystream", ID: fmt.Sprintf("%d-0", i), Values:
[]string{"f", "v"}}).Err())
+ }
+ require.NoError(t, rdb.XGroupCreate(ctx, "mystream", "g1",
"0").Err())
+ require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
+ Group: "g1",
+ Consumer: "c1",
+ Streams: []string{"mystream", ">"},
+ Count: 3,
+ }).Err())
+
+ require.NoError(t, rdb.XTrimMinID(ctx, "mystream", "4-0").Err())
+ require.EqualValues(t, 0, rdb.XLen(ctx, "mystream").Val())
+
+ groups := rdb.XInfoGroups(ctx, "mystream").Val()
+ require.Len(t, groups, 1)
+ require.EqualValues(t, "g1", groups[0].Name)
+ require.EqualValues(t, 3, groups[0].Pending)
+ require.EqualValues(t, "3-0", groups[0].LastDeliveredID)
+
+ // groups[0].Pending above is only the count cached in group
metadata; read the PEL
+ // subkeys back with XPENDING's extended form to prove the
entries themselves survived
+ // the range tombstone (which stops at the first
group/consumer/PEL subkey).
+ pending, err := rdb.XPendingExt(ctx, &redis.XPendingExtArgs{
+ Stream: "mystream",
+ Group: "g1",
+ Start: "-",
+ End: "+",
+ Count: 10,
+ }).Result()
+ require.NoError(t, err)
+ require.Len(t, pending, 3)
+ require.EqualValues(t, "1-0", pending[0].ID)
+ require.EqualValues(t, "2-0", pending[1].ID)
+ require.EqualValues(t, "3-0", pending[2].ID)
+ })
+
+ t.Run("XADD after trimming a stream to empty keeps the new entries
visible", func(t *testing.T) {
+ require.NoError(t, rdb.Del(ctx, "mystream").Err())
+ for i := 1; i <= 3; i++ {
+ require.NoError(t, rdb.XAdd(ctx,
&redis.XAddArgs{Stream: "mystream", ID: fmt.Sprintf("%d-0", i), Values:
[]string{"f", "v"}}).Err())
+ }
+ require.NoError(t, rdb.XTrimMaxLen(ctx, "mystream", 0).Err())
+ require.EqualValues(t, 0, rdb.XLen(ctx, "mystream").Val())
+
+ require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{Stream:
"mystream", ID: "4-0", Values: []string{"f", "v"}}).Err())
+ require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{Stream:
"mystream", ID: "5-0", Values: []string{"f", "v"}}).Err())
+ items := rdb.XRange(ctx, "mystream", "-", "+").Val()
+ require.Len(t, items, 2)
+ require.EqualValues(t, "4-0", items[0].ID)
+ require.EqualValues(t, "5-0", items[1].ID)
+ })
+
+ t.Run("repeated XTRIM MINID interleaved with XADD keeps exact survivors
and bookkeeping", func(t *testing.T) {
+ require.NoError(t, rdb.Del(ctx, "mystream").Err())
+ for i := 1; i <= 4; i++ {
+ require.NoError(t, rdb.XAdd(ctx,
&redis.XAddArgs{Stream: "mystream", ID: fmt.Sprintf("%d-0", i), Values:
[]string{"f", "v"}}).Err())
+ }
+ require.NoError(t, rdb.XTrimMinID(ctx, "mystream", "3-0").Err())
+ require.EqualValues(t, 2, rdb.XLen(ctx, "mystream").Val())
+
+ for i := 5; i <= 6; i++ {
+ require.NoError(t, rdb.XAdd(ctx,
&redis.XAddArgs{Stream: "mystream", ID: fmt.Sprintf("%d-0", i), Values:
[]string{"f", "v"}}).Err())
+ }
+ require.NoError(t, rdb.XTrimMinID(ctx, "mystream", "5-0").Err())
+
+ items := rdb.XRange(ctx, "mystream", "-", "+").Val()
+ require.Len(t, items, 2)
+ require.EqualValues(t, "5-0", items[0].ID)
+ require.EqualValues(t, "6-0", items[1].ID)
+
+ r := rdb.XInfoStreamFull(ctx, "mystream", 0).Val()
+ require.Equal(t, "5-0", r.RecordedFirstEntryID)
+ require.Equal(t, "4-0", r.MaxDeletedEntryID)
+ require.EqualValues(t, 6, r.EntriesAdded)
+ })
+
t.Run("XADD with LIMIT consecutive calls", func(t *testing.T) {
require.NoError(t, rdb.Del(ctx, "mystream").Err())
for i := 0; i < 100; i++ {
