diff --git a/src/commands/cmd_cuckoo_filter.cc b/src/commands/cmd_cuckoo_filter.cc index 2002912f94d..ab40e8c8a91 100644 --- a/src/commands/cmd_cuckoo_filter.cc +++ b/src/commands/cmd_cuckoo_filter.cc @@ -18,10 +18,14 @@ * */ +#include +#include + #include "command_parser.h" #include "commander.h" #include "error_constants.h" #include "server/server.h" +#include "types/redis_bloom_chain.h" #include "types/redis_cuckoo_chain.h" namespace redis { @@ -118,21 +122,109 @@ class CommandCFAdd : public Commander { Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override { redis::CuckooChain cuckoo_db(srv->storage, conn->GetNamespace()); - bool added = false; - auto s = cuckoo_db.Add(ctx, args_[1], args_[2], &added); + redis::CuckooFilterInsertResult ret = CuckooFilterInsertResult::kOk; + auto s = cuckoo_db.Add(ctx, args_[1], args_[2], ret); if (!s.ok()) { return {Status::RedisExecErr, s.ToString()}; } // Duplicate items are allowed, so successful insertions return 1. - *output = redis::Integer(added ? 1 : 0); + switch (ret) { + case CuckooFilterInsertResult::kOk: + *output = redis::Integer(1); + break; + case CuckooFilterInsertResult::kExist: + return {Status::RedisExecErr, "unexpected cuckoo filter insert result"}; + case CuckooFilterInsertResult::kFull: + *output = redis::Error({Status::NotOK, "filter is full"}); + break; + } + return Status::OK(); + } +}; + +class CommandCFInsert : public Commander { + public: + Status Parse(const std::vector &args) override { + // CF.INSERT key [CAPACITY capacity] [NOCREATE] ITEMS item [item ...] + if (args.size() < 4) { + return {Status::RedisParseErr, errWrongNumOfArguments}; + } + + CommandParser parser(args, 2); + while (parser.Good()) { + if (parser.EatEqICase("CAPACITY")) { + auto parse_capacity = parser.TakeInt(); + if (!parse_capacity.IsOK()) { + return {Status::RedisParseErr, "invalid capacity"}; + } + insert_options_.capacity = parse_capacity.GetValue(); + if (insert_options_.capacity <= 0) { + return {Status::RedisParseErr, "capacity must be larger than 0"}; + } + } else if (parser.EatEqICase("NOCREATE")) { + insert_options_.auto_create = false; + } else if (parser.EatEqICase("ITEMS")) { + has_items_ = true; + break; + } else { + return {Status::RedisParseErr, errInvalidSyntax}; + } + } + + if (!has_items_) { + return {Status::RedisParseErr, errInvalidSyntax}; + } + + while (parser.Good()) { + items_.emplace_back(GET_OR_RET(parser.TakeStr())); + } + + if (items_.empty()) { + return {Status::RedisParseErr, "num of items should be greater than 0"}; + } + + return Commander::Parse(args); + } + + Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override { + redis::CuckooChain cuckoo_db(srv->storage, conn->GetNamespace()); + std::vector rets(items_.size(), CuckooFilterInsertResult::kOk); + + auto s = cuckoo_db.Insert(ctx, args_[1], items_, insert_options_, rets); + + if (!s.ok()) { + return {Status::RedisExecErr, s.ToString()}; + } + + *output = redis::MultiLen(items_.size()); + for (const auto &ret : rets) { + switch (ret) { + case CuckooFilterInsertResult::kOk: + *output += redis::Integer(1); + break; + case CuckooFilterInsertResult::kExist: + std::cout << "Item exists result in add command is not possible" << std::endl; + std::abort(); + case CuckooFilterInsertResult::kFull: + *output += redis::Integer(-1); + break; + } + } + return Status::OK(); } + + private: + CuckooFilterInsertOptions insert_options_; + bool has_items_ = false; + std::vector items_; }; // Register the CF.RESERVE and CF.ADD commands REDIS_REGISTER_COMMANDS(CuckooFilter, MakeCmdAttr("cf.reserve", -3, "write", 1, 1, 1), - MakeCmdAttr("cf.add", 3, "write", 1, 1, 1)) + MakeCmdAttr("cf.add", 3, "write", 1, 1, 1), + MakeCmdAttr("cf.insert", -4, "write", 1, 1, 1)) } // namespace redis diff --git a/src/types/cuckoo_filter_page.h b/src/types/cuckoo_filter_page.h index ff2efeb50c7..a1938f9240d 100644 --- a/src/types/cuckoo_filter_page.h +++ b/src/types/cuckoo_filter_page.h @@ -49,6 +49,8 @@ class CuckooPageCache { void DiscardCachedPages(); + uint8_t GetBucketSize() const { return bucket_size_; } + private: struct PageEntry { std::string data; diff --git a/src/types/cuckoo_filter_sub_filter.cc b/src/types/cuckoo_filter_sub_filter.cc index 0a0748cc354..185d21fd2e8 100644 --- a/src/types/cuckoo_filter_sub_filter.cc +++ b/src/types/cuckoo_filter_sub_filter.cc @@ -21,28 +21,24 @@ #include "cuckoo_filter_sub_filter.h" #include "cuckoo_filter.h" +#include "types/cuckoo_filter_page.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) {} +CuckooSubFilter::CuckooSubFilter(CuckooPageCache *pages, uint16_t filter_index, uint32_t num_buckets) + : bucket_size_(pages->GetBucketSize()), filter_index_(filter_index), num_buckets_(num_buckets), pages_(pages) {} 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); + 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); + 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); + return pages_->TryInsertInBucket(filter_index_, num_buckets_, bucket2_idx, fingerprint, inserted); } rocksdb::Status CuckooSubFilter::TryKickOutInsert(uint64_t hash, uint8_t fingerprint, uint16_t max_iterations, @@ -55,14 +51,14 @@ rocksdb::Status CuckooSubFilter::TryKickOutInsert(uint64_t hash, uint8_t fingerp for (uint16_t iteration = 0; iteration < max_iterations; ++iteration) { uint8_t old_fp = 0; - auto s = pages_.GetBucketSlot(filter_index_, num_buckets_, current_bucket_idx, victim_slot, &old_fp); + auto s = pages_->GetBucketSlot(filter_index_, num_buckets_, current_bucket_idx, victim_slot, &old_fp); if (!s.ok()) { - pages_.DiscardCachedPages(); + pages_->DiscardCachedPages(); return s; } - s = pages_.SetBucketSlot(filter_index_, num_buckets_, current_bucket_idx, victim_slot, current_fp); + s = pages_->SetBucketSlot(filter_index_, num_buckets_, current_bucket_idx, victim_slot, current_fp); if (!s.ok()) { - pages_.DiscardCachedPages(); + pages_->DiscardCachedPages(); return s; } current_fp = old_fp; @@ -75,9 +71,9 @@ rocksdb::Status CuckooSubFilter::TryKickOutInsert(uint64_t hash, uint8_t fingerp uint32_t alt_bucket_idx = CuckooFilterHelper::GetAltBucketIndex(current_bucket_idx, current_fp, num_buckets_); bool inserted_in_alt_bucket = false; - s = pages_.TryInsertInBucket(filter_index_, num_buckets_, alt_bucket_idx, current_fp, &inserted_in_alt_bucket); + s = pages_->TryInsertInBucket(filter_index_, num_buckets_, alt_bucket_idx, current_fp, &inserted_in_alt_bucket); if (!s.ok()) { - pages_.DiscardCachedPages(); + pages_->DiscardCachedPages(); return s; } if (inserted_in_alt_bucket) { @@ -89,12 +85,12 @@ rocksdb::Status CuckooSubFilter::TryKickOutInsert(uint64_t hash, uint8_t fingerp victim_slot = (victim_slot + 1) % bucket_size_; } - pages_.DiscardCachedPages(); + pages_->DiscardCachedPages(); return rocksdb::Status::OK(); } rocksdb::Status CuckooSubFilter::WriteToBatch(rocksdb::WriteBatchBase *batch) { - return pages_.WriteBackDirtyPages(batch); + return pages_->WriteBackDirtyPages(batch); } uint32_t CuckooSubFilter::getPrimaryBucketIndex(uint64_t hash) const { return hash % num_buckets_; } diff --git a/src/types/cuckoo_filter_sub_filter.h b/src/types/cuckoo_filter_sub_filter.h index 2bd26df8541..84fb39d1b83 100644 --- a/src/types/cuckoo_filter_sub_filter.h +++ b/src/types/cuckoo_filter_sub_filter.h @@ -31,9 +31,7 @@ namespace redis { class CuckooSubFilter { public: - 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); + CuckooSubFilter(CuckooPageCache *pages, uint16_t filter_index, uint32_t num_buckets); uint16_t Index() const { return filter_index_; } uint32_t NumBuckets() const { return num_buckets_; } @@ -51,7 +49,7 @@ class CuckooSubFilter { uint8_t bucket_size_ = 0; uint16_t filter_index_ = 0; uint32_t num_buckets_ = 0; - CuckooPageCache pages_; + CuckooPageCache *pages_; }; } // namespace redis diff --git a/src/types/redis_cuckoo_chain.cc b/src/types/redis_cuckoo_chain.cc index f7074ce9458..81214e749b3 100644 --- a/src/types/redis_cuckoo_chain.cc +++ b/src/types/redis_cuckoo_chain.cc @@ -22,7 +22,7 @@ #include "cuckoo_filter.h" #include "cuckoo_filter_sub_filter.h" -#include "logging.h" +#include "types/cuckoo_filter_page.h" namespace redis { @@ -47,9 +47,6 @@ rocksdb::Status CuckooChain::validateMetadata(const CuckooChainMetadata &metadat 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(); } @@ -124,17 +121,41 @@ rocksdb::Status CuckooChain::Reserve(engine::Context &ctx, const Slice &user_key return storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); } -rocksdb::Status CuckooChain::Add(engine::Context &ctx, const Slice &user_key, const Slice &item, bool *added) { +rocksdb::Status CuckooChain::Add(engine::Context &ctx, const Slice &user_key, const std::string &item, + CuckooFilterInsertResult &ret) { + std::vector tmp{CuckooFilterInsertResult::kOk}; + CuckooFilterInsertOptions options; + // RedisBloom CF.ADD auto-creates the filter when the key does not exist: + // https://redis.io/docs/latest/commands/cf.add/ + auto s = Insert(ctx, user_key, {item}, options, tmp); + ret = tmp[0]; + return s; +} + +rocksdb::Status CuckooChain::Insert(engine::Context &ctx, const Slice &user_key, const std::vector &items, + CuckooFilterInsertOptions &insert_options, + std::vector &ret) { 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/ + if (!insert_options.auto_create) { + return s; + } + + if (insert_options.capacity < 2) { + return rocksdb::Status::InvalidArgument("capacity must be at least 2"); + } + + if (!CuckooFilterHelper::IsCapacitySupported(insert_options.capacity, kCFDefaultBucketSize)) { + return rocksdb::Status::InvalidArgument("capacity is too large"); + } + metadata = CuckooChainMetadata(); metadata.size = 0; - metadata.base_capacity = kCFDefaultCapacity; + metadata.base_capacity = insert_options.capacity; metadata.bucket_size = kCFDefaultBucketSize; metadata.max_iterations = kCFDefaultMaxIterations; metadata.expansion = kCFDefaultExpansion; @@ -148,40 +169,54 @@ rocksdb::Status CuckooChain::Add(engine::Context &ctx, const Slice &user_key, co 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); + CuckooPageCache pages(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata.version, metadata.bucket_size, + metadata.page_size); + bool inserted_something = false; - bool inserted = false; - s = tryCuckooInsert(ctx, user_key, ns_key, &metadata, hash, fingerprint, &inserted); - if (!s.ok()) return s; - if (inserted) { - *added = true; - return rocksdb::Status::OK(); - } + for (size_t i = 0; i < items.size(); ++i) { + const auto &item = items[i]; + uint64_t hash = CuckooFilterHelper::Hash(item.data(), item.size()); + uint8_t fingerprint = CuckooFilterHelper::GenerateFingerprint(hash); - s = tryCuckooKickOut(ctx, user_key, ns_key, &metadata, hash, fingerprint, &inserted); - if (!s.ok()) return s; - if (inserted) { - *added = true; - return rocksdb::Status::OK(); + bool inserted = false; + s = tryCuckooInsert(&metadata, &pages, hash, fingerprint, &inserted); + if (!s.ok()) return s; + if (inserted) { + inserted_something = true; + ret[i] = CuckooFilterInsertResult::kOk; + continue; + } + + s = tryCuckooKickOut(&metadata, &pages, hash, fingerprint, &inserted); + if (!s.ok()) return s; + if (inserted) { + inserted_something = true; + ret[i] = CuckooFilterInsertResult::kOk; + continue; + } + + s = expandAndInsertCuckooChain(&metadata, &pages, hash, fingerprint, &inserted); + if (!s.ok()) return s; + if (inserted) { + inserted_something = true; + ret[i] = CuckooFilterInsertResult::kOk; + continue; + } + + // No expansion allowed and filter is full + ret[i] = CuckooFilterInsertResult::kFull; } - s = expandAndInsertCuckooChain(ctx, user_key, ns_key, &metadata, hash, fingerprint, &inserted); - if (!s.ok()) return s; - if (inserted) { - *added = true; - return rocksdb::Status::OK(); + if (inserted_something) { + s = commitPagesAndMetadata(ctx, user_key, ns_key, &metadata, &pages); + if (!s.ok()) return s; } - // No expansion allowed and filter is full - *added = false; - return rocksdb::Status::Aborted("filter is full"); + return rocksdb::Status::OK(); } -rocksdb::Status CuckooChain::tryCuckooInsert(engine::Context &ctx, const Slice &user_key, const std::string &ns_key, - CuckooChainMetadata *metadata, uint64_t hash, uint8_t fingerprint, - bool *inserted) { +rocksdb::Status CuckooChain::tryCuckooInsert(CuckooChainMetadata *metadata, CuckooPageCache *pages, uint64_t hash, + uint8_t fingerprint, bool *inserted) { *inserted = false; // RedisBloom prioritizes the newest sub-filter to avoid repeatedly probing older, fuller filters. @@ -192,15 +227,13 @@ rocksdb::Status CuckooChain::tryCuckooInsert(engine::Context &ctx, const Slice & 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); + CuckooSubFilter sub_filter(pages, current_filter_idx, num_buckets); bool current_inserted = false; s = sub_filter.TryInsert(hash, fingerprint, ¤t_inserted); if (!s.ok()) return s; if (current_inserted) { - s = commitSubFilterAndMetadata(ctx, user_key, ns_key, metadata, &sub_filter); - if (!s.ok()) return s; + metadata->size++; *inserted = true; return rocksdb::Status::OK(); } @@ -209,9 +242,8 @@ rocksdb::Status CuckooChain::tryCuckooInsert(engine::Context &ctx, const Slice & return rocksdb::Status::OK(); } -rocksdb::Status CuckooChain::tryCuckooKickOut(engine::Context &ctx, const Slice &user_key, const std::string &ns_key, - CuckooChainMetadata *metadata, uint64_t hash, uint8_t fingerprint, - bool *inserted) { +rocksdb::Status CuckooChain::tryCuckooKickOut(CuckooChainMetadata *metadata, CuckooPageCache *pages, uint64_t hash, + uint8_t fingerprint, bool *inserted) { *inserted = false; // No space found in any filter, try kick-out on the last filter @@ -221,14 +253,12 @@ rocksdb::Status CuckooChain::tryCuckooKickOut(engine::Context &ctx, const Slice last_filter_idx, &num_buckets); if (!s.ok()) return s; - CuckooSubFilter last_filter(storage_, ctx, ns_key, storage_->IsSlotIdEncoded(), metadata->version, - metadata->bucket_size, metadata->page_size, last_filter_idx, num_buckets); + CuckooSubFilter last_filter(pages, last_filter_idx, num_buckets); bool kickout_inserted = false; s = last_filter.TryKickOutInsert(hash, fingerprint, metadata->max_iterations, &kickout_inserted); if (!s.ok()) return s; if (kickout_inserted) { - s = commitSubFilterAndMetadata(ctx, user_key, ns_key, metadata, &last_filter); - if (!s.ok()) return s; + metadata->size++; *inserted = true; return rocksdb::Status::OK(); } @@ -236,8 +266,7 @@ rocksdb::Status CuckooChain::tryCuckooKickOut(engine::Context &ctx, const Slice return rocksdb::Status::OK(); } -rocksdb::Status CuckooChain::expandAndInsertCuckooChain(engine::Context &ctx, const Slice &user_key, - const std::string &ns_key, CuckooChainMetadata *metadata, +rocksdb::Status CuckooChain::expandAndInsertCuckooChain(CuckooChainMetadata *metadata, CuckooPageCache *pages, uint64_t hash, uint8_t fingerprint, bool *inserted) { *inserted = false; @@ -256,33 +285,30 @@ rocksdb::Status CuckooChain::expandAndInsertCuckooChain(engine::Context &ctx, co } 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); + CuckooSubFilter new_filter(pages, new_filter_idx, new_num_buckets); bool new_filter_inserted = false; s = new_filter.TryInsert(hash, fingerprint, &new_filter_inserted); if (!s.ok()) return s; if (!new_filter_inserted) return rocksdb::Status::Corruption("failed to insert into new cuckoo filter"); metadata->n_filters++; - s = commitSubFilterAndMetadata(ctx, user_key, ns_key, metadata, &new_filter); - if (!s.ok()) return s; + metadata->size++; *inserted = true; return rocksdb::Status::OK(); } -rocksdb::Status CuckooChain::commitSubFilterAndMetadata(engine::Context &ctx, const Slice &user_key, - const std::string &ns_key, CuckooChainMetadata *metadata, - CuckooSubFilter *sub_filter) { +rocksdb::Status CuckooChain::commitPagesAndMetadata(engine::Context &ctx, const Slice &user_key, + const std::string &ns_key, CuckooChainMetadata *metadata, + CuckooPageCache *pages) { auto batch = storage_->GetWriteBatchBase(); WriteBatchLogData log_data(kRedisCuckooFilter, std::vector{"add", user_key.ToString()}); auto s = batch->PutLogData(log_data.Encode()); if (!s.ok()) return s; - s = sub_filter->WriteToBatch(batch.Get()); + s = pages->WriteBackDirtyPages(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); diff --git a/src/types/redis_cuckoo_chain.h b/src/types/redis_cuckoo_chain.h index ae20b056cf3..cc089a7f922 100644 --- a/src/types/redis_cuckoo_chain.h +++ b/src/types/redis_cuckoo_chain.h @@ -21,8 +21,10 @@ #pragma once #include "cuckoo_filter.h" +#include "rocksdb/status.h" #include "storage/redis_db.h" #include "storage/redis_metadata.h" +#include "types/cuckoo_filter_page.h" namespace redis { @@ -35,6 +37,17 @@ const uint16_t kCFDefaultMaxIterations = 20; const uint16_t kCFDefaultExpansion = 1; const uint16_t kCFMaxExpansion = 32768; +enum class CuckooFilterInsertResult { + kOk, + kExist, + kFull, +}; + +struct CuckooFilterInsertOptions { + uint64_t capacity = kCFDefaultCapacity; + bool auto_create = true; +}; + class CuckooChain : public Database { public: CuckooChain(engine::Storage *storage, const std::string &ns) : Database(storage, ns) {} @@ -45,7 +58,11 @@ class CuckooChain : public Database { // Adds one item to the cuckoo filter. // Duplicate items are allowed, so added is true whenever insertion succeeds. - rocksdb::Status Add(engine::Context &ctx, const Slice &user_key, const Slice &item, bool *added); + rocksdb::Status Add(engine::Context &ctx, const Slice &user_key, const std::string &item, + CuckooFilterInsertResult &res); + + rocksdb::Status Insert(engine::Context &ctx, const Slice &user_key, const std::vector &items, + CuckooFilterInsertOptions &options, std::vector &rets); private: // Loads metadata for a cuckoo filter key. @@ -53,15 +70,14 @@ class CuckooChain : public Database { static rocksdb::Status validateMetadata(const CuckooChainMetadata &metadata); - rocksdb::Status tryCuckooInsert(engine::Context &ctx, const Slice &user_key, const std::string &ns_key, - CuckooChainMetadata *metadata, uint64_t hash, uint8_t fingerprint, bool *inserted); - rocksdb::Status tryCuckooKickOut(engine::Context &ctx, const Slice &user_key, const std::string &ns_key, - CuckooChainMetadata *metadata, uint64_t hash, uint8_t fingerprint, bool *inserted); - rocksdb::Status expandAndInsertCuckooChain(engine::Context &ctx, const Slice &user_key, const std::string &ns_key, - CuckooChainMetadata *metadata, uint64_t hash, uint8_t fingerprint, - bool *inserted); - rocksdb::Status commitSubFilterAndMetadata(engine::Context &ctx, const Slice &user_key, const std::string &ns_key, - CuckooChainMetadata *metadata, CuckooSubFilter *sub_filter); + static rocksdb::Status tryCuckooInsert(CuckooChainMetadata *metadata, CuckooPageCache *pages, uint64_t hash, + uint8_t fingerprint, bool *inserted); + static rocksdb::Status tryCuckooKickOut(CuckooChainMetadata *metadata, CuckooPageCache *pages, uint64_t hash, + uint8_t fingerprint, bool *inserted); + static rocksdb::Status expandAndInsertCuckooChain(CuckooChainMetadata *metadata, CuckooPageCache *pages, + uint64_t hash, uint8_t fingerprint, bool *inserted); + rocksdb::Status commitPagesAndMetadata(engine::Context &ctx, const Slice &user_key, const std::string &ns_key, + CuckooChainMetadata *metadata, CuckooPageCache *pages); }; } // namespace redis diff --git a/tests/cppunit/disk_test.cc b/tests/cppunit/disk_test.cc index 58c605cebbe..1d168a1afbe 100644 --- a/tests/cppunit/disk_test.cc +++ b/tests/cppunit/disk_test.cc @@ -264,8 +264,8 @@ TEST_F(RedisDiskTest, CuckooFilterDisk) { uint64_t key_size = 0; EXPECT_TRUE(disk->GetKeySize(*ctx_, key_, kRedisCuckooFilter, &key_size).ok()); - bool added = false; - EXPECT_TRUE(cuckoo->Add(*ctx_, key_, "item", &added).ok()); - EXPECT_TRUE(added); + redis::CuckooFilterInsertResult added = redis::CuckooFilterInsertResult::kOk; + EXPECT_TRUE(cuckoo->Add(*ctx_, key_, "item", added).ok()); + EXPECT_TRUE(added == redis::CuckooFilterInsertResult::kOk); EXPECT_TRUE(disk->GetKeySize(*ctx_, key_, kRedisCuckooFilter, &key_size).ok()); } diff --git a/tests/cppunit/types/cuckoo_filter_test.cc b/tests/cppunit/types/cuckoo_filter_test.cc index 66a24ad425d..8739c7f9b60 100644 --- a/tests/cppunit/types/cuckoo_filter_test.cc +++ b/tests/cppunit/types/cuckoo_filter_test.cc @@ -83,10 +83,10 @@ class RedisCuckooFilterTest : public TestBase { void addAndVerify(const std::string &key, const std::string &item, uint64_t capacity, uint8_t bucket_size, uint16_t max_iterations, uint16_t expansion, uint64_t expected_size, uint16_t n_filters = 1) { - bool added = false; - auto s = cuckoo_->Add(*ctx_, key, item, &added); + redis::CuckooFilterInsertResult ret = redis::CuckooFilterInsertResult::kOk; + auto s = cuckoo_->Add(*ctx_, key, item, ret); ASSERT_TRUE(s.ok()) << key << ": add '" << item << "' failed: " << s.ToString(); - ASSERT_TRUE(added) << key << ": item '" << item << "' should have been added"; + ASSERT_TRUE(ret == redis::CuckooFilterInsertResult::kOk) << key << ": item '" << item << "' should have been added"; verifyMetadata(key, capacity, bucket_size, max_iterations, expansion, expected_size, n_filters, 0); } @@ -234,14 +234,14 @@ TEST_F(RedisCuckooFilterTest, ReserveKeepsZeroExpansionNonScaling) { uint64_t added_count = 0; bool full = false; for (int i = 0; i < 100; ++i) { - bool added = false; - auto s = cuckoo_->Add(*ctx_, key_, "item_" + std::to_string(i), &added); - if (!s.ok()) { - ASSERT_TRUE(s.IsAborted()) << s.ToString(); + redis::CuckooFilterInsertResult ret = redis::CuckooFilterInsertResult::kOk; + auto s = cuckoo_->Add(*ctx_, key_, "item_" + std::to_string(i), ret); + ASSERT_TRUE(s.ok()) << s.ToString(); + if (ret == redis::CuckooFilterInsertResult::kFull) { full = true; break; } - ASSERT_TRUE(added); + ASSERT_EQ(ret, redis::CuckooFilterInsertResult::kOk); ++added_count; } @@ -564,16 +564,18 @@ TEST_F(RedisCuckooFilterTest, AddSmallFilterCapacity) { bool full = false; for (int i = 0; i < 100; ++i) { std::string item = "item_" + std::to_string(i); - bool added = false; - auto s = cuckoo_->Add(*ctx_, key_, item, &added); + redis::CuckooFilterInsertResult ret = redis::CuckooFilterInsertResult::kOk; + auto s = cuckoo_->Add(*ctx_, key_, item, ret); if (!s.ok()) { - ASSERT_TRUE(s.IsAborted()) << "Should be Aborted status when full"; + FAIL() << "Unexpected error: " << s.ToString(); + } + if (ret == redis::CuckooFilterInsertResult::kFull) { full = true; break; } - ASSERT_TRUE(added) << "Item should have been added before the filter is full"; + ASSERT_EQ(ret, redis::CuckooFilterInsertResult::kOk) << "Item should have been added before the filter is full"; ++added_count; } @@ -688,9 +690,10 @@ TEST_F(RedisCuckooFilterTest, KickOutErrorDiscardsDirtyPages) { writePage(makePageKey(key_, metadata, 0, 0), original_page); writePage(makePageKey(key_, metadata, 0, 1), std::string(2, static_cast(9))); - redis::CuckooSubFilter sub_filter(storage_.get(), *ctx_, db_->AppendNamespacePrefix(key_), + redis::CuckooPageCache page_cache(storage_.get(), *ctx_, db_->AppendNamespacePrefix(key_), storage_->IsSlotIdEncoded(), metadata.version, metadata.bucket_size, - metadata.page_size, 0, num_buckets); + metadata.page_size); + redis::CuckooSubFilter sub_filter(&page_cache, 0, num_buckets); bool inserted = true; auto s = sub_filter.TryKickOutInsert(hash, fingerprint, metadata.max_iterations, &inserted); ASSERT_TRUE(s.IsCorruption()) << s.ToString(); @@ -716,10 +719,10 @@ TEST_F(RedisCuckooFilterTest, ExpansionWritesNewFilterIndexPage) { CuckooChainMetadata metadata(false); uint64_t added_count = 0; for (int i = 0; i < 100; ++i) { - bool added = false; - auto s = cuckoo_->Add(*ctx_, key_, "item_" + std::to_string(i), &added); + redis::CuckooFilterInsertResult ret = redis::CuckooFilterInsertResult::kOk; + auto s = cuckoo_->Add(*ctx_, key_, "item_" + std::to_string(i), ret); ASSERT_TRUE(s.ok()) << s.ToString(); - ASSERT_TRUE(added); + ASSERT_EQ(ret, redis::CuckooFilterInsertResult::kOk); ++added_count; metadata = getMetadata(key_); @@ -743,3 +746,130 @@ TEST_F(RedisCuckooFilterTest, ExpansionWritesNewFilterIndexPage) { ASSERT_TRUE(s.ok()) << s.ToString(); EXPECT_EQ(page.size(), expected_page_size); } + +TEST_F(RedisCuckooFilterTest, InsertBasic) { + std::vector items = {"item1", "item2", "item3"}; + std::vector rets(3, redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.ok()) << s.ToString(); + ASSERT_EQ(rets.size(), 3); + for (const auto &ret : rets) { + EXPECT_EQ(ret, redis::CuckooFilterInsertResult::kOk); + } + verifyMetadata(key_, redis::kCFDefaultCapacity, redis::kCFDefaultBucketSize, redis::kCFDefaultMaxIterations, + redis::kCFDefaultExpansion, 3, 1); +} + +TEST_F(RedisCuckooFilterTest, InsertNoCreateNonExistent) { + std::vector items = {"item1"}; + std::vector rets(1, redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + options.auto_create = false; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.IsNotFound()) << s.ToString(); +} + +TEST_F(RedisCuckooFilterTest, InsertNoCreateExisting) { + reserveAndVerify(key_, 1000, 4, 500, 2); + std::vector items = {"item1", "item2"}; + std::vector rets(2, redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + options.auto_create = false; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.ok()) << s.ToString(); + ASSERT_EQ(rets.size(), 2); + for (const auto &ret : rets) { + EXPECT_EQ(ret, redis::CuckooFilterInsertResult::kOk); + } + verifyMetadata(key_, 1000, 4, 500, 2, 2, 1); +} + +TEST_F(RedisCuckooFilterTest, InsertWithCustomCapacity) { + std::vector items = {"item1", "item2"}; + std::vector rets(2, redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + options.capacity = 5000; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.ok()) << s.ToString(); + ASSERT_EQ(rets.size(), 2); + for (const auto &ret : rets) { + EXPECT_EQ(ret, redis::CuckooFilterInsertResult::kOk); + } + verifyMetadata(key_, 5000, redis::kCFDefaultBucketSize, redis::kCFDefaultMaxIterations, redis::kCFDefaultExpansion, 2, + 1); +} + +TEST_F(RedisCuckooFilterTest, InsertInvalidCapacity) { + std::vector items = {"item1"}; + std::vector rets(1, redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + + options.capacity = 1; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); + + options.capacity = std::numeric_limits::max(); + s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString(); +} + +TEST_F(RedisCuckooFilterTest, InsertExistingFilterIgnoresCapacity) { + reserveAndVerify(key_, 1000, 4, 500, 2); + + std::vector items = {"item1"}; + std::vector rets(1, redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + options.capacity = 9999; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.ok()) << s.ToString(); + ASSERT_EQ(rets.size(), 1); + EXPECT_EQ(rets[0], redis::CuckooFilterInsertResult::kOk); + verifyMetadata(key_, 1000, 4, 500, 2, 1, 1); +} + +TEST_F(RedisCuckooFilterTest, InsertNonScalingFilterFull) { + reserveAndVerify(key_, 2, 1, 1, 0); + + std::vector items(50); + for (int i = 0; i < 50; ++i) { + items[i] = "item_" + std::to_string(i); + } + + std::vector rets(items.size(), redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.ok()) << s.ToString(); + ASSERT_EQ(rets.size(), items.size()); + + size_t ok_count = 0; + size_t full_count = 0; + for (const auto &ret : rets) { + if (ret == redis::CuckooFilterInsertResult::kOk) { + ++ok_count; + } else if (ret == redis::CuckooFilterInsertResult::kFull) { + ++full_count; + } + } + + EXPECT_GT(ok_count, 0); + EXPECT_GT(full_count, 0); + EXPECT_EQ(ok_count + full_count, items.size()); + + auto metadata = getMetadata(key_); + EXPECT_EQ(metadata.size, ok_count); +} + +TEST_F(RedisCuckooFilterTest, InsertDuplicateItems) { + std::vector items = {"duplicate", "duplicate", "duplicate"}; + std::vector rets(3, redis::CuckooFilterInsertResult::kOk); + redis::CuckooFilterInsertOptions options; + auto s = cuckoo_->Insert(*ctx_, key_, items, options, rets); + ASSERT_TRUE(s.ok()) << s.ToString(); + ASSERT_EQ(rets.size(), 3); + for (const auto &ret : rets) { + EXPECT_EQ(ret, redis::CuckooFilterInsertResult::kOk); + } + verifyMetadata(key_, redis::kCFDefaultCapacity, redis::kCFDefaultBucketSize, redis::kCFDefaultMaxIterations, + redis::kCFDefaultExpansion, 3, 1); +} diff --git a/tests/gocase/unit/type/bloom/cuckoo_filter_test.go b/tests/gocase/unit/type/bloom/cuckoo_filter_test.go index 852dde75ed0..cb23697b6d6 100644 --- a/tests/gocase/unit/type/bloom/cuckoo_filter_test.go +++ b/tests/gocase/unit/type/bloom/cuckoo_filter_test.go @@ -188,3 +188,121 @@ func TestCuckooFilter(t *testing.T) { require.Equal(t, int64(1), result.Val()) }) } + +func TestCuckooFilterInsert(t *testing.T) { + srv := util.StartServer(t, map[string]string{}) + defer srv.Close() + ctx := context.Background() + rdb := srv.NewClient() + defer func() { require.NoError(t, rdb.Close()) }() + + t.Run("Insert wrong number of arguments", func(t *testing.T) { + require.Error(t, rdb.Do(ctx, "cf.insert").Err()) + require.Error(t, rdb.Do(ctx, "cf.insert", "key_only").Err()) + require.Error(t, rdb.Do(ctx, "cf.insert", "key_only", "ITEMS").Err()) + }) + + t.Run("Insert missing ITEMS keyword returns syntax error", func(t *testing.T) { + key := "test_cf_insert_no_items_kw" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.ErrorContains(t, rdb.Do(ctx, "cf.insert", key, "item1").Err(), "ERR wrong number of arguments") + }) + + t.Run("Insert auto-creates filter", func(t *testing.T) { + key := "test_cf_insert_autocreate" + require.NoError(t, rdb.Del(ctx, key).Err()) + result := rdb.Do(ctx, "cf.insert", key, "ITEMS", "item1", "item2", "item3") + require.NoError(t, result.Err()) + require.Equal(t, []interface{}{int64(1), int64(1), int64(1)}, result.Val()) + require.Equal(t, "MBbloomCF", rdb.Type(ctx, key).Val()) + }) + + t.Run("Insert NOCREATE on non-existent key returns error", func(t *testing.T) { + key := "test_cf_insert_nocreate_missing" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.ErrorContains(t, rdb.Do(ctx, "cf.insert", key, "NOCREATE", "ITEMS", "item1").Err(), "ERR NotFound:") + }) + + t.Run("Insert NOCREATE on existing filter succeeds", func(t *testing.T) { + key := "test_cf_insert_nocreate_existing" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.NoError(t, rdb.Do(ctx, "cf.reserve", key, "1000").Err()) + result := rdb.Do(ctx, "cf.insert", key, "NOCREATE", "ITEMS", "item1", "item2") + require.NoError(t, result.Err()) + require.Equal(t, []interface{}{int64(1), int64(1)}, result.Val()) + }) + + t.Run("Insert invalid CAPACITY returns error", func(t *testing.T) { + key := "test_cf_insert_bad_capacity" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.ErrorContains(t, rdb.Do(ctx, "cf.insert", key, "CAPACITY", "abc", "ITEMS", "item1").Err(), "invalid capacity") + require.ErrorContains(t, rdb.Do(ctx, "cf.insert", key, "CAPACITY", "0", "ITEMS", "item1").Err(), "capacity must be larger than 0") + require.ErrorContains(t, rdb.Do(ctx, "cf.insert", key, "CAPACITY", "1", "ITEMS", "item1").Err(), "capacity must be at least 2") + }) + + t.Run("Insert CAPACITY ignored when filter already exists", func(t *testing.T) { + key := "test_cf_insert_capacity_ignored" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.NoError(t, rdb.Do(ctx, "cf.reserve", key, "1000").Err()) + result := rdb.Do(ctx, "cf.insert", key, "CAPACITY", "9999", "ITEMS", "item1") + require.NoError(t, result.Err()) + require.Equal(t, []interface{}{int64(1)}, result.Val()) + // Filter key was already reserved with capacity=1000, confirmed by re-reserving failing + require.ErrorContains(t, rdb.Do(ctx, "cf.reserve", key, "1000").Err(), "already exists") + }) + + t.Run("Insert multiple items returns per-item results", func(t *testing.T) { + key := "test_cf_insert_multi" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.NoError(t, rdb.Do(ctx, "cf.reserve", key, "1000").Err()) + result := rdb.Do(ctx, "cf.insert", key, "ITEMS", "alpha", "beta", "gamma") + require.NoError(t, result.Err()) + require.Equal(t, []interface{}{int64(1), int64(1), int64(1)}, result.Val()) + }) + + t.Run("Insert duplicate items all succeed (CF.INSERT allows duplicates)", func(t *testing.T) { + key := "test_cf_insert_duplicates" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.NoError(t, rdb.Do(ctx, "cf.reserve", key, "1000").Err()) + result := rdb.Do(ctx, "cf.insert", key, "ITEMS", "dup", "dup", "dup") + require.NoError(t, result.Err()) + require.Equal(t, []interface{}{int64(1), int64(1), int64(1)}, result.Val()) + }) + + t.Run("Insert into non-scaling full filter", func(t *testing.T) { + key := "test_cf_insert_full_nonscaling" + // expansion=0 disables scaling; small capacity fills quickly + require.NoError(t, rdb.Do(ctx, "cf.reserve", key, "4", "BUCKETSIZE", "1", "MAXITERATIONS", "1", "EXPANSION", "0").Err()) + full := false + for i := 0; i < 100; i++ { + result := rdb.Do(ctx, "cf.add", key, fmt.Sprintf("full_item_%d", i)) + if result.Err() != nil { + require.ErrorContains(t, result.Err(), "filter is full") + full = true + } + } + require.True(t, full, "Non-scaling filter should eventually become full") + result := rdb.Do(ctx, "cf.insert", key, "ITEMS", "full_item_101", "full_item_102") + require.NoError(t, result.Err()) // command succeeds at the protocol level + vals := result.Val().([]interface{}) + require.Equal(t, 2, len(vals)) + for _, v := range vals { + // Full items embed an error entry in the multi-bulk response, never int64(1). + require.Equal(t, int64(-1), v) + } + }) + + t.Run("Insert empty string item", func(t *testing.T) { + key := "test_cf_insert_empty_item" + require.NoError(t, rdb.Del(ctx, key).Err()) + result := rdb.Do(ctx, "cf.insert", key, "ITEMS", "") + require.NoError(t, result.Err()) + require.Equal(t, []interface{}{int64(1)}, result.Val()) + }) + + t.Run("Insert unknown option returns syntax error", func(t *testing.T) { + key := "test_cf_insert_unknown_opt" + require.NoError(t, rdb.Del(ctx, key).Err()) + require.ErrorContains(t, rdb.Do(ctx, "cf.insert", key, "UNKNOWNOPT", "ITEMS", "item1").Err(), "syntax error") + }) +}