Copilot commented on code in PR #3504:
URL: https://github.com/apache/kvrocks/pull/3504#discussion_r3367463771
##########
src/storage/batch_extractor.cc:
##########
@@ -38,6 +39,8 @@ void WriteBatchExtractor::LogData(const rocksdb::Slice &blob)
{
// Redis type log data
if (auto s = log_data_.Decode(blob); !s.IsOK()) {
WARN("Failed to decode Redis type log: {}", s.Msg());
+ } else {
+ seen_xackdel_entry_keys_.clear();
}
}
Review Comment:
`seen_xackdel_entry_keys_` is only cleared when `log_data_.Decode(blob)`
succeeds. If decoding fails for a new batch, the dedup set can retain keys from
the previous batch and incorrectly suppress XACKDEL replication commands in
subsequent `DeleteCF` calls. Clear the set unconditionally at the start of the
Redis-type log-data branch to avoid stale dedup state.
##########
src/types/redis_stream.cc:
##########
@@ -394,6 +395,425 @@ rocksdb::Status Stream::DeletePelEntries(engine::Context
&ctx, const Slice &stre
return storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
}
+rocksdb::Status Stream::getGroupNames(engine::Context &ctx, const std::string
&ns_key, const StreamMetadata &metadata,
+ std::vector<std::string> *group_names) {
+ group_names->clear();
+
+ std::string subkey_type_delimiter;
+ PutFixed64(&subkey_type_delimiter, UINT64_MAX);
+ PutFixed8(&subkey_type_delimiter,
static_cast<uint8_t>(StreamSubkeyType::StreamConsumerGroupMetadata));
+
+ std::string next_version_prefix_key =
+ InternalKey(ns_key, subkey_type_delimiter, metadata.version + 1,
storage_->IsSlotIdEncoded()).Encode();
+ std::string prefix_key =
+ InternalKey(ns_key, subkey_type_delimiter, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+
+ rocksdb::ReadOptions read_options = ctx.DefaultScanOptions();
+ rocksdb::Slice upper_bound(next_version_prefix_key);
+ read_options.iterate_upper_bound = &upper_bound;
+ rocksdb::Slice lower_bound(prefix_key);
+ read_options.iterate_lower_bound = &lower_bound;
+
+ auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_);
+ for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+ if (identifySubkeyType(iter->key()) !=
StreamSubkeyType::StreamConsumerGroupMetadata) {
+ continue;
+ }
+ group_names->push_back(groupNameFromInternalKey(iter->key()));
+ }
+ return iter->status();
+}
+
+rocksdb::Status Stream::deleteEntryAndUpdateMeta(rocksdb::WriteBatchBase
*batch, const std::string &entry_key,
+ const StreamEntryID &id,
StreamMetadata *metadata,
+ uint64_t *deleted_cnt) {
+ rocksdb::Status s = batch->Delete(stream_cf_handle_, entry_key);
+ if (!s.ok()) return s;
+ (*deleted_cnt)++;
+
+ if (metadata->max_deleted_entry_id < id) {
+ metadata->max_deleted_entry_id = id;
+ }
+
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Stream::cleanPelFromAllGroups(
+ engine::Context &ctx, const std::string &ns_key, const StreamMetadata
&metadata, const StreamEntryID &id,
+ rocksdb::WriteBatchBase *batch, bool *batch_modified, const
std::vector<std::string> &group_names,
+ std::map<std::string, uint64_t> *group_pending_decrements,
+ std::map<std::string, std::map<std::string, uint64_t>>
*consumer_pending_decrements) {
+ for (const auto &group_name : group_names) {
+ std::string pel_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata,
group_name, id);
+ std::string pel_value;
+ auto pel_s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_,
pel_key, &pel_value);
+ if (pel_s.ok()) {
+ rocksdb::Status s = batch->Delete(stream_cf_handle_, pel_key);
+ if (!s.ok()) return s;
+ *batch_modified = true;
+
+ auto pel_entry = decodeStreamPelEntryValue(pel_value);
+ (*group_pending_decrements)[group_name]++;
+ (*consumer_pending_decrements)[group_name][pel_entry.consumer_name]++;
+ } else if (!pel_s.IsNotFound()) {
+ return pel_s;
+ }
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Stream::flushPendingNumberUpdates(
+ engine::Context &ctx, const std::string &ns_key, const StreamMetadata
&metadata, rocksdb::WriteBatchBase *batch,
+ const std::map<std::string, uint64_t> &group_pending_decrements,
+ const std::map<std::string, std::map<std::string, uint64_t>>
&consumer_pending_decrements) {
+ for (const auto &[group_name, decrement] : group_pending_decrements) {
+ auto group_key = internalKeyFromGroupName(ns_key, metadata, group_name);
+ std::string group_value;
+ auto s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_,
group_key, &group_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.ok()) {
+ auto group_meta = decodeStreamConsumerGroupMetadataValue(group_value);
+ group_meta.pending_number -= decrement;
+ s = batch->Put(stream_cf_handle_, group_key,
encodeStreamConsumerGroupMetadataValue(group_meta));
+ if (!s.ok()) return s;
+ }
+ }
+
+ for (const auto &[group_name, consumers] : consumer_pending_decrements) {
+ for (const auto &[consumer_name, decrement] : consumers) {
+ auto consumer_key = internalKeyFromConsumerName(ns_key, metadata,
group_name, consumer_name);
+ std::string consumer_value;
+ auto s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_,
consumer_key, &consumer_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.ok()) {
+ auto consumer_meta = decodeStreamConsumerMetadataValue(consumer_value);
+ consumer_meta.pending_number -= decrement;
+ s = batch->Put(stream_cf_handle_, consumer_key,
encodeStreamConsumerMetadataValue(consumer_meta));
+ if (!s.ok()) return s;
+ }
+ }
+ }
+
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Stream::isAckedByAllGroups(
+ engine::Context &ctx, const std::string &ns_key, const StreamMetadata
&metadata, const StreamEntryID &id,
+ const std::vector<std::string> &group_names,
+ const std::unordered_map<std::string, StreamEntryID>
&last_delivered_ids_by_group, bool *all_acked) {
+ *all_acked = true;
+ for (const auto &group_name : group_names) {
+ // A group with last_delivered_id < id has never delivered this entry,
+ // so it cannot be considered acknowledged.
+ auto it = last_delivered_ids_by_group.find(group_name);
+ if (it != last_delivered_ids_by_group.end() && id > it->second) {
+ *all_acked = false;
+ return rocksdb::Status::OK();
+ }
+
+ std::string pel_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata,
group_name, id);
+ std::string pel_value;
+ auto pel_s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_,
pel_key, &pel_value);
+ if (pel_s.ok()) {
+ *all_acked = false;
+ return rocksdb::Status::OK();
+ } else if (!pel_s.IsNotFound()) {
+ return pel_s;
+ }
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Stream::DeleteEntriesAndAck(engine::Context &ctx, const Slice
&stream_name,
+ const std::string &group_name,
const std::vector<StreamEntryID> &ids,
+ StreamDeleteOption option,
std::vector<int> *results) {
+ results->assign(ids.size(),
static_cast<int>(StreamEntryDeleteResult::kEntryNotFound));
+
+ std::string ns_key = AppendNamespacePrefix(stream_name);
+
+ StreamMetadata metadata(false);
+ rocksdb::Status s = GetMetadata(ctx, ns_key, &metadata);
+ if (!s.ok()) {
+ if (s.IsNotFound()) {
+ // Missing keys return per-ID not-found results, not a command error.
+ return rocksdb::Status::OK();
+ }
+ return s;
+ }
+
+ if (ids.empty()) {
+ return rocksdb::Status::OK();
+ }
+
+ std::string group_key = internalKeyFromGroupName(ns_key, metadata,
group_name);
+ std::string get_group_value;
+ s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_, group_key,
&get_group_value);
+ if (!s.ok()) {
+ if (s.IsNotFound()) {
+ return rocksdb::Status::OK();
+ }
+ return s;
+ }
+
+ auto batch = storage_->GetWriteBatchBase();
+ std::string option_str;
+ switch (option) {
+ case StreamDeleteOption::DelRef:
+ option_str = "DELREF";
+ break;
+ case StreamDeleteOption::Acked:
+ option_str = "ACKED";
+ break;
+ default:
+ option_str = "KEEPREF";
+ break;
+ }
+ WriteBatchLogData log_data(kRedisStream, {"XACKDEL", group_name,
option_str});
+ s = batch->PutLogData(log_data.Encode());
+ if (!s.ok()) return s;
+
+ std::string next_version_prefix_key =
+ InternalKey(ns_key, "", metadata.version + 1,
storage_->IsSlotIdEncoded()).Encode();
+ std::string prefix_key = InternalKey(ns_key, "", metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+
+ rocksdb::ReadOptions read_options = ctx.DefaultScanOptions();
+ rocksdb::Slice upper_bound(next_version_prefix_key);
+ read_options.iterate_upper_bound = &upper_bound;
+ rocksdb::Slice lower_bound(prefix_key);
+ read_options.iterate_lower_bound = &lower_bound;
+
+ auto iter = util::UniqueIterator(ctx, read_options, stream_cf_handle_);
+
+ std::vector<std::string> all_groups;
+ bool need_groups = (option == StreamDeleteOption::DelRef || option ==
StreamDeleteOption::Acked);
+ if (need_groups) {
+ s = getGroupNames(ctx, ns_key, metadata, &all_groups);
+ if (!s.ok()) return s;
+ }
+
+ // Prefetch last-delivered IDs so ACKED can distinguish acked entries
+ // from entries that were never delivered.
+ std::unordered_map<std::string, StreamEntryID> last_delivered_ids_by_group;
+ if (option == StreamDeleteOption::Acked) {
+ for (const auto &candidate_group_name : all_groups) {
+ std::string group_metadata_key = internalKeyFromGroupName(ns_key,
metadata, candidate_group_name);
+ std::string group_metadata_value;
+ s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_,
group_metadata_key, &group_metadata_value);
+ if (s.ok()) {
+ auto group_metadata =
decodeStreamConsumerGroupMetadataValue(group_metadata_value);
+ last_delivered_ids_by_group[candidate_group_name] =
group_metadata.last_delivered_id;
+ } else if (!s.IsNotFound()) {
+ return s;
+ }
+ }
+ }
+
+ std::map<std::string, uint64_t> consumer_acknowledges;
+ std::map<std::string, uint64_t> other_group_pending_decrements;
+ std::map<std::string, std::map<std::string, uint64_t>>
other_consumer_pending_decrements;
+ uint64_t deleted_cnt = 0;
+ uint64_t acknowledged_cnt = 0;
+ bool batch_modified = false;
+
+ std::unordered_set<std::string> seen_entry_keys;
+ seen_entry_keys.reserve(ids.size());
+ std::unordered_set<std::string> deleted_entry_keys;
+ deleted_entry_keys.reserve(ids.size());
+ StreamEntryID original_first_entry_id = metadata.first_entry_id;
+ StreamEntryID original_last_entry_id = metadata.last_entry_id;
+
+ for (size_t i = 0; i < ids.size(); i++) {
+ const auto &id = ids[i];
+
+ std::string entry_key = internalKeyFromEntryID(ns_key, metadata, id);
+
+ if (!seen_entry_keys.insert(entry_key).second) {
+ (*results)[i] =
static_cast<int>(StreamEntryDeleteResult::kEntryNotFound);
+ continue;
+ }
+
+ // Look up the current group's PEL entry first.
+ std::string pel_key = internalPelKeyFromGroupAndEntryId(ns_key, metadata,
group_name, id);
+ std::string pel_value;
+ s = storage_->Get(ctx, ctx.GetReadOptions(), stream_cf_handle_, pel_key,
&pel_value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ (*results)[i] =
static_cast<int>(StreamEntryDeleteResult::kEntryNotFound);
+ continue;
+ }
+
+ s = batch->Delete(stream_cf_handle_, pel_key);
+ if (!s.ok()) return s;
+ acknowledged_cnt++;
+ batch_modified = true;
+
+ auto pel_entry = decodeStreamPelEntryValue(pel_value);
+ consumer_acknowledges[pel_entry.consumer_name]++;
+
+ std::string value;
+ s = storage_->Get(ctx, read_options, stream_cf_handle_, entry_key, &value);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ bool stream_entry_exists = s.ok();
+
+ std::vector<std::string> other_groups;
+ if (need_groups) {
+ for (const auto &candidate_group_name : all_groups) {
+ if (candidate_group_name != group_name)
other_groups.push_back(candidate_group_name);
+ }
+ }
Review Comment:
`other_groups` is rebuilt on every ID, and it grows to (groups-1) elements.
When processing many IDs, this repeatedly reallocates and copies strings,
increasing CPU and heap churn. At minimum, reserve capacity before filling to
avoid per-ID reallocation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
