Skip to content

feat(CuckooFilter): Add CF.INSERT command #3593

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
geetanshjuneja wants to merge 5 commits into
base: unstable
Choose a base branch
from
Open

feat(CuckooFilter): Add CF.INSERT command #3593

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
cdd7140
Added cf.insert command
geetanshjuneja Aug 16, 2026
94e7461
Added cf.insert command tests
geetanshjuneja Aug 16, 2026
2240ddd
nits
geetanshjuneja Aug 16, 2026
3093171
Move page cache to cuckoo filter chain level
geetanshjuneja Aug 23, 2026
4d57d88
Nits
geetanshjuneja Aug 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 96 additions & 4 deletions src/commands/cmd_cuckoo_filter.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,14 @@
*
*/

#include <cstdlib>
#include <vector>

#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 {
Expand Down Expand Up @@ -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<std::string> &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<uint64_t>();
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<CuckooFilterInsertResult> 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<std::string> items_;
};

// Register the CF.RESERVE and CF.ADD commands
REDIS_REGISTER_COMMANDS(CuckooFilter, MakeCmdAttr<CommandCFReserve>("cf.reserve", -3, "write", 1, 1, 1),
MakeCmdAttr<CommandCFAdd>("cf.add", 3, "write", 1, 1, 1))
MakeCmdAttr<CommandCFAdd>("cf.add", 3, "write", 1, 1, 1),
MakeCmdAttr<CommandCFInsert>("cf.insert", -4, "write", 1, 1, 1))

} // namespace redis
2 changes: 2 additions & 0 deletions src/types/cuckoo_filter_page.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,8 @@ class CuckooPageCache {

void DiscardCachedPages();

uint8_t GetBucketSize() const { return bucket_size_; }

private:
struct PageEntry {
std::string data;
Expand Down
32 changes: 14 additions & 18 deletions src/types/cuckoo_filter_sub_filter.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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();
Comment on lines +54 to +61

@nagisa-kunhah nagisa-kunhah Aug 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the INSERT command, I think we may not be able to simply discard the cached pages here. Since an INSERT can involve multiple items, some items may have already been inserted successfully before a later insertion fails.

Do we need a rollback/restore mechanism here to preserve the changes from the previous successful insertions? Sorry I missed this in the original design.

return s;
}
current_fp = old_fp;
Expand All @@ -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) {
Expand All @@ -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_; }
Expand Down
6 changes: 2 additions & 4 deletions src/types/cuckoo_filter_sub_filter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_; }
Expand All @@ -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
Loading
Loading