jihuayu commented on code in PR #3481: URL: https://github.com/apache/kvrocks/pull/3481#discussion_r3296821779
########## src/types/redis_cuckoo_chain.cc: ########## @@ -0,0 +1,269 @@ +/* + * 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 "cuckoo_filter.h" +#include "cuckoo_filter_sub_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); +} + +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 (!CuckooFilterHelper::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 (!CuckooFilterHelper::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; + + // Create a write batch for atomic operation + auto batch = storage_->GetWriteBatchBase(); + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"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()); +} + +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 = CuckooFilterHelper::Hash(item.data(), item.size()); + uint8_t fingerprint = CuckooFilterHelper::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); + uint32_t num_buckets = 0; + s = CuckooFilterHelper::GetFilterNumBuckets(metadata.base_capacity, metadata.expansion, metadata.bucket_size, + current_filter_idx, &num_buckets); + if (!s.ok()) return s; + + CuckooSubFilter sub_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, + metadata.bucket_size, metadata.page_size, current_filter_idx, num_buckets); + bool inserted = false; + s = sub_filter.TryInsert(hash, fingerprint, &inserted); + if (!s.ok()) return s; + + if (inserted) { + auto batch = storage_->GetWriteBatchBase(); + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"add", user_key.ToString()}); + s = batch->PutLogData(log_data.Encode()); + if (!s.ok()) return s; + s = sub_filter.WriteToBatch(batch.Get()); + if (!s.ok()) return s; + + metadata.size++; + std::string metadata_bytes; + metadata.Encode(&metadata_bytes); + s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); + if (!s.ok()) return s; + + 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; + uint32_t num_buckets = 0; + s = CuckooFilterHelper::GetFilterNumBuckets(metadata.base_capacity, metadata.expansion, metadata.bucket_size, + last_filter_idx, &num_buckets); + if (!s.ok()) return s; + + bool inserted = false; + auto batch = storage_->GetWriteBatchBase(); + CuckooSubFilter last_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, + metadata.bucket_size, metadata.page_size, last_filter_idx, num_buckets); + s = last_filter.KickOutInsert(hash, fingerprint, metadata.max_iterations, &inserted); + if (s.ok() && inserted) { + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"add", user_key.ToString()}); + s = batch->PutLogData(log_data.Encode()); + if (!s.ok()) return s; + s = last_filter.WriteToBatch(batch.Get()); + if (!s.ok()) return s; + + metadata.size++; + std::string metadata_bytes; + metadata.Encode(&metadata_bytes); + s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); + if (!s.ok()) return s; + + s = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); + if (!s.ok()) return s; + + *added = true; + return rocksdb::Status::OK(); + } + + // Kick-out failed, try to expand if allowed + if (metadata.expansion > 0) { + if (metadata.n_filters >= UINT16_MAX) return rocksdb::Status::Aborted("maximum number of filters reached"); + + metadata.n_filters++; + INFO("add expanded to {} filters", metadata.n_filters); Review Comment: We don't need INFO ########## src/types/redis_cuckoo_chain.cc: ########## @@ -0,0 +1,269 @@ +/* + * 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 "cuckoo_filter.h" +#include "cuckoo_filter_sub_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); +} + +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 (!CuckooFilterHelper::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 (!CuckooFilterHelper::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; + + // Create a write batch for atomic operation + auto batch = storage_->GetWriteBatchBase(); + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"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()); +} + +rocksdb::Status CuckooChain::Add(engine::Context &ctx, const Slice &user_key, const Slice &item, bool *added) { Review Comment: I feel this function is too long. Could we encapsulate the standard Cuckoo Filter steps into private helper functions? For example: `tryCuckooInsert`, `tryCuckooKickOut`, and `expandCuckooChain`. ########## src/types/redis_cuckoo_chain.cc: ########## @@ -0,0 +1,269 @@ +/* + * 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 "cuckoo_filter.h" +#include "cuckoo_filter_sub_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); +} + +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 (!CuckooFilterHelper::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 (!CuckooFilterHelper::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; + + // Create a write batch for atomic operation + auto batch = storage_->GetWriteBatchBase(); + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"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()); +} + +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 = CuckooFilterHelper::Hash(item.data(), item.size()); + uint8_t fingerprint = CuckooFilterHelper::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); + uint32_t num_buckets = 0; + s = CuckooFilterHelper::GetFilterNumBuckets(metadata.base_capacity, metadata.expansion, metadata.bucket_size, + current_filter_idx, &num_buckets); + if (!s.ok()) return s; + + CuckooSubFilter sub_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, + metadata.bucket_size, metadata.page_size, current_filter_idx, num_buckets); + bool inserted = false; + s = sub_filter.TryInsert(hash, fingerprint, &inserted); + if (!s.ok()) return s; + + if (inserted) { + auto batch = storage_->GetWriteBatchBase(); + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"add", user_key.ToString()}); + s = batch->PutLogData(log_data.Encode()); + if (!s.ok()) return s; + s = sub_filter.WriteToBatch(batch.Get()); + if (!s.ok()) return s; + + metadata.size++; + std::string metadata_bytes; + metadata.Encode(&metadata_bytes); + s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); + if (!s.ok()) return s; + + 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; + uint32_t num_buckets = 0; + s = CuckooFilterHelper::GetFilterNumBuckets(metadata.base_capacity, metadata.expansion, metadata.bucket_size, + last_filter_idx, &num_buckets); + if (!s.ok()) return s; + + bool inserted = false; + auto batch = storage_->GetWriteBatchBase(); + CuckooSubFilter last_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, + metadata.bucket_size, metadata.page_size, last_filter_idx, num_buckets); + s = last_filter.KickOutInsert(hash, fingerprint, metadata.max_iterations, &inserted); + if (s.ok() && inserted) { + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"add", user_key.ToString()}); + s = batch->PutLogData(log_data.Encode()); + if (!s.ok()) return s; + s = last_filter.WriteToBatch(batch.Get()); + if (!s.ok()) return s; + + metadata.size++; + std::string metadata_bytes; + metadata.Encode(&metadata_bytes); + s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); + if (!s.ok()) return s; + + s = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); + if (!s.ok()) return s; + + *added = true; + return rocksdb::Status::OK(); + } + + // Kick-out failed, try to expand if allowed + if (metadata.expansion > 0) { + if (metadata.n_filters >= UINT16_MAX) return rocksdb::Status::Aborted("maximum number of filters reached"); + + metadata.n_filters++; Review Comment: Could you move it to after the code that successfully creates the sub-filter? That would make the code clearer. ########## src/types/cuckoo_filter_sub_filter.cc: ########## @@ -0,0 +1,100 @@ +/* + * 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 "cuckoo_filter_sub_filter.h" + +#include "cuckoo_filter.h" + +namespace redis { + +CuckooSubFilter::CuckooSubFilter(engine::Storage *storage, engine::Context &ctx, const Slice &ns_key, + bool slot_id_encoded, uint64_t version, uint8_t bucket_size, uint32_t page_size, + uint16_t filter_index, uint32_t num_buckets) + : bucket_size_(bucket_size), + filter_index_(filter_index), + num_buckets_(num_buckets), + pages_(storage, ctx, ns_key, slot_id_encoded, version, bucket_size, page_size) {} + +rocksdb::Status CuckooSubFilter::TryInsert(uint64_t hash, uint8_t fingerprint, bool *inserted) { + *inserted = false; + uint32_t bucket1_idx = getPrimaryBucketIndex(hash); + uint32_t bucket2_idx = getSecondaryBucketIndex(hash, fingerprint); + auto s = pages_.PrefetchBuckets(filter_index_, num_buckets_, bucket1_idx, bucket2_idx); + if (!s.ok()) return s; + + s = pages_.TryInsertInBucket(filter_index_, num_buckets_, bucket1_idx, fingerprint, inserted); + if (!s.ok() || *inserted || bucket1_idx == bucket2_idx) return s; + + return pages_.TryInsertInBucket(filter_index_, num_buckets_, bucket2_idx, fingerprint, inserted); +} + +rocksdb::Status CuckooSubFilter::TryInsertPrimaryBucket(uint64_t hash, uint8_t fingerprint, bool *inserted) { + return pages_.TryInsertInBucket(filter_index_, num_buckets_, getPrimaryBucketIndex(hash), fingerprint, inserted); +} + +rocksdb::Status CuckooSubFilter::KickOutInsert(uint64_t hash, uint8_t fingerprint, uint16_t max_iterations, + bool *inserted) { + *inserted = false; + + uint32_t current_bucket_idx = getPrimaryBucketIndex(hash); + uint8_t current_fp = fingerprint; + uint32_t victim_slot = 0; + + for (uint16_t iteration = 0; iteration < max_iterations; ++iteration) { Review Comment: This code pollutes dirty pages when the kick operation fails, which is inconsistent with the function's semantics. We can use the way RedisBloom do: if the kick fails, it swaps the previously swapped slots back in reverse order. ########## src/types/cuckoo_filter_page.cc: ########## @@ -0,0 +1,231 @@ +/* + * 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 "cuckoo_filter_page.h" + +#include <algorithm> + +#include "common/encoding.h" +#include "storage/redis_db.h" + +namespace redis { + +namespace { + +uint32_t GetBucketsPerPage(uint32_t page_size, uint8_t bucket_size) { + return std::max<uint32_t>(1, page_size / bucket_size); +} + +uint32_t GetPageIndex(uint32_t bucket_index, uint32_t buckets_per_page) { return bucket_index / buckets_per_page; } + +uint32_t GetBucketOffset(uint32_t bucket_index, uint32_t buckets_per_page, uint8_t bucket_size) { + return (bucket_index % buckets_per_page) * bucket_size; +} + +uint32_t GetPageValueSize(uint32_t page_index, uint32_t num_buckets, uint32_t buckets_per_page, uint8_t bucket_size) { Review Comment: What does `PageValueSize` mean? I'm not quite sure what this name means. Can you change a better name? ########## src/types/cuckoo_filter_page.cc: ########## @@ -0,0 +1,231 @@ +/* + * 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 "cuckoo_filter_page.h" + +#include <algorithm> + +#include "common/encoding.h" +#include "storage/redis_db.h" + +namespace redis { + +namespace { + +uint32_t GetBucketsPerPage(uint32_t page_size, uint8_t bucket_size) { + return std::max<uint32_t>(1, page_size / bucket_size); +} + +uint32_t GetPageIndex(uint32_t bucket_index, uint32_t buckets_per_page) { return bucket_index / buckets_per_page; } + +uint32_t GetBucketOffset(uint32_t bucket_index, uint32_t buckets_per_page, uint8_t bucket_size) { + return (bucket_index % buckets_per_page) * bucket_size; +} + +uint32_t GetPageValueSize(uint32_t page_index, uint32_t num_buckets, uint32_t buckets_per_page, uint8_t bucket_size) { + uint32_t first_bucket = page_index * buckets_per_page; + uint32_t page_bucket_count = std::min(buckets_per_page, num_buckets - first_bucket); + return page_bucket_count * bucket_size; +} + +std::string GetCuckooPageKey(const Slice &ns_key, uint64_t version, bool slot_id_encoded, uint16_t filter_index, + uint32_t page_index) { + std::string sub_key; + PutFixed16(&sub_key, filter_index); + PutFixed32(&sub_key, page_index); + return InternalKey(ns_key, sub_key, version, slot_id_encoded).Encode(); +} + +} // namespace + +CuckooPageCache::CuckooPageCache(engine::Storage *storage, engine::Context &ctx, const Slice &ns_key, + bool slot_id_encoded, uint64_t version, uint8_t bucket_size, uint32_t page_size) + : storage_(storage), + ctx_(ctx), + ns_key_(ns_key.ToString()), + slot_id_encoded_(slot_id_encoded), + version_(version), + bucket_size_(bucket_size), + page_size_(page_size) {} + +rocksdb::Status CuckooPageCache::PrefetchBuckets(uint16_t filter_index, uint32_t num_buckets, uint32_t bucket1_index, + uint32_t bucket2_index) { + BucketLocation location1, location2; + auto s = resolveBucketLocation(filter_index, num_buckets, bucket1_index, &location1); + if (!s.ok()) return s; + s = resolveBucketLocation(filter_index, num_buckets, bucket2_index, &location2); + if (!s.ok()) return s; + + std::vector<BucketLocation> missing_locations; + if (pages_.find(location1.page_key) == pages_.end()) missing_locations.push_back(location1); + if (location2.page_key != location1.page_key && pages_.find(location2.page_key) == pages_.end()) { + missing_locations.push_back(location2); + } + return loadPages(missing_locations); +} + +rocksdb::Status CuckooPageCache::TryInsertInBucket(uint16_t filter_index, uint32_t num_buckets, uint32_t bucket_index, + uint8_t fingerprint, bool *inserted) { + *inserted = false; + BucketRef bucket; + auto s = ensureBucketLoaded(filter_index, num_buckets, bucket_index, &bucket); + if (!s.ok()) return s; + + size_t slot_idx = 0; + *inserted = tryInsertInBucketRef(bucket, fingerprint, &slot_idx); + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::GetBucketSlot(uint16_t filter_index, uint32_t num_buckets, uint32_t bucket_index, + uint32_t slot_idx, uint8_t *fingerprint) { + if (slot_idx >= bucket_size_) return rocksdb::Status::InvalidArgument("invalid cuckoo filter bucket slot"); + + BucketRef bucket; + auto s = ensureBucketLoaded(filter_index, num_buckets, bucket_index, &bucket); + if (!s.ok()) return s; + *fingerprint = getBucketRefSlot(bucket, slot_idx); + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::SetBucketSlot(uint16_t filter_index, uint32_t num_buckets, uint32_t bucket_index, + uint32_t slot_idx, uint8_t fingerprint) { + if (slot_idx >= bucket_size_) return rocksdb::Status::InvalidArgument("invalid cuckoo filter bucket slot"); + + BucketRef bucket; + auto s = ensureBucketLoaded(filter_index, num_buckets, bucket_index, &bucket); + if (!s.ok()) return s; + setBucketRefSlot(bucket, slot_idx, fingerprint); + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::WriteBackDirtyPages(rocksdb::WriteBatchBase *batch) { + for (const auto &entry : pages_) { + if (!entry.second.is_dirty) continue; + auto s = batch->Put(entry.first, entry.second.data); + if (!s.ok()) return s; + } + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::resolveBucketLocation(uint16_t filter_index, uint32_t num_buckets, + uint32_t bucket_index, BucketLocation *location) const { + if (bucket_size_ == 0 || num_buckets == 0 || bucket_index >= num_buckets) { + return rocksdb::Status::Corruption("invalid cuckoo filter bucket location"); + } + + uint32_t buckets_per_page = GetBucketsPerPage(page_size_, bucket_size_); + uint32_t page_index = GetPageIndex(bucket_index, buckets_per_page); + location->page_key = GetCuckooPageKey(ns_key_, version_, slot_id_encoded_, filter_index, page_index); + location->offset = GetBucketOffset(bucket_index, buckets_per_page, bucket_size_); + location->expected_page_size = GetPageValueSize(page_index, num_buckets, buckets_per_page, bucket_size_); + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::ensureBucketLoaded(uint16_t filter_index, uint32_t num_buckets, uint32_t bucket_index, + BucketRef *bucket) { + BucketLocation location; + auto s = resolveBucketLocation(filter_index, num_buckets, bucket_index, &location); + if (!s.ok()) return s; + + PageEntry *page = nullptr; + s = loadPage(location, &page); + if (!s.ok()) return s; + + bucket->page = page; + bucket->offset = location.offset; + bucket->size = bucket_size_; + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::loadPage(const BucketLocation &location, PageEntry **page) { + auto iter = pages_.find(location.page_key); + if (iter != pages_.end()) { + *page = &iter->second; + return rocksdb::Status::OK(); + } + + PageEntry page_entry; + auto s = storage_->Get(ctx_, ctx_.GetReadOptions(), location.page_key, &page_entry.data); + if (!s.ok() && !s.IsNotFound()) return s; + s = normalizePage(s, location.expected_page_size, &page_entry); + if (!s.ok()) return s; + + auto result = pages_.emplace(location.page_key, std::move(page_entry)); + *page = &result.first->second; + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::loadPages(const std::vector<BucketLocation> &locations) { + if (locations.empty()) return rocksdb::Status::OK(); + if (locations.size() == 1) { + PageEntry *page = nullptr; + return loadPage(locations[0], &page); + } + + std::vector<rocksdb::Slice> keys; + keys.reserve(locations.size()); + for (const auto &location : locations) keys.emplace_back(location.page_key); + + std::vector<rocksdb::PinnableSlice> values(locations.size()); + std::vector<rocksdb::Status> statuses(locations.size()); + storage_->MultiGet(ctx_, ctx_.DefaultMultiGetOptions(), storage_->GetDB()->DefaultColumnFamily(), keys.size(), + keys.data(), values.data(), statuses.data()); + + for (size_t i = 0; i < locations.size(); ++i) { + PageEntry page_entry; + if (statuses[i].ok()) page_entry.data.assign(values[i].data(), values[i].size()); + auto s = normalizePage(statuses[i], locations[i].expected_page_size, &page_entry); + if (!s.ok()) return s; + pages_.emplace(locations[i].page_key, std::move(page_entry)); + } + return rocksdb::Status::OK(); +} + +rocksdb::Status CuckooPageCache::normalizePage(const rocksdb::Status &status, uint32_t expected_size, PageEntry *page) { + if (!status.ok() && !status.IsNotFound()) return status; + if (status.IsNotFound()) page->data.clear(); + if (page->data.size() > expected_size) return rocksdb::Status::Corruption("invalid cuckoo filter page size"); + if (page->data.size() < expected_size) page->data.resize(expected_size, 0); Review Comment: I think we need to distinguish whether this page has already been created. If the page already exists, this should be treated as page corruption rather than automatically filling it with zeros. ########## src/types/redis_cuckoo_chain.cc: ########## @@ -0,0 +1,269 @@ +/* + * 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 "cuckoo_filter.h" +#include "cuckoo_filter_sub_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); +} + +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 (!CuckooFilterHelper::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 (!CuckooFilterHelper::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; + + // Create a write batch for atomic operation + auto batch = storage_->GetWriteBatchBase(); + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"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()); +} + +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 = CuckooFilterHelper::Hash(item.data(), item.size()); + uint8_t fingerprint = CuckooFilterHelper::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); + uint32_t num_buckets = 0; + s = CuckooFilterHelper::GetFilterNumBuckets(metadata.base_capacity, metadata.expansion, metadata.bucket_size, + current_filter_idx, &num_buckets); + if (!s.ok()) return s; + + CuckooSubFilter sub_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, + metadata.bucket_size, metadata.page_size, current_filter_idx, num_buckets); + bool inserted = false; + s = sub_filter.TryInsert(hash, fingerprint, &inserted); + if (!s.ok()) return s; + + if (inserted) { + auto batch = storage_->GetWriteBatchBase(); + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"add", user_key.ToString()}); + s = batch->PutLogData(log_data.Encode()); + if (!s.ok()) return s; + s = sub_filter.WriteToBatch(batch.Get()); + if (!s.ok()) return s; + + metadata.size++; + std::string metadata_bytes; + metadata.Encode(&metadata_bytes); + s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); + if (!s.ok()) return s; + + 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; + uint32_t num_buckets = 0; + s = CuckooFilterHelper::GetFilterNumBuckets(metadata.base_capacity, metadata.expansion, metadata.bucket_size, + last_filter_idx, &num_buckets); + if (!s.ok()) return s; + + bool inserted = false; + auto batch = storage_->GetWriteBatchBase(); + CuckooSubFilter last_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, + metadata.bucket_size, metadata.page_size, last_filter_idx, num_buckets); + s = last_filter.KickOutInsert(hash, fingerprint, metadata.max_iterations, &inserted); + if (s.ok() && inserted) { + WriteBatchLogData log_data(kRedisCuckooFilter, std::vector<std::string>{"add", user_key.ToString()}); + s = batch->PutLogData(log_data.Encode()); + if (!s.ok()) return s; + s = last_filter.WriteToBatch(batch.Get()); + if (!s.ok()) return s; + + metadata.size++; + std::string metadata_bytes; + metadata.Encode(&metadata_bytes); + s = batch->Put(metadata_cf_handle_, ns_key, metadata_bytes); + if (!s.ok()) return s; + + s = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); + if (!s.ok()) return s; + + *added = true; + return rocksdb::Status::OK(); + } + + // Kick-out failed, try to expand if allowed + if (metadata.expansion > 0) { + if (metadata.n_filters >= UINT16_MAX) return rocksdb::Status::Aborted("maximum number of filters reached"); + + metadata.n_filters++; + INFO("add expanded to {} filters", metadata.n_filters); + + // Retry insertion in the new expanded filter + uint16_t new_filter_idx = metadata.n_filters - 1; + uint32_t new_num_buckets = 0; + s = CuckooFilterHelper::GetFilterNumBuckets(metadata.base_capacity, metadata.expansion, metadata.bucket_size, + new_filter_idx, &new_num_buckets); + if (s.IsCorruption()) { + return rocksdb::Status::Aborted("maximum filter capacity reached"); + } + if (!s.ok()) return s; + + CuckooSubFilter new_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, + metadata.bucket_size, metadata.page_size, new_filter_idx, new_num_buckets); + s = new_filter.TryInsertPrimaryBucket(hash, fingerprint, &inserted); Review Comment: Although the new sub-filter’s TryInsertPrimaryBucket can guarantee correctness, it feels semantically confusing. I suggest adding the necessary comments or using a normal insert instead. ########## src/types/cuckoo_filter_sub_filter.cc: ########## @@ -0,0 +1,100 @@ +/* + * 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 "cuckoo_filter_sub_filter.h" + +#include "cuckoo_filter.h" + +namespace redis { + +CuckooSubFilter::CuckooSubFilter(engine::Storage *storage, engine::Context &ctx, const Slice &ns_key, + bool slot_id_encoded, uint64_t version, uint8_t bucket_size, uint32_t page_size, + uint16_t filter_index, uint32_t num_buckets) + : bucket_size_(bucket_size), + filter_index_(filter_index), + num_buckets_(num_buckets), + pages_(storage, ctx, ns_key, slot_id_encoded, version, bucket_size, page_size) {} + +rocksdb::Status CuckooSubFilter::TryInsert(uint64_t hash, uint8_t fingerprint, bool *inserted) { + *inserted = false; + uint32_t bucket1_idx = getPrimaryBucketIndex(hash); + uint32_t bucket2_idx = getSecondaryBucketIndex(hash, fingerprint); + auto s = pages_.PrefetchBuckets(filter_index_, num_buckets_, bucket1_idx, bucket2_idx); + if (!s.ok()) return s; + + s = pages_.TryInsertInBucket(filter_index_, num_buckets_, bucket1_idx, fingerprint, inserted); + if (!s.ok() || *inserted || bucket1_idx == bucket2_idx) return s; + + return pages_.TryInsertInBucket(filter_index_, num_buckets_, bucket2_idx, fingerprint, inserted); +} + +rocksdb::Status CuckooSubFilter::TryInsertPrimaryBucket(uint64_t hash, uint8_t fingerprint, bool *inserted) { + return pages_.TryInsertInBucket(filter_index_, num_buckets_, getPrimaryBucketIndex(hash), fingerprint, inserted); +} + +rocksdb::Status CuckooSubFilter::KickOutInsert(uint64_t hash, uint8_t fingerprint, uint16_t max_iterations, Review Comment: I think `TryKickOutInsert` is better. It may insert failed. -- 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]
