jihuayu commented on code in PR #3481:
URL: https://github.com/apache/kvrocks/pull/3481#discussion_r3232791534
##########
src/storage/redis_metadata.h:
##########
@@ -334,6 +335,50 @@ class BloomChainMetadata : public Metadata {
bool IsScaling() const { return expansion != 0; };
};
+constexpr uint32_t kCuckooFilterDefaultPageSize = 2048;
Review Comment:
We need a comment here to indicate the units.
##########
src/storage/redis_metadata.h:
##########
@@ -334,6 +335,50 @@ class BloomChainMetadata : public Metadata {
bool IsScaling() const { return expansion != 0; };
};
+constexpr uint32_t kCuckooFilterDefaultPageSize = 2048;
+
+class CuckooChainMetadata : public Metadata {
+ public:
+ /// The number of sub-filters in the chain
+ uint16_t n_filters;
+
+ /// Expansion factor for new filters
+ /// When a filter is full, a new one is created with capacity =
base_capacity * expansion^n
+ uint16_t expansion;
+
+ /// The capacity of the first filter
+ uint64_t base_capacity;
Review Comment:
Why we need store it at each Cuckoo Filter? Could you explain your decision?
##########
src/storage/redis_metadata.h:
##########
@@ -54,12 +54,13 @@ enum RedisType : uint8_t {
kRedisHyperLogLog = 11,
kRedisTDigest = 12,
kRedisTimeSeries = 13,
+ kRedisCuckooFilter = 14,
kRedisTypeMax
};
inline constexpr const std::array<std::string_view, kRedisTypeMax>
RedisTypeNames = {
- "none", "string", "hash", "list", "set", "zset",
"bitmap",
- "sortedint", "stream", "MBbloom--", "ReJSON-RL", "hyperloglog",
"TDIS-TYPE", "timeseries"};
+ "none", "string", "hash", "list", "set", "zset",
"bitmap", "sortedint",
+ "stream", "MBbloom--", "ReJSON-RL", "hyperloglog", "TDIS-TYPE",
"timeseries", "cuckoofilter"};
Review Comment:
The type is `MBbloomCF` in redis source.
##########
src/types/cuckoo_filter.h:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 <rocksdb/status.h>
+
+#include <cstdint>
+#include <limits>
+#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.
+// Buckets are grouped into page values in RocksDB. The Cuckoo algorithm still
+// works with logical bucket indexes, while the storage layer maps buckets to
pages.
+//
+// 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:
+ static bool IsCapacitySupported(uint64_t capacity, uint8_t bucket_size) {
+ uint32_t num_buckets = 0;
+ return OptimalNumBuckets(capacity, bucket_size, &num_buckets).ok();
+ }
+
+ // Calculate the optimal number of buckets for the filter.
+ static rocksdb::Status OptimalNumBuckets(uint64_t capacity, uint8_t
bucket_size, uint32_t* num_buckets) {
+ if (bucket_size == 0) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be larger than
0");
+ }
+
+ constexpr long double kLoadFactor = 0.955L;
Review Comment:
We'd better extract it and keep all the constants in one place.
##########
src/types/cuckoo_filter.h:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 <rocksdb/status.h>
+
+#include <cstdint>
+#include <limits>
+#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.
+// Buckets are grouped into page values in RocksDB. The Cuckoo algorithm still
+// works with logical bucket indexes, while the storage layer maps buckets to
pages.
+//
+// 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:
+ static bool IsCapacitySupported(uint64_t capacity, uint8_t bucket_size) {
+ uint32_t num_buckets = 0;
+ return OptimalNumBuckets(capacity, bucket_size, &num_buckets).ok();
+ }
+
+ // Calculate the optimal number of buckets for the filter.
+ static rocksdb::Status OptimalNumBuckets(uint64_t capacity, uint8_t
bucket_size, uint32_t* num_buckets) {
Review Comment:
This function name is no good. May be `CalculateRequiredBuckets` is better.
##########
src/types/cuckoo_filter_page.h:
##########
@@ -0,0 +1,91 @@
+/*
+ * 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 <rocksdb/status.h>
+#include <rocksdb/write_batch.h>
+
+#include <cstdint>
+#include <string>
+#include <unordered_map>
+#include <vector>
+
+#include "storage/redis_metadata.h"
+#include "storage/storage.h"
+
+namespace redis {
+
+class CuckooPageSet {
Review Comment:
The set suffix looks a bit misleading. It's just a single page, not a
collection of pages, so it could easily cause confusion.
And I feel like the API boundary for this class is a bit blurry. The biggest
issue is that it mixes 'storage access' and 'insertion policy' within the same
class's public surface. Callers might get confused about whether this class is
responsible for 'page storage access' or 'Cuckoo bucket insertion policy'.
##########
src/types/redis_cuckoo_chain.h:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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 uint16_t kCFDefaultExpansion = 2;
+const uint16_t kCFMaxExpansion = 32768;
+
+class CuckooChain : public Database {
Review Comment:
It seems we are missing a layer of abstraction, SubFilter.
It feels a bit too low-level for CuckooChain to interact with pages directly.
```text
CuckooChain
-> SubFilter 0
-> Page 0
-> Page 1
-> Page 2
-> SubFilter 1
-> Page 0
-> Page 1
-> SubFilter 2
-> Page 0
-> ...
```
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,350 @@
+/*
+ * 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 <limits>
+
+#include "cuckoo_filter.h"
+#include "cuckoo_filter_page.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);
+}
+
+rocksdb::Status CuckooChain::validateMetadata(const CuckooChainMetadata
&metadata) {
+ if (metadata.n_filters == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: n_filters is 0");
+ }
+ if (metadata.base_capacity == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: base_capacity is 0");
+ }
+ if (metadata.bucket_size == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: bucket_size is 0");
+ }
+ if (metadata.max_iterations == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: max_iterations is
0");
+ }
+ if (metadata.page_size < metadata.bucket_size) {
+ return rocksdb::Status::Corruption("invalid metadata: page_size is smaller
than bucket_size");
+ }
+ if (!CuckooFilter::IsCapacitySupported(metadata.base_capacity,
metadata.bucket_size)) {
+ return rocksdb::Status::Corruption("invalid metadata: base_capacity is too
large");
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint16_t expansion,
+ uint32_t page_size) {
+ 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");
+ }
+ if (page_size == 0) {
+ return rocksdb::Status::InvalidArgument("page_size must be larger than 0");
+ }
+ if (page_size < bucket_size) {
+ return rocksdb::Status::InvalidArgument("page_size must be at least
bucket_size");
+ }
+ if (expansion > kCFMaxExpansion) {
+ return rocksdb::Status::InvalidArgument("expansion must be between 0 and
32768");
+ }
+ if (!CuckooFilter::IsCapacitySupported(capacity, bucket_size)) {
+ return rocksdb::Status::InvalidArgument("capacity is too large");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ CuckooChainMetadata existing_metadata;
+ auto s = getCuckooChainMetadata(ctx, ns_key, &existing_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+
+ CuckooChainMetadata metadata;
+
+ 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;
+ metadata.page_size = page_size;
+
+ // Calculate the number of buckets needed for this filter
+ uint32_t num_buckets = 0;
+ s = CuckooFilter::OptimalNumBuckets(capacity, bucket_size, &num_buckets);
+ if (!s.ok()) return s;
+
+ INFO(
+ "Creating cuckoo filter with capacity={}, bucket_size={},
num_buckets={}, max_iterations={}, expansion={}, "
+ "page_size={}",
+ capacity, bucket_size, num_buckets, max_iterations,
static_cast<int>(expansion), page_size);
+
+ // Create a write batch for atomic operation
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.RESERVE", user_key.ToString()});
Review Comment:
use `reserve` not "CF.RESERVE"
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,350 @@
+/*
+ * 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 <limits>
+
+#include "cuckoo_filter.h"
+#include "cuckoo_filter_page.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);
+}
+
+rocksdb::Status CuckooChain::validateMetadata(const CuckooChainMetadata
&metadata) {
+ if (metadata.n_filters == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: n_filters is 0");
+ }
+ if (metadata.base_capacity == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: base_capacity is 0");
+ }
+ if (metadata.bucket_size == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: bucket_size is 0");
+ }
+ if (metadata.max_iterations == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: max_iterations is
0");
+ }
+ if (metadata.page_size < metadata.bucket_size) {
+ return rocksdb::Status::Corruption("invalid metadata: page_size is smaller
than bucket_size");
+ }
+ if (!CuckooFilter::IsCapacitySupported(metadata.base_capacity,
metadata.bucket_size)) {
+ return rocksdb::Status::Corruption("invalid metadata: base_capacity is too
large");
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint16_t expansion,
+ uint32_t page_size) {
+ 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");
+ }
+ if (page_size == 0) {
+ return rocksdb::Status::InvalidArgument("page_size must be larger than 0");
+ }
+ if (page_size < bucket_size) {
+ return rocksdb::Status::InvalidArgument("page_size must be at least
bucket_size");
+ }
+ if (expansion > kCFMaxExpansion) {
+ return rocksdb::Status::InvalidArgument("expansion must be between 0 and
32768");
+ }
+ if (!CuckooFilter::IsCapacitySupported(capacity, bucket_size)) {
+ return rocksdb::Status::InvalidArgument("capacity is too large");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ CuckooChainMetadata existing_metadata;
+ auto s = getCuckooChainMetadata(ctx, ns_key, &existing_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+
+ CuckooChainMetadata metadata;
+
+ 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;
+ metadata.page_size = page_size;
+
+ // Calculate the number of buckets needed for this filter
+ uint32_t num_buckets = 0;
+ s = CuckooFilter::OptimalNumBuckets(capacity, bucket_size, &num_buckets);
+ if (!s.ok()) return s;
+
+ INFO(
+ "Creating cuckoo filter with capacity={}, bucket_size={},
num_buckets={}, max_iterations={}, expansion={}, "
+ "page_size={}",
+ capacity, bucket_size, num_buckets, max_iterations,
static_cast<int>(expansion), page_size);
+
+ // Create a write batch for atomic operation
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.RESERVE", user_key.ToString()});
+ s = batch->PutLogData(log_data.Encode());
+ if (!s.ok()) return s;
+
+ std::string metadata_bytes;
+ metadata.Encode(&metadata_bytes);
+ s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes);
+ if (!s.ok()) return s;
+
+ // Pages are created lazily on first write. Reserve only persists metadata
so sparse filters don't preallocate page
+ // values that may never be used.
+
+ return storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+}
+
+static bool CalculateFilterCapacity(uint64_t base_capacity, uint16_t
expansion, uint16_t filter_index,
+ uint64_t *filter_capacity) {
+ uint64_t capacity = base_capacity;
+ for (uint16_t i = 0; i < filter_index; ++i) {
+ if (expansion != 0 && capacity > std::numeric_limits<uint64_t>::max() /
expansion) return false;
+ capacity *= expansion;
+ }
+ *filter_capacity = capacity;
+ return true;
+}
+
+rocksdb::Status CuckooChain::Add(engine::Context &ctx, const Slice &user_key,
const Slice &item, bool *added) {
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ CuckooChainMetadata metadata(false);
+ auto s = getCuckooChainMetadata(ctx, ns_key, &metadata);
+ if (s.IsNotFound()) {
+ // RedisBloom CF.ADD auto-creates the filter when the key does not exist:
+ // https://redis.io/docs/latest/commands/cf.add/
+ metadata = CuckooChainMetadata();
+ metadata.size = 0;
+ metadata.base_capacity = kCFDefaultCapacity;
+ metadata.bucket_size = kCFDefaultBucketSize;
+ metadata.max_iterations = kCFDefaultMaxIterations;
+ metadata.expansion = kCFDefaultExpansion;
+ metadata.n_filters = 1;
+ metadata.num_deleted_items = 0;
+ metadata.page_size = kCuckooFilterDefaultPageSize;
+ }
+ if (!s.ok() && !s.IsNotFound()) return s;
+
+ s = validateMetadata(metadata);
+ if (!s.ok()) return s;
+
+ // Calculate hash and fingerprint for the item
+ uint64_t hash = CuckooFilter::Hash(item.data(), item.size());
+ uint8_t fingerprint = CuckooFilter::GenerateFingerprint(hash);
+
+ // RedisBloom prioritizes the newest sub-filter to avoid repeatedly probing
older, fuller filters.
+ for (int filter_idx = static_cast<int>(metadata.n_filters) - 1; filter_idx
>= 0; --filter_idx) {
+ auto current_filter_idx = static_cast<uint16_t>(filter_idx);
+ uint64_t filter_capacity = 0;
+ if (!CalculateFilterCapacity(metadata.base_capacity, metadata.expansion,
current_filter_idx, &filter_capacity) ||
+ !CuckooFilter::IsCapacitySupported(filter_capacity,
metadata.bucket_size)) {
+ return rocksdb::Status::Corruption("invalid metadata: filter capacity is
too large");
+ }
+ uint32_t num_buckets = 0;
+ s = CuckooFilter::OptimalNumBuckets(filter_capacity, metadata.bucket_size,
&num_buckets);
+ if (!s.ok()) return s;
+
+ // 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;
+
+ CuckooPageSet pages(storage_, ctx, ns_key, metadata,
storage_->IsSlotIdEncoded());
+ bool inserted = false;
+ s = pages.TryInsertInCandidateBuckets(current_filter_idx, num_buckets,
bucket1_idx, bucket2_idx, fingerprint,
+ &inserted);
+ if (!s.ok()) return s;
+
+ if (inserted) {
+ // Successfully inserted, write to storage atomically
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisCuckooFilter,
std::vector<std::string>{"CF.ADD", user_key.ToString()});
Review Comment:
use `add` not "CF.ADD"
##########
src/types/cuckoo_filter.h:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 <rocksdb/status.h>
+
+#include <cstdint>
+#include <limits>
+#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.
+// Buckets are grouped into page values in RocksDB. The Cuckoo algorithm still
+// works with logical bucket indexes, while the storage layer maps buckets to
pages.
+//
+// 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:
+ static bool IsCapacitySupported(uint64_t capacity, uint8_t bucket_size) {
+ uint32_t num_buckets = 0;
+ return OptimalNumBuckets(capacity, bucket_size, &num_buckets).ok();
+ }
+
+ // Calculate the optimal number of buckets for the filter.
+ static rocksdb::Status OptimalNumBuckets(uint64_t capacity, uint8_t
bucket_size, uint32_t* num_buckets) {
+ if (bucket_size == 0) {
+ return rocksdb::Status::InvalidArgument("bucket_size must be larger than
0");
+ }
+
+ constexpr long double kLoadFactor = 0.955L;
+ constexpr uint64_t kMaxSupportedBuckets =
std::numeric_limits<uint32_t>::max() / 2 + 1ULL;
+ auto max_supported_capacity = static_cast<uint64_t>(kMaxSupportedBuckets *
bucket_size * kLoadFactor);
+ if (capacity > max_supported_capacity) {
+ return rocksdb::Status::InvalidArgument("capacity is too large");
+ }
+
+ auto exact_buckets = static_cast<long double>(capacity) / bucket_size /
kLoadFactor;
+ auto required_buckets = static_cast<uint64_t>(exact_buckets);
+ if (static_cast<long double>(required_buckets) < exact_buckets)
required_buckets++;
+ if (required_buckets == 0) required_buckets = 1;
+
+ // Round up to next power of 2 for better hash distribution.
+ uint32_t power = 1;
+ while (power < required_buckets) power <<= 1;
+ *num_buckets = power;
+ return rocksdb::Status::OK();
+ }
+
+ // Generate fingerprint from hash (8-bit fingerprint, non-zero, range: 1-255)
Review Comment:
This line is a redundant comment; the code is already clear.
##########
src/storage/redis_metadata.h:
##########
@@ -334,6 +335,50 @@ class BloomChainMetadata : public Metadata {
bool IsScaling() const { return expansion != 0; };
};
+constexpr uint32_t kCuckooFilterDefaultPageSize = 2048;
+
+class CuckooChainMetadata : public Metadata {
+ public:
+ /// The number of sub-filters in the chain
+ uint16_t n_filters;
+
+ /// Expansion factor for new filters
+ /// When a filter is full, a new one is created with capacity =
base_capacity * expansion^n
+ uint16_t expansion;
+
+ /// The capacity of the first filter
+ uint64_t base_capacity;
+
+ /// Number of fingerprints per bucket
+ uint8_t bucket_size;
+
+ /// Maximum number of cuckoo kicks before considering filter full
+ uint16_t max_iterations;
+
+ /// Track number of deleted items for maintenance
+ uint64_t num_deleted_items;
+
+ /// Target maximum payload size for each persisted Cuckoo Filter page
Review Comment:
We need number units.
##########
src/types/redis_cuckoo_chain.cc:
##########
@@ -0,0 +1,350 @@
+/*
+ * 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 <limits>
+
+#include "cuckoo_filter.h"
+#include "cuckoo_filter_page.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);
+}
+
+rocksdb::Status CuckooChain::validateMetadata(const CuckooChainMetadata
&metadata) {
+ if (metadata.n_filters == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: n_filters is 0");
+ }
+ if (metadata.base_capacity == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: base_capacity is 0");
+ }
+ if (metadata.bucket_size == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: bucket_size is 0");
+ }
+ if (metadata.max_iterations == 0) {
+ return rocksdb::Status::Corruption("invalid metadata: max_iterations is
0");
+ }
+ if (metadata.page_size < metadata.bucket_size) {
+ return rocksdb::Status::Corruption("invalid metadata: page_size is smaller
than bucket_size");
+ }
+ if (!CuckooFilter::IsCapacitySupported(metadata.base_capacity,
metadata.bucket_size)) {
+ return rocksdb::Status::Corruption("invalid metadata: base_capacity is too
large");
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice
&user_key, uint64_t capacity,
+ uint8_t bucket_size, uint16_t
max_iterations, uint16_t expansion,
+ uint32_t page_size) {
+ 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");
+ }
+ if (page_size == 0) {
+ return rocksdb::Status::InvalidArgument("page_size must be larger than 0");
+ }
+ if (page_size < bucket_size) {
+ return rocksdb::Status::InvalidArgument("page_size must be at least
bucket_size");
+ }
+ if (expansion > kCFMaxExpansion) {
+ return rocksdb::Status::InvalidArgument("expansion must be between 0 and
32768");
+ }
+ if (!CuckooFilter::IsCapacitySupported(capacity, bucket_size)) {
+ return rocksdb::Status::InvalidArgument("capacity is too large");
+ }
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+
+ CuckooChainMetadata existing_metadata;
+ auto s = getCuckooChainMetadata(ctx, ns_key, &existing_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("the key already exists");
+ }
+
+ CuckooChainMetadata metadata;
+
+ 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;
+ metadata.page_size = page_size;
+
+ // Calculate the number of buckets needed for this filter
+ uint32_t num_buckets = 0;
+ s = CuckooFilter::OptimalNumBuckets(capacity, bucket_size, &num_buckets);
+ if (!s.ok()) return s;
+
+ INFO(
+ "Creating cuckoo filter with capacity={}, bucket_size={},
num_buckets={}, max_iterations={}, expansion={}, "
+ "page_size={}",
+ capacity, bucket_size, num_buckets, max_iterations,
static_cast<int>(expansion), page_size);
Review Comment:
It seems we don't need the info
##########
src/types/redis_cuckoo_chain.h:
##########
@@ -0,0 +1,60 @@
+/*
+ * 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 uint16_t kCFDefaultExpansion = 2;
+const uint16_t kCFMaxExpansion = 32768;
+
+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, uint16_t expansion,
uint32_t page_size);
+
+ // CF.ADD command - adds an item to the cuckoo filter.
Review Comment:
CuckooChain should be a storage abstraction (rather than an instruction
execution abstraction), so its Reserve and Add functions need to be designed
around storage. The commands will have a thin orchestration layer to call
CuckooChain.
So, there's no need to point out the corresponding command names in the
comments.
##########
src/types/cuckoo_filter.h:
##########
@@ -0,0 +1,103 @@
+/*
+ * 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 <rocksdb/status.h>
+
+#include <cstdint>
+#include <limits>
+#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.
+// Buckets are grouped into page values in RocksDB. The Cuckoo algorithm still
+// works with logical bucket indexes, while the storage layer maps buckets to
pages.
+//
+// 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 {
Review Comment:
This class looks like a helper, but the current naming could easily mislead
us into thinking it's the Cuckoo Filter object itself.
--
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]
