Copilot commented on code in PR #3481:
URL: https://github.com/apache/kvrocks/pull/3481#discussion_r3186173542
##########
src/storage/redis_metadata.cc:
##########
@@ -644,3 +644,48 @@ rocksdb::Status TimeSeriesMetadata::Decode(Slice *input) {
return rocksdb::Status::OK();
}
+
+void CuckooChainMetadata::Encode(std::string *dst) const {
+ Metadata::Encode(dst);
+
+ PutFixed16(dst, n_filters);
+ PutFixed16(dst, expansion);
+ PutFixed64(dst, base_capacity);
+ PutFixed8(dst, bucket_size);
+ PutFixed16(dst, max_iterations);
+ PutFixed64(dst, num_deleted_items);
+}
+
+rocksdb::Status CuckooChainMetadata::Decode(Slice *input) {
+ if (auto s = Metadata::Decode(input); !s.ok()) {
+ return s;
+ }
+
+ if (input->size() < 21) {
Review Comment:
CuckooChainMetadata::Decode checks `input->size() < 21`, but Encode writes
23 bytes after the base Metadata (2+2+8+1+2+8). This can lead to out-of-bounds
reads when decoding truncated metadata. Update the minimum-length check to
match the encoded field sizes (preferably computed from `sizeof` to avoid
drift).
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "redis_cuckoo_chain.h"
+
+#include <unordered_map>
+
+#include "cuckoo_filter.h"
+#include "logging.h"
+
+namespace redis {
+
+rocksdb::Status CuckooChain::getCuckooChainMetadata(engine::Context &ctx,
const Slice &ns_key,
+ CuckooChainMetadata
*metadata) {
+ return Database::GetMetadata(ctx, {kRedisCuckooFilter}, ns_key, metadata);
+}
+
+std::string CuckooChain::getBucketKey(const Slice &ns_key, const
CuckooChainMetadata &metadata, uint16_t filter_index,
+ uint32_t bucket_index) {
+ // Create a sub-key that includes both filter index and bucket index
+ std::string sub_key;
+ PutFixed16(&sub_key, filter_index);
+ PutFixed32(&sub_key, bucket_index);
+
+ // Create the internal key using the storage encoding
+ std::string bucket_key = InternalKey(ns_key, sub_key, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ return bucket_key;
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint8_t expansion) {
+ // Validate parameters
+ if (capacity == 0) {
+ return rocksdb::Status::InvalidArgument("capacity must be larger than 0");
+ }
+
+ // RedisBloom requires minimum capacity to ensure at least one bucket can be
created
+ // With load factor 0.955, capacity=1 and bucket_size=4 results in 0 buckets
+ if (capacity < 2) {
+ return rocksdb::Status::InvalidArgument("capacity must be at least 2");
+ }
+
+ if (bucket_size == 0 || bucket_size > 255) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be between 1 and
255");
+ }
+
+ if (max_iterations == 0) {
+ return rocksdb::Status::InvalidArgument("max_iterations must be larger
than 0");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Check if the key already exists
+ // Read without snapshot to ensure we see any committed data
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (s.ok()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+ if (!s.IsNotFound()) {
+ return s; // Return other errors
Review Comment:
Reserve checks key existence via a raw `storage_->Get` on the metadata CF
with custom ReadOptions. This bypasses Database::GetMetadata parsing (type
validation + expiry handling) and may incorrectly reject a key that is expired
(still present in metadata CF) or mishandle wrong-type cases. Consider using
`getCuckooChainMetadata()`/`Database::GetMetadata()` like BloomChain does, and
treat `IsNotFound()` (including expired) as creatable.
##########
src/types/redis_cuckoo_chain.h:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#pragma once
+
+#include "cuckoo_filter.h"
+#include "storage/redis_db.h"
+#include "storage/redis_metadata.h"
+
+namespace redis {
+
+// Default values for Cuckoo Filter
+const uint32_t kCFDefaultCapacity = 1024;
+const uint8_t kCFDefaultBucketSize = 4; // 4 fingerprints per bucket
+const uint16_t kCFDefaultMaxIterations = 500;
+const uint8_t kCFDefaultExpansion = 2;
+
+class CuckooChain : public Database {
+ public:
+ CuckooChain(engine::Storage *storage, const std::string &ns) :
Database(storage, ns) {}
+
+ // CF.RESERVE command - creates a new cuckoo filter with specified capacity
+ rocksdb::Status Reserve(engine::Context &ctx, const Slice &user_key,
uint64_t capacity, uint8_t bucket_size,
+ uint16_t max_iterations, uint8_t expansion);
+
+ // CF.ADD command - adds an item to the cuckoo filter
+ // Returns true if item was added, false if item already exists (probably)
Review Comment:
The header comment says CF.ADD returns false when the item already exists,
but the current Add implementation never checks for existing fingerprints (and
the tests expect duplicates to always be “added”). This makes the `added`
out-param misleading and makes CF.ADD effectively always return 1 on success.
Either implement the (probabilistic) existence check in the two candidate
buckets or adjust the API contract/comment to match the behavior.
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "redis_cuckoo_chain.h"
+
+#include <unordered_map>
+
+#include "cuckoo_filter.h"
+#include "logging.h"
+
+namespace redis {
+
+rocksdb::Status CuckooChain::getCuckooChainMetadata(engine::Context &ctx,
const Slice &ns_key,
+ CuckooChainMetadata
*metadata) {
+ return Database::GetMetadata(ctx, {kRedisCuckooFilter}, ns_key, metadata);
+}
+
+std::string CuckooChain::getBucketKey(const Slice &ns_key, const
CuckooChainMetadata &metadata, uint16_t filter_index,
+ uint32_t bucket_index) {
+ // Create a sub-key that includes both filter index and bucket index
+ std::string sub_key;
+ PutFixed16(&sub_key, filter_index);
+ PutFixed32(&sub_key, bucket_index);
+
+ // Create the internal key using the storage encoding
+ std::string bucket_key = InternalKey(ns_key, sub_key, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ return bucket_key;
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint8_t expansion) {
+ // Validate parameters
+ if (capacity == 0) {
+ return rocksdb::Status::InvalidArgument("capacity must be larger than 0");
+ }
+
+ // RedisBloom requires minimum capacity to ensure at least one bucket can be
created
+ // With load factor 0.955, capacity=1 and bucket_size=4 results in 0 buckets
+ if (capacity < 2) {
+ return rocksdb::Status::InvalidArgument("capacity must be at least 2");
+ }
+
+ if (bucket_size == 0 || bucket_size > 255) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be between 1 and
255");
+ }
+
+ if (max_iterations == 0) {
+ return rocksdb::Status::InvalidArgument("max_iterations must be larger
than 0");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Check if the key already exists
+ // Read without snapshot to ensure we see any committed data
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (s.ok()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+ if (!s.IsNotFound()) {
+ return s; // Return other errors
+ }
+
+ // Initialize metadata for the new cuckoo filter
+ CuckooChainMetadata metadata;
+
+ // Initialize metadata for the new cuckoo filter
+ metadata.size = 0;
+ metadata.base_capacity = capacity;
+ metadata.bucket_size = bucket_size;
+ metadata.max_iterations = max_iterations;
+ metadata.expansion = expansion;
+ metadata.n_filters = 1;
+ metadata.num_deleted_items = 0;
+
+ // Calculate the number of buckets needed for this filter
+ uint32_t num_buckets = CuckooFilter::OptimalNumBuckets(capacity,
bucket_size);
+
+ INFO("Creating cuckoo filter with capacity={}, bucket_size={},
num_buckets={}, max_iterations={}, expansion={}",
+ capacity, bucket_size, num_buckets, max_iterations,
static_cast<int>(expansion));
+
+ // Create a write batch for atomic operation
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.RESERVE", user_key.ToString()});
+ batch->PutLogData(log_data.Encode());
+
+ // Store the metadata
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
Review Comment:
WriteBatch operations ignore return statuses (e.g., `PutLogData`, `Put` on
metadata/bucket keys). Other code (e.g., BloomChain) checks these statuses and
propagates failures. Consider checking and returning early on any batch
operation failure to avoid silently dropping writes.
##########
src/types/cuckoo_filter.h:
##########
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "vendor/murmurhash2.h"
+
+namespace redis {
+
+// Cuckoo filter implementation from the paper:
+// "Cuckoo Filter: Practically Better Than Bloom" by Fan et al.
+// This is a bucket-based storage implementation where each bucket is stored
+// as an independent key-value pair in RocksDB
+//
+// Hash calculation follows RedisBloom's design:
+// - fp = hash % 255 + 1 (fingerprint, non-zero, range: 1-255)
+// - h1 = hash (primary hash)
+// - h2 = h1 ^ (fp * 0x5bd1e995) (alternate hash via XOR)
+// - bucket_index = hash % num_buckets (only apply modulo when indexing)
+class CuckooFilter {
+ public:
+ // Calculate the optimal number of buckets for the filter
+ static uint32_t OptimalNumBuckets(uint64_t capacity, uint8_t bucket_size) {
+ // A load factor of 95.5% is chosen for the cuckoo filter
+ auto num_buckets = static_cast<uint32_t>(static_cast<long
double>(capacity) / bucket_size / 0.955L);
+ // Round up to next power of 2 for better hash distribution
+ if (num_buckets == 0) num_buckets = 1;
+ uint32_t power = 1;
+ while (power < num_buckets) power <<= 1;
+ return power;
+ }
Review Comment:
OptimalNumBuckets truncates a `uint64_t capacity` into a `uint32_t
num_buckets` via cast. For large capacities this can overflow/truncate silently
(potentially producing a tiny filter). Either validate `capacity/bucket_size`
so the bucket count fits in 32 bits (and return InvalidArgument when it
doesn’t) or switch the bucket count/indexing and encoding to 64-bit.
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "redis_cuckoo_chain.h"
+
+#include <unordered_map>
+
+#include "cuckoo_filter.h"
+#include "logging.h"
+
+namespace redis {
+
+rocksdb::Status CuckooChain::getCuckooChainMetadata(engine::Context &ctx,
const Slice &ns_key,
+ CuckooChainMetadata
*metadata) {
+ return Database::GetMetadata(ctx, {kRedisCuckooFilter}, ns_key, metadata);
+}
+
+std::string CuckooChain::getBucketKey(const Slice &ns_key, const
CuckooChainMetadata &metadata, uint16_t filter_index,
+ uint32_t bucket_index) {
+ // Create a sub-key that includes both filter index and bucket index
+ std::string sub_key;
+ PutFixed16(&sub_key, filter_index);
+ PutFixed32(&sub_key, bucket_index);
+
+ // Create the internal key using the storage encoding
+ std::string bucket_key = InternalKey(ns_key, sub_key, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ return bucket_key;
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint8_t expansion) {
+ // Validate parameters
+ if (capacity == 0) {
+ return rocksdb::Status::InvalidArgument("capacity must be larger than 0");
+ }
+
+ // RedisBloom requires minimum capacity to ensure at least one bucket can be
created
+ // With load factor 0.955, capacity=1 and bucket_size=4 results in 0 buckets
+ if (capacity < 2) {
+ return rocksdb::Status::InvalidArgument("capacity must be at least 2");
+ }
+
+ if (bucket_size == 0 || bucket_size > 255) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be between 1 and
255");
+ }
+
+ if (max_iterations == 0) {
+ return rocksdb::Status::InvalidArgument("max_iterations must be larger
than 0");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Check if the key already exists
+ // Read without snapshot to ensure we see any committed data
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (s.ok()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+ if (!s.IsNotFound()) {
+ return s; // Return other errors
+ }
+
+ // Initialize metadata for the new cuckoo filter
+ CuckooChainMetadata metadata;
+
+ // Initialize metadata for the new cuckoo filter
+ metadata.size = 0;
+ metadata.base_capacity = capacity;
+ metadata.bucket_size = bucket_size;
+ metadata.max_iterations = max_iterations;
+ metadata.expansion = expansion;
+ metadata.n_filters = 1;
+ metadata.num_deleted_items = 0;
+
+ // Calculate the number of buckets needed for this filter
+ uint32_t num_buckets = CuckooFilter::OptimalNumBuckets(capacity,
bucket_size);
+
+ INFO("Creating cuckoo filter with capacity={}, bucket_size={},
num_buckets={}, max_iterations={}, expansion={}",
+ capacity, bucket_size, num_buckets, max_iterations,
static_cast<int>(expansion));
+
+ // Create a write batch for atomic operation
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.RESERVE", user_key.ToString()});
+ batch->PutLogData(log_data.Encode());
+
+ // Store the metadata
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
+
+ // Note: With bucket-based storage, we don't pre-allocate all buckets
+ // Buckets will be created lazily on first write
+ // This saves memory for sparse filters
+
+ // Optionally, we could create the first few buckets to ensure the filter is
ready
+ // But for now, we'll keep it fully lazy for maximum memory efficiency
+
+ return storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+}
+
+// Helper function: calculate integer power (avoid std::pow for integers)
+static uint64_t IntPow(uint64_t base, uint16_t exp) {
+ uint64_t result = 1;
+ for (uint16_t i = 0; i < exp; ++i) {
+ result *= base;
+ }
+ return result;
+}
+
+// Helper function: try to find empty slot in a bucket and insert fingerprint
+static bool TryInsertInBucket(std::string &bucket_data, uint8_t bucket_size,
uint8_t fingerprint, size_t *slot_idx) {
+ for (size_t i = 0; i < bucket_size; ++i) {
+ if (static_cast<uint8_t>(bucket_data[i]) == 0) {
+ bucket_data[i] = static_cast<char>(fingerprint);
+ *slot_idx = i;
+ return true;
+ }
+ }
+ return false;
+}
+
+// Helper function: read bucket from storage and ensure correct size
+static rocksdb::Status ReadBucket(engine::Storage *storage, engine::Context
&ctx, const std::string &bucket_key,
+ uint8_t bucket_size, std::string
*bucket_data) {
+ rocksdb::ReadOptions read_opts = ctx.DefaultScanOptions();
Review Comment:
ReadBucket uses `ctx.DefaultScanOptions()` for point `Get` operations. Scan
options can differ from normal read options (e.g., snapshot usage, cache/prefix
settings) and may break transactional/snapshot semantics for CF.ADD. Use
`ctx.GetReadOptions()` (or another point-read option) for bucket gets.
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "redis_cuckoo_chain.h"
+
+#include <unordered_map>
+
+#include "cuckoo_filter.h"
+#include "logging.h"
+
+namespace redis {
+
+rocksdb::Status CuckooChain::getCuckooChainMetadata(engine::Context &ctx,
const Slice &ns_key,
+ CuckooChainMetadata
*metadata) {
+ return Database::GetMetadata(ctx, {kRedisCuckooFilter}, ns_key, metadata);
+}
+
+std::string CuckooChain::getBucketKey(const Slice &ns_key, const
CuckooChainMetadata &metadata, uint16_t filter_index,
+ uint32_t bucket_index) {
+ // Create a sub-key that includes both filter index and bucket index
+ std::string sub_key;
+ PutFixed16(&sub_key, filter_index);
+ PutFixed32(&sub_key, bucket_index);
+
+ // Create the internal key using the storage encoding
+ std::string bucket_key = InternalKey(ns_key, sub_key, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ return bucket_key;
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint8_t expansion) {
+ // Validate parameters
+ if (capacity == 0) {
+ return rocksdb::Status::InvalidArgument("capacity must be larger than 0");
+ }
+
+ // RedisBloom requires minimum capacity to ensure at least one bucket can be
created
+ // With load factor 0.955, capacity=1 and bucket_size=4 results in 0 buckets
+ if (capacity < 2) {
+ return rocksdb::Status::InvalidArgument("capacity must be at least 2");
+ }
+
+ if (bucket_size == 0 || bucket_size > 255) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be between 1 and
255");
+ }
+
+ if (max_iterations == 0) {
+ return rocksdb::Status::InvalidArgument("max_iterations must be larger
than 0");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Check if the key already exists
+ // Read without snapshot to ensure we see any committed data
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (s.ok()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+ if (!s.IsNotFound()) {
+ return s; // Return other errors
+ }
+
+ // Initialize metadata for the new cuckoo filter
+ CuckooChainMetadata metadata;
+
+ // Initialize metadata for the new cuckoo filter
+ metadata.size = 0;
+ metadata.base_capacity = capacity;
+ metadata.bucket_size = bucket_size;
+ metadata.max_iterations = max_iterations;
+ metadata.expansion = expansion;
+ metadata.n_filters = 1;
+ metadata.num_deleted_items = 0;
+
+ // Calculate the number of buckets needed for this filter
+ uint32_t num_buckets = CuckooFilter::OptimalNumBuckets(capacity,
bucket_size);
+
+ INFO("Creating cuckoo filter with capacity={}, bucket_size={},
num_buckets={}, max_iterations={}, expansion={}",
+ capacity, bucket_size, num_buckets, max_iterations,
static_cast<int>(expansion));
+
+ // Create a write batch for atomic operation
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.RESERVE", user_key.ToString()});
+ batch->PutLogData(log_data.Encode());
+
+ // Store the metadata
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
+
+ // Note: With bucket-based storage, we don't pre-allocate all buckets
+ // Buckets will be created lazily on first write
+ // This saves memory for sparse filters
+
+ // Optionally, we could create the first few buckets to ensure the filter is
ready
+ // But for now, we'll keep it fully lazy for maximum memory efficiency
+
+ return storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+}
+
+// Helper function: calculate integer power (avoid std::pow for integers)
+static uint64_t IntPow(uint64_t base, uint16_t exp) {
+ uint64_t result = 1;
+ for (uint16_t i = 0; i < exp; ++i) {
+ result *= base;
+ }
+ return result;
+}
+
+// Helper function: try to find empty slot in a bucket and insert fingerprint
+static bool TryInsertInBucket(std::string &bucket_data, uint8_t bucket_size,
uint8_t fingerprint, size_t *slot_idx) {
+ for (size_t i = 0; i < bucket_size; ++i) {
+ if (static_cast<uint8_t>(bucket_data[i]) == 0) {
+ bucket_data[i] = static_cast<char>(fingerprint);
+ *slot_idx = i;
+ return true;
+ }
+ }
+ return false;
+}
+
+// Helper function: read bucket from storage and ensure correct size
+static rocksdb::Status ReadBucket(engine::Storage *storage, engine::Context
&ctx, const std::string &bucket_key,
+ uint8_t bucket_size, std::string
*bucket_data) {
+ rocksdb::ReadOptions read_opts = ctx.DefaultScanOptions();
+ auto s = storage->Get(ctx, read_opts, bucket_key, bucket_data);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ bucket_data->clear();
+ }
+ if (bucket_data->size() < bucket_size) {
+ bucket_data->resize(bucket_size, 0);
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status CuckooChain::Add(engine::Context &ctx, const Slice &user_key,
const Slice &item, bool *added) {
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Get metadata - use read options without snapshot to see latest data
+ CuckooChainMetadata metadata(false);
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (!s.ok()) {
+ if (s.IsNotFound()) {
+ return rocksdb::Status::NotFound("key not found");
+ }
+ return s;
+ }
+
+ // Decode metadata
+ Slice slice(raw_value);
+ s = metadata.Decode(&slice);
+ if (!s.ok()) {
+ return s;
+ }
+
Review Comment:
Add fetches and decodes raw metadata directly from RocksDB, skipping
Database::ParseMetadata checks (expired-key handling and wrong-type
validation). This can allow operations on expired keys and diverges from other
types’ behavior. Prefer `getCuckooChainMetadata()`/`Database::GetMetadata()`
and map `kErrMsgKeyExpired` to a NotFound-style response.
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "redis_cuckoo_chain.h"
+
+#include <unordered_map>
+
+#include "cuckoo_filter.h"
+#include "logging.h"
+
+namespace redis {
+
+rocksdb::Status CuckooChain::getCuckooChainMetadata(engine::Context &ctx,
const Slice &ns_key,
+ CuckooChainMetadata
*metadata) {
+ return Database::GetMetadata(ctx, {kRedisCuckooFilter}, ns_key, metadata);
+}
+
+std::string CuckooChain::getBucketKey(const Slice &ns_key, const
CuckooChainMetadata &metadata, uint16_t filter_index,
+ uint32_t bucket_index) {
+ // Create a sub-key that includes both filter index and bucket index
+ std::string sub_key;
+ PutFixed16(&sub_key, filter_index);
+ PutFixed32(&sub_key, bucket_index);
+
+ // Create the internal key using the storage encoding
+ std::string bucket_key = InternalKey(ns_key, sub_key, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ return bucket_key;
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint8_t expansion) {
+ // Validate parameters
+ if (capacity == 0) {
+ return rocksdb::Status::InvalidArgument("capacity must be larger than 0");
+ }
+
+ // RedisBloom requires minimum capacity to ensure at least one bucket can be
created
+ // With load factor 0.955, capacity=1 and bucket_size=4 results in 0 buckets
+ if (capacity < 2) {
+ return rocksdb::Status::InvalidArgument("capacity must be at least 2");
+ }
+
+ if (bucket_size == 0 || bucket_size > 255) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be between 1 and
255");
+ }
+
+ if (max_iterations == 0) {
+ return rocksdb::Status::InvalidArgument("max_iterations must be larger
than 0");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Check if the key already exists
+ // Read without snapshot to ensure we see any committed data
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (s.ok()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+ if (!s.IsNotFound()) {
+ return s; // Return other errors
+ }
+
+ // Initialize metadata for the new cuckoo filter
+ CuckooChainMetadata metadata;
+
+ // Initialize metadata for the new cuckoo filter
+ metadata.size = 0;
+ metadata.base_capacity = capacity;
+ metadata.bucket_size = bucket_size;
+ metadata.max_iterations = max_iterations;
+ metadata.expansion = expansion;
+ metadata.n_filters = 1;
+ metadata.num_deleted_items = 0;
Review Comment:
Cuckoo filters are created with `metadata.size = 0`. In the core Metadata
logic, types that are not considered “emptyable” are treated as expired when
`size == 0`, which can cause a newly reserved CF key to be immediately
considered expired/eligible for compaction deletion. Add `kRedisCuckooFilter`
to `Metadata::IsEmptyableType()` (or otherwise ensure `size==0` is not treated
as expired for this type).
##########
src/commands/cmd_cuckoo_filter.cc:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "command_parser.h"
+#include "commander.h"
+#include "error_constants.h"
+#include "server/server.h"
+#include "types/redis_cuckoo_chain.h"
+
+namespace redis {
+
+class CommandCFReserve : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ // CF.RESERVE key capacity [BUCKETSIZE bs] [MAXITERATIONS mi] [EXPANSION
ex]
+ if (args.size() < 3) {
+ return {Status::RedisParseErr, "wrong number of arguments"};
Review Comment:
These Parse methods return a hardcoded "wrong number of arguments" string.
Most commands in this codebase use shared constants like
`errWrongNumOfArguments` to keep error messages consistent. Consider switching
to the standard constant here as well.
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,421 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "redis_cuckoo_chain.h"
+
+#include <unordered_map>
+
+#include "cuckoo_filter.h"
+#include "logging.h"
+
+namespace redis {
+
+rocksdb::Status CuckooChain::getCuckooChainMetadata(engine::Context &ctx,
const Slice &ns_key,
+ CuckooChainMetadata
*metadata) {
+ return Database::GetMetadata(ctx, {kRedisCuckooFilter}, ns_key, metadata);
+}
+
+std::string CuckooChain::getBucketKey(const Slice &ns_key, const
CuckooChainMetadata &metadata, uint16_t filter_index,
+ uint32_t bucket_index) {
+ // Create a sub-key that includes both filter index and bucket index
+ std::string sub_key;
+ PutFixed16(&sub_key, filter_index);
+ PutFixed32(&sub_key, bucket_index);
+
+ // Create the internal key using the storage encoding
+ std::string bucket_key = InternalKey(ns_key, sub_key, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ return bucket_key;
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint8_t expansion) {
+ // Validate parameters
+ if (capacity == 0) {
+ return rocksdb::Status::InvalidArgument("capacity must be larger than 0");
+ }
+
+ // RedisBloom requires minimum capacity to ensure at least one bucket can be
created
+ // With load factor 0.955, capacity=1 and bucket_size=4 results in 0 buckets
+ if (capacity < 2) {
+ return rocksdb::Status::InvalidArgument("capacity must be at least 2");
+ }
+
+ if (bucket_size == 0 || bucket_size > 255) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be between 1 and
255");
+ }
+
+ if (max_iterations == 0) {
+ return rocksdb::Status::InvalidArgument("max_iterations must be larger
than 0");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Check if the key already exists
+ // Read without snapshot to ensure we see any committed data
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (s.ok()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+ if (!s.IsNotFound()) {
+ return s; // Return other errors
+ }
+
+ // Initialize metadata for the new cuckoo filter
+ CuckooChainMetadata metadata;
+
+ // Initialize metadata for the new cuckoo filter
+ metadata.size = 0;
+ metadata.base_capacity = capacity;
+ metadata.bucket_size = bucket_size;
+ metadata.max_iterations = max_iterations;
+ metadata.expansion = expansion;
+ metadata.n_filters = 1;
+ metadata.num_deleted_items = 0;
+
+ // Calculate the number of buckets needed for this filter
+ uint32_t num_buckets = CuckooFilter::OptimalNumBuckets(capacity,
bucket_size);
+
+ INFO("Creating cuckoo filter with capacity={}, bucket_size={},
num_buckets={}, max_iterations={}, expansion={}",
+ capacity, bucket_size, num_buckets, max_iterations,
static_cast<int>(expansion));
+
+ // Create a write batch for atomic operation
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.RESERVE", user_key.ToString()});
+ batch->PutLogData(log_data.Encode());
+
+ // Store the metadata
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
+
+ // Note: With bucket-based storage, we don't pre-allocate all buckets
+ // Buckets will be created lazily on first write
+ // This saves memory for sparse filters
+
+ // Optionally, we could create the first few buckets to ensure the filter is
ready
+ // But for now, we'll keep it fully lazy for maximum memory efficiency
+
+ return storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+}
+
+// Helper function: calculate integer power (avoid std::pow for integers)
+static uint64_t IntPow(uint64_t base, uint16_t exp) {
+ uint64_t result = 1;
+ for (uint16_t i = 0; i < exp; ++i) {
+ result *= base;
+ }
+ return result;
+}
+
+// Helper function: try to find empty slot in a bucket and insert fingerprint
+static bool TryInsertInBucket(std::string &bucket_data, uint8_t bucket_size,
uint8_t fingerprint, size_t *slot_idx) {
+ for (size_t i = 0; i < bucket_size; ++i) {
+ if (static_cast<uint8_t>(bucket_data[i]) == 0) {
+ bucket_data[i] = static_cast<char>(fingerprint);
+ *slot_idx = i;
+ return true;
+ }
+ }
+ return false;
+}
+
+// Helper function: read bucket from storage and ensure correct size
+static rocksdb::Status ReadBucket(engine::Storage *storage, engine::Context
&ctx, const std::string &bucket_key,
+ uint8_t bucket_size, std::string
*bucket_data) {
+ rocksdb::ReadOptions read_opts = ctx.DefaultScanOptions();
+ auto s = storage->Get(ctx, read_opts, bucket_key, bucket_data);
+ if (!s.ok() && !s.IsNotFound()) {
+ return s;
+ }
+ if (s.IsNotFound()) {
+ bucket_data->clear();
+ }
+ if (bucket_data->size() < bucket_size) {
+ bucket_data->resize(bucket_size, 0);
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status CuckooChain::Add(engine::Context &ctx, const Slice &user_key,
const Slice &item, bool *added) {
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ // Get metadata - use read options without snapshot to see latest data
+ CuckooChainMetadata metadata(false);
+ std::string raw_value;
+ rocksdb::ReadOptions read_options; // No snapshot
+ auto s = storage_->Get(ctx, read_options, metadata_cf_handle_, ns_key,
&raw_value);
+ if (!s.ok()) {
+ if (s.IsNotFound()) {
+ return rocksdb::Status::NotFound("key not found");
+ }
+ return s;
+ }
+
+ // Decode metadata
+ Slice slice(raw_value);
+ s = metadata.Decode(&slice);
+ if (!s.ok()) {
+ return s;
+ }
+
+ // Validate metadata
+ if (metadata.n_filters == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: n_filters is 0");
+ }
+
+ // Calculate hash and fingerprint for the item
+ uint64_t hash = CuckooFilter::Hash(item.data(), item.size());
+ uint8_t fingerprint = CuckooFilter::GenerateFingerprint(hash);
+
+ // Try to insert in each sub-filter (starting from the first/smallest one)
+ // This follows RedisBloom's behavior and is more efficient
+ for (uint16_t filter_idx = 0; filter_idx < metadata.n_filters; ++filter_idx)
{
+ // Calculate capacity for this filter using integer power
+ uint64_t filter_capacity = metadata.base_capacity *
IntPow(metadata.expansion, filter_idx);
+ uint32_t num_buckets = CuckooFilter::OptimalNumBuckets(filter_capacity,
metadata.bucket_size);
+
+ // Calculate bucket indices
+ uint32_t bucket1_idx = hash % num_buckets;
+ uint64_t alt_hash = CuckooFilter::GetAltHash(fingerprint, hash);
+ uint32_t bucket2_idx = alt_hash % num_buckets;
+
+ // Read both buckets using helper function
+ std::string bucket1_key = getBucketKey(ns_key, metadata, filter_idx,
bucket1_idx);
+ std::string bucket2_key = getBucketKey(ns_key, metadata, filter_idx,
bucket2_idx);
+
+ std::string bucket1_data, bucket2_data;
+ s = ReadBucket(storage_, ctx, bucket1_key, metadata.bucket_size,
&bucket1_data);
+ if (!s.ok()) return s;
+
+ s = ReadBucket(storage_, ctx, bucket2_key, metadata.bucket_size,
&bucket2_data);
+ if (!s.ok()) return s;
+
+ // Try simple insertion in bucket1 or bucket2
+ size_t slot_idx = 0;
+ std::string *target_bucket_data = nullptr;
+ std::string target_bucket_key;
+
+ if (TryInsertInBucket(bucket1_data, metadata.bucket_size, fingerprint,
&slot_idx)) {
+ target_bucket_data = &bucket1_data;
+ target_bucket_key = bucket1_key;
+ } else if (TryInsertInBucket(bucket2_data, metadata.bucket_size,
fingerprint, &slot_idx)) {
+ target_bucket_data = &bucket2_data;
+ target_bucket_key = bucket2_key;
+ }
+
+ if (target_bucket_data != nullptr) {
+ // Successfully inserted, write to storage atomically
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.ADD", user_key.ToString()});
+ batch->PutLogData(log_data.Encode());
+ batch->Put(target_bucket_key, *target_bucket_data);
+
+ metadata.size++;
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
+
+ s = storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+ if (!s.ok()) return s;
+
+ *added = true;
+ return rocksdb::Status::OK();
+ }
+ }
+
+ // No space found in any filter, try kick-out on the last filter
+ uint16_t last_filter_idx = metadata.n_filters - 1;
+ uint64_t filter_capacity = metadata.base_capacity *
IntPow(metadata.expansion, last_filter_idx);
+ uint32_t num_buckets = CuckooFilter::OptimalNumBuckets(filter_capacity,
metadata.bucket_size);
+
+ bool inserted = false;
+ s = kickOutInsert(ctx, user_key, ns_key, metadata, last_filter_idx,
num_buckets, fingerprint, hash, &inserted);
+ if (s.ok() && inserted) {
+ // Update metadata after successful kick-out
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.ADD", user_key.ToString()});
+ batch->PutLogData(log_data.Encode());
+
+ metadata.size++;
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
+
+ s = storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+ if (!s.ok()) return s;
+
+ *added = true;
Review Comment:
On the kick-out path, `kickOutInsert` persists bucket modifications in its
own write batch, and then Add updates `metadata.size` in a separate write
batch. This loses atomicity: a crash or write failure between the two can leave
the filter mutated while metadata isn’t updated (or vice versa in future
changes). Consider writing bucket changes and metadata update in a single batch
(e.g., have kickOutInsert populate a caller-provided batch or return the
modified buckets for the caller to write together with metadata).
--
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]
