You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@kvrocks.apache.org by "zncleon (via GitHub)" <gi...@apache.org> on 2023/08/03 16:13:15 UTC
[GitHub] [kvrocks] zncleon opened a new pull request, #1632: [WIP] BloomFilter command Draft and implemention
zncleon opened a new pull request, #1632:
URL: https://github.com/apache/kvrocks/pull/1632
This is part of Redis BloomFilter Implemention.
- [ ] BF.ADD, BF.MADD basic impl
- [ ] BF.EXIST, BF.MEXIST basic impl
- [ ] BF.RESERVE basic impl
- [ ] BF.INSERT (insertCommon compl, need to compl command parse)
- [ ] BF.CARD basic impl
- [ ] which cf to save bfmetadata need to think
- [ ] some "todo" need to modify
- [ ] more testcases
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1288505422
##########
src/commands/cmd_bloomfilter.cc:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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 "commander.h"
+#include "error_constants.h"
+#include "server/server.h"
+#include "types/redis_bloomfilter.h"
+
+namespace redis {
+
+class CommandBFReserve : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ auto parse_error_rate = ParseFloat<double>(args[2]);
+ if (!parse_error_rate) {
+ return {Status::RedisParseErr, errValueIsNotFloat};
+ }
+ error_rate_ = *parse_error_rate;
+ if (error_rate_ >= 1 || error_rate_ <= 0) {
+ return {Status::RedisParseErr, "0 < error rate range < 1"};
+ }
+
+ auto parse_capacity = ParseInt<uint64_t>(args[3], 10);
+ if (!parse_capacity) {
+ return {Status::RedisParseErr, errValueNotInteger};
+ }
+ capacity_ = *parse_capacity;
+ if (capacity_ <= 0) {
+ return {Status::RedisParseErr, "capacity should be larger than 0"};
+ }
+
+ size_t expect_size = 4;
+ auto iter = find(args.begin(), args.end(), "NONSCALING");
+ if (iter != args.end()) {
+ scaling_ = 0;
+ expect_size += 1;
+ }
+ iter = find(args.begin(), args.end(), "EXPANSION");
+ if (iter != args.end()) {
+ if (scaling_ == 0) {
+ return {Status::RedisParseErr, "nonscaling filters cannot expand"};
+ } else if (++iter == args.end()) {
+ return {Status::RedisParseErr, "no expansion"};
+ } else {
+ auto parse_expansion = ParseInt<uint16_t>(*iter, 10);
+ if (!parse_expansion) {
+ return {Status::RedisParseErr, "ERR bad expansion"};
+ }
+ expansion_ = *parse_expansion;
+ if (expansion_ < 1) {
+ return {Status::RedisParseErr, "expansion should be greater or equal to 1"};
+ }
+ expect_size += 2;
+ }
+ }
+
+ if (args.size() != expect_size) {
+ return {Status::RedisParseErr, errInvalidSyntax};
+ }
+ return Commander::Parse(args);
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ redis::BloomFilter bloomfilter_db(svr->storage, conn->GetNamespace());
+ auto s = bloomfilter_db.Reserve(args_[1], error_rate_, capacity_, expansion_, scaling_);
+ if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
+
+ *output = redis::SimpleString("OK");
+ return Status::OK();
+ }
+
+ private:
+ double error_rate_;
+ uint64_t capacity_;
+ uint16_t expansion_ = kBFDefaultExpansion;
+ uint16_t scaling_ = 1;
+};
+
+class CommandBFADD : public Commander {
+ public:
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ redis::BloomFilter bloom_db(svr->storage, conn->GetNamespace());
+ int ret = 0;
+ auto s = bloom_db.Add(args_[1], args_[2], ret);
+ if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
+
+ *output = redis::Integer(ret);
+ return Status::OK();
+ }
+};
+
+class CommandBFMADD : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ for (size_t i = 2; i < args_.size(); ++i) {
+ items_.emplace_back(args_[i]);
+ }
+ return Commander::Parse(args);
+ }
+
+ Status Execute(Server *svr, Connection *conn, std::string *output) override {
+ redis::BloomFilter bloom_db(svr->storage, conn->GetNamespace());
+ std::vector<int> rets(items_.size(), 0);
+ auto s = bloom_db.MAdd(args_[1], items_, rets);
+ if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
+
+ *output = redis::MultiInteger(rets);
+ return Status::OK();
+ }
+
+ private:
+ std::vector<Slice> items_;
+};
+
+class CommandBFINSERT : public Commander {
Review Comment:
Maybe `CommandBFInsert` would be ok here?
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
+ public:
+ uint16_t n_filters;
+ uint16_t options;
+ uint16_t growth;
+
+ explicit SBChainMetadata(bool generate_version = true) : Metadata(kRedisBloomFilter, generate_version) {}
+
+ void Encode(std::string *dst) override;
+ rocksdb::Status Decode(const std::string &bytes) override;
+};
+
+class BFMetadata : public Metadata {
+ public:
Review Comment:
As a simple implemention. We can use an aligned bloom filter. And keep only:
```
class BFMetadata : public Metadata {
public:
uint32_t hashes;
uint64_t entries;
double error;
double bpe;
uint64_t bf_bytes;
};
```
##########
src/types/redis_bloomfilter.cc:
##########
@@ -0,0 +1,378 @@
+/*
+ * 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_bloomfilter.h"
+
+namespace redis {
+
+double BloomFilter::calcBpe(double error) {
+ static const double denom = 0.480453013918201; // ln(2)^2
+ double num = log(error);
+
+ double bpe = -(num / denom);
+ if (bpe < 0) {
+ bpe = -bpe;
+ }
+ return bpe;
+}
+
+void BloomFilter::appendBFSuffix(const Slice &ns_key, uint16_t filters_index, std::string *output) {
+ output->clear();
+
+ output->append(ns_key.data(), ns_key.size());
+ output->append(kBloomFilterSeparator);
+ PutFixed16(output, filters_index);
+}
+
+void BloomFilter ::appendBFMetaSuffix(const Slice &bf_key, std::string *output) {
+ output->clear();
+
+ output->append(bf_key.data(), bf_key.size());
+ output->append(kBloomFilterSeparator);
+ output->append("meta");
+}
+
+void BloomFilter::bfInit(BFMetadata *bf_metadata, uint64_t entries, double error, unsigned int options) {
+ bf_metadata->error = error;
+ bf_metadata->bf_bits = 0;
+ bf_metadata->entries = entries;
+ bf_metadata->bpe = calcBpe(error);
+ bf_metadata->size = 0;
+
+ uint64_t bits = 0;
+
+ {
+ double bn2 = logb(static_cast<double>(entries) * bf_metadata->bpe);
+ // if (bn2 > 63 || bn2 == INFINITY) {
+ // return 1;
+ // }
+ bf_metadata->n2 = static_cast<uint8_t>(bn2) + 1;
+ bits = 1LLU << bf_metadata->n2;
+
+ // Determine the number of extra bits available for more items. We rounded
+ // up the number of bits to the next-highest power of two. This means we
+ // might have up to 2x the bits available to us.
+ uint64_t bit_diff = bits - static_cast<uint64_t>(static_cast<double>(entries) * bf_metadata->bpe);
+ // The number of additional items we can store is the extra number of bits
+ // divided by bits-per-element
+ auto item_diff = static_cast<uint64_t>(static_cast<double>(bit_diff) / bf_metadata->bpe);
+ bf_metadata->entries += item_diff;
+ }
+
+ if (bits % 64) {
+ bf_metadata->bf_bytes = ((bits / 64) + 1) * 8;
+ } else {
+ bf_metadata->bf_bytes = bits / 8;
+ }
+ bf_metadata->bf_bits = bf_metadata->bf_bytes * 8;
+
+ bf_metadata->hashes = (int)ceil(0.693147180559945 * bf_metadata->bpe); // ln(2)
+}
+
+BloomHashval BloomFilter::bloomCalcHash64(const void *buffer, int len) {
+ BloomHashval rv;
+ rv.a = MurmurHash64A_Bloom(buffer, len, 0xc6a4a7935bd1e995ULL);
+ rv.b = MurmurHash64A_Bloom(buffer, len, rv.a);
+ return rv;
+}
+
+int BloomFilter::testBitSetBit(std::string &value, uint64_t x, ReadWriteMode mode) {
+ uint8_t mask = 1 << (x % 8);
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ uint8_t c = value[byte_index]; // expensive memory access
+
+ if (c & mask) {
+ return 1;
+ } else {
+ if (mode == ReadWriteMode::MODE_WRITE) {
+ uint8_t s = c | mask;
+ value[byte_index] = static_cast<char>(s);
+ }
+ return 0;
+ }
+}
+
+rocksdb::Status BloomFilter::getSBChainMetadata(const Slice &ns_key, SBChainMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, ns_key, metadata);
+}
+
+rocksdb::Status BloomFilter::getBFMetadata(const Slice &bf_meta_key, BFMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, bf_meta_key, metadata);
+}
+
+rocksdb::Status BloomFilter::createSBChain(const Slice &ns_key, double error_rate, uint64_t capacity,
+ uint16_t expansion, uint16_t scaling, SBChainMetadata &sb_chain_metadata) {
+ sb_chain_metadata.n_filters = 1;
+ sb_chain_metadata.growth = expansion;
+ sb_chain_metadata.options = scaling;
+ sb_chain_metadata.size = 0;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createSBChain"});
+ batch->PutLogData(log_data.Encode());
+
+ std::string sb_chain_bytes;
+ sb_chain_metadata.Encode(&sb_chain_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_bytes);
+
+ std::string bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ bfInit(&bf_metadata, capacity, error_rate, scaling);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::createBloomFilter(const Slice &ns_key, SBChainMetadata &sb_chain_metadata,
+ BFMetadata &bf_metadata, std::string &bf_key) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createBloomFilter"});
+ batch->PutLogData(log_data.Encode());
+
+ BFMetadata old_bf_metadata = bf_metadata;
+ sb_chain_metadata.n_filters += 1;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ bfInit(&bf_metadata, sb_chain_metadata.growth * old_bf_metadata.entries,
+ old_bf_metadata.error * kErrorTighteningRatio, sb_chain_metadata.options);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::bloomCheckAdd64(const Slice &bf_key, BFMetadata &bf_metadata, BloomHashval hashval,
+ ReadWriteMode mode, ObserverOrUniquePtr<rocksdb::WriteBatchBase> &batch,
+ int &ret) {
+ int found_unset = 0;
+ const uint64_t mod = (1LLU << bf_metadata.n2);
+ for (uint64_t i = 0; i < bf_metadata.hashes; i++) {
+ uint64_t x = ((hashval.a + i * hashval.b)) % mod;
+ uint32_t index = (x / kBloomFilterSegmentBits) * kBloomFilterSegmentBytes;
+ std::string sub_key, value;
+ InternalKey(bf_key, std::to_string(index), bf_metadata.version, storage_->IsSlotIdEncoded()).Encode(&sub_key);
+ rocksdb::Status s =
+ storage_->Get(rocksdb::ReadOptions(), sub_key, &value); // todo: use bf_key in InternalKey maybe error
+ if (!s.ok() && !s.IsNotFound()) return s;
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ if (byte_index >= value.size()) { // expand the bloomfilter
+ size_t expand_size = 0;
+ if (byte_index >= value.size() * 2) {
+ expand_size = byte_index - value.size() + 1;
+ } else if (value.size() * 2 > kBloomFilterSegmentBytes) {
+ expand_size = kBloomFilterSegmentBytes - value.size();
+ } else {
+ expand_size = value.size();
+ }
+ value.append(expand_size, 0);
+ }
+
+ if (!testBitSetBit(value, x, mode)) {
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 0;
+ return rocksdb::Status::OK();
+ } else {
+ found_unset = 1;
+ batch->Put(sub_key, value);
+ storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+ // todo: if the indexs are same, the value need to be updated immediately, if have other ways to compl it
+ }
+ }
+ }
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 1;
+ } else {
+ ret = found_unset;
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status BloomFilter::Reserve(const Slice &user_key, double error_rate, uint64_t capacity, uint16_t expansion,
+ uint16_t scaling) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("item exists");
+ }
+
+ return createSBChain(ns_key, error_rate, capacity, expansion, scaling, sb_chain_metadata);
+}
+
+rocksdb::Status BloomFilter::Add(const Slice &user_key, const Slice &item, int &ret) {
+ std::vector<int> tmp(1, 0);
+ rocksdb::Status s = MAdd(user_key, {item}, tmp);
+ ret = tmp[0];
+ return s;
+}
+
+rocksdb::Status BloomFilter::MAdd(const Slice &user_key, const std::vector<Slice> &items, std::vector<int> &rets) {
+ BFInsertOptions insert_options = {.capacity = kBFDefaultInitCapacity,
+ .error_rate = kBFDefaultErrorRate,
+ .autocreate = 1,
+ .expansion = kBFDefaultExpansion,
+ .scaling = 1};
+ return InsertCommon(user_key, items, &insert_options, rets);
+}
+
+rocksdb::Status BloomFilter::InsertCommon(const Slice &user_key, const std::vector<Slice> &items,
+ const BFInsertOptions *insert_options, std::vector<int> &rets) {
Review Comment:
```suggestion
const BFInsertOptions& insert_options, std::vector<int>* rets) {
```
I think it's better to mark `insert_options` as const reference here.
##########
src/types/redis_bloomfilter.cc:
##########
@@ -0,0 +1,378 @@
+/*
+ * 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_bloomfilter.h"
+
+namespace redis {
+
+double BloomFilter::calcBpe(double error) {
+ static const double denom = 0.480453013918201; // ln(2)^2
+ double num = log(error);
+
+ double bpe = -(num / denom);
+ if (bpe < 0) {
+ bpe = -bpe;
+ }
+ return bpe;
+}
+
+void BloomFilter::appendBFSuffix(const Slice &ns_key, uint16_t filters_index, std::string *output) {
+ output->clear();
+
+ output->append(ns_key.data(), ns_key.size());
+ output->append(kBloomFilterSeparator);
+ PutFixed16(output, filters_index);
+}
+
+void BloomFilter ::appendBFMetaSuffix(const Slice &bf_key, std::string *output) {
+ output->clear();
+
+ output->append(bf_key.data(), bf_key.size());
+ output->append(kBloomFilterSeparator);
+ output->append("meta");
+}
+
+void BloomFilter::bfInit(BFMetadata *bf_metadata, uint64_t entries, double error, unsigned int options) {
+ bf_metadata->error = error;
+ bf_metadata->bf_bits = 0;
+ bf_metadata->entries = entries;
+ bf_metadata->bpe = calcBpe(error);
+ bf_metadata->size = 0;
+
+ uint64_t bits = 0;
+
+ {
+ double bn2 = logb(static_cast<double>(entries) * bf_metadata->bpe);
+ // if (bn2 > 63 || bn2 == INFINITY) {
+ // return 1;
+ // }
+ bf_metadata->n2 = static_cast<uint8_t>(bn2) + 1;
+ bits = 1LLU << bf_metadata->n2;
+
+ // Determine the number of extra bits available for more items. We rounded
+ // up the number of bits to the next-highest power of two. This means we
+ // might have up to 2x the bits available to us.
+ uint64_t bit_diff = bits - static_cast<uint64_t>(static_cast<double>(entries) * bf_metadata->bpe);
+ // The number of additional items we can store is the extra number of bits
+ // divided by bits-per-element
+ auto item_diff = static_cast<uint64_t>(static_cast<double>(bit_diff) / bf_metadata->bpe);
+ bf_metadata->entries += item_diff;
+ }
+
+ if (bits % 64) {
+ bf_metadata->bf_bytes = ((bits / 64) + 1) * 8;
+ } else {
+ bf_metadata->bf_bytes = bits / 8;
+ }
+ bf_metadata->bf_bits = bf_metadata->bf_bytes * 8;
+
+ bf_metadata->hashes = (int)ceil(0.693147180559945 * bf_metadata->bpe); // ln(2)
+}
+
+BloomHashval BloomFilter::bloomCalcHash64(const void *buffer, int len) {
+ BloomHashval rv;
+ rv.a = MurmurHash64A_Bloom(buffer, len, 0xc6a4a7935bd1e995ULL);
+ rv.b = MurmurHash64A_Bloom(buffer, len, rv.a);
+ return rv;
+}
+
+int BloomFilter::testBitSetBit(std::string &value, uint64_t x, ReadWriteMode mode) {
+ uint8_t mask = 1 << (x % 8);
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ uint8_t c = value[byte_index]; // expensive memory access
+
+ if (c & mask) {
+ return 1;
+ } else {
+ if (mode == ReadWriteMode::MODE_WRITE) {
+ uint8_t s = c | mask;
+ value[byte_index] = static_cast<char>(s);
+ }
+ return 0;
+ }
+}
+
+rocksdb::Status BloomFilter::getSBChainMetadata(const Slice &ns_key, SBChainMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, ns_key, metadata);
+}
+
+rocksdb::Status BloomFilter::getBFMetadata(const Slice &bf_meta_key, BFMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, bf_meta_key, metadata);
+}
+
+rocksdb::Status BloomFilter::createSBChain(const Slice &ns_key, double error_rate, uint64_t capacity,
+ uint16_t expansion, uint16_t scaling, SBChainMetadata &sb_chain_metadata) {
+ sb_chain_metadata.n_filters = 1;
+ sb_chain_metadata.growth = expansion;
+ sb_chain_metadata.options = scaling;
+ sb_chain_metadata.size = 0;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createSBChain"});
+ batch->PutLogData(log_data.Encode());
+
+ std::string sb_chain_bytes;
+ sb_chain_metadata.Encode(&sb_chain_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_bytes);
+
+ std::string bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ bfInit(&bf_metadata, capacity, error_rate, scaling);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::createBloomFilter(const Slice &ns_key, SBChainMetadata &sb_chain_metadata,
+ BFMetadata &bf_metadata, std::string &bf_key) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createBloomFilter"});
+ batch->PutLogData(log_data.Encode());
+
+ BFMetadata old_bf_metadata = bf_metadata;
+ sb_chain_metadata.n_filters += 1;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ bfInit(&bf_metadata, sb_chain_metadata.growth * old_bf_metadata.entries,
+ old_bf_metadata.error * kErrorTighteningRatio, sb_chain_metadata.options);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::bloomCheckAdd64(const Slice &bf_key, BFMetadata &bf_metadata, BloomHashval hashval,
+ ReadWriteMode mode, ObserverOrUniquePtr<rocksdb::WriteBatchBase> &batch,
+ int &ret) {
+ int found_unset = 0;
+ const uint64_t mod = (1LLU << bf_metadata.n2);
+ for (uint64_t i = 0; i < bf_metadata.hashes; i++) {
+ uint64_t x = ((hashval.a + i * hashval.b)) % mod;
+ uint32_t index = (x / kBloomFilterSegmentBits) * kBloomFilterSegmentBytes;
+ std::string sub_key, value;
+ InternalKey(bf_key, std::to_string(index), bf_metadata.version, storage_->IsSlotIdEncoded()).Encode(&sub_key);
+ rocksdb::Status s =
+ storage_->Get(rocksdb::ReadOptions(), sub_key, &value); // todo: use bf_key in InternalKey maybe error
+ if (!s.ok() && !s.IsNotFound()) return s;
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ if (byte_index >= value.size()) { // expand the bloomfilter
+ size_t expand_size = 0;
+ if (byte_index >= value.size() * 2) {
+ expand_size = byte_index - value.size() + 1;
+ } else if (value.size() * 2 > kBloomFilterSegmentBytes) {
+ expand_size = kBloomFilterSegmentBytes - value.size();
+ } else {
+ expand_size = value.size();
+ }
+ value.append(expand_size, 0);
+ }
+
+ if (!testBitSetBit(value, x, mode)) {
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 0;
+ return rocksdb::Status::OK();
+ } else {
+ found_unset = 1;
+ batch->Put(sub_key, value);
+ storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+ // todo: if the indexs are same, the value need to be updated immediately, if have other ways to compl it
+ }
+ }
+ }
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 1;
+ } else {
+ ret = found_unset;
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status BloomFilter::Reserve(const Slice &user_key, double error_rate, uint64_t capacity, uint16_t expansion,
+ uint16_t scaling) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("item exists");
+ }
+
+ return createSBChain(ns_key, error_rate, capacity, expansion, scaling, sb_chain_metadata);
+}
+
+rocksdb::Status BloomFilter::Add(const Slice &user_key, const Slice &item, int &ret) {
+ std::vector<int> tmp(1, 0);
+ rocksdb::Status s = MAdd(user_key, {item}, tmp);
+ ret = tmp[0];
+ return s;
+}
+
+rocksdb::Status BloomFilter::MAdd(const Slice &user_key, const std::vector<Slice> &items, std::vector<int> &rets) {
+ BFInsertOptions insert_options = {.capacity = kBFDefaultInitCapacity,
+ .error_rate = kBFDefaultErrorRate,
+ .autocreate = 1,
+ .expansion = kBFDefaultExpansion,
+ .scaling = 1};
+ return InsertCommon(user_key, items, &insert_options, rets);
+}
+
+rocksdb::Status BloomFilter::InsertCommon(const Slice &user_key, const std::vector<Slice> &items,
+ const BFInsertOptions *insert_options, std::vector<int> &rets) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+
+ if (s.IsNotFound() && insert_options->autocreate) {
+ rocksdb::Status create_s = createSBChain(ns_key, insert_options->error_rate, insert_options->capacity,
+ insert_options->expansion, insert_options->scaling, sb_chain_metadata);
+ if (!create_s.ok()) return create_s;
+ } else if (!s.ok()) {
+ return s;
+ }
+
+ std::vector<BFMetadata> bf_metadata_list;
+ for (u_int16_t i = 0; i < sb_chain_metadata.n_filters; ++i) {
+ std::string bf_key, bf_meta_key;
+ appendBFSuffix(ns_key, i, &bf_key);
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ s = getBFMetadata(bf_meta_key, &bf_metadata);
+ if (!s.ok()) return s;
+ bf_metadata_list.push_back(bf_metadata);
+ }
+
+ BFMetadata cur_bf_metadata = bf_metadata_list.back();
+ std::string cur_bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &cur_bf_key);
+ if (!sb_chain_metadata.options && cur_bf_metadata.size + items.size() > cur_bf_metadata.entries) {
+ return rocksdb::Status::Aborted("filter is full and is nonscaling");
+ }
+
+ int suc_insert = 0;
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"insert"});
+ batch->PutLogData(log_data.Encode());
+ for (size_t i = 0; i < items.size(); ++i) {
+ bool item_exist = false;
+ const auto &item = items[i];
+ BloomHashval h = bloomCalcHash64(item.data(), static_cast<int>(item.size()));
+
+ // check
+ for (int ii = sb_chain_metadata.n_filters - 1; ii >= 0; --ii) {
+ std::string bf_key;
+ appendBFSuffix(ns_key, ii, &bf_key);
+ int ret = 0;
+ s = bloomCheckAdd64(bf_key, bf_metadata_list[ii], h, ReadWriteMode::MODE_READ, batch, ret);
+ if (ret == 1) {
+ item_exist = true;
+ break;
+ }
+ }
+
+ // insert
+ if (!item_exist) {
+ if (cur_bf_metadata.size >= cur_bf_metadata.entries) {
+ std::string old_bf_metadata_bytes, cur_bf_meta_key;
+ cur_bf_metadata.Encode(&old_bf_metadata_bytes);
+ appendBFMetaSuffix(cur_bf_key, &cur_bf_meta_key);
+ batch->Put(metadata_cf_handle_, cur_bf_meta_key, old_bf_metadata_bytes);
+ createBloomFilter(ns_key, sb_chain_metadata, cur_bf_metadata,
+ cur_bf_key); // todo: the scaling function need to test
+ }
+ int ret = 0;
+ s = bloomCheckAdd64(cur_bf_key, cur_bf_metadata, h, ReadWriteMode::MODE_WRITE, batch, ret);
+ rets[i] = ret;
+ suc_insert += ret;
+ cur_bf_metadata.size += ret;
+ }
+ }
+ sb_chain_metadata.size += suc_insert;
+
+ std::string cur_bf_metadata_bytes, cur_bf_meta_key, sb_chain_metadata_bytes;
+ cur_bf_metadata.Encode(&cur_bf_metadata_bytes);
+ sb_chain_metadata.Encode(&sb_chain_metadata_bytes);
+ appendBFMetaSuffix(cur_bf_key, &cur_bf_meta_key);
+ batch->Put(metadata_cf_handle_, cur_bf_meta_key, cur_bf_metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_metadata_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::Exist(const Slice &user_key, const Slice &item, int &ret) {
+ std::vector<int> tmp(1, 0);
+ rocksdb::Status s = MExist(user_key, {item}, tmp);
+ ret = tmp[0];
+ return s;
+}
+
+rocksdb::Status BloomFilter::MExist(const Slice &user_key, const std::vector<Slice> &items, std::vector<int> &rets) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok()) return s;
+
+ std::vector<BFMetadata> bf_metadata_list;
+ for (u_int16_t i = 0; i < sb_chain_metadata.n_filters; ++i) {
+ std::string bf_key, bf_meta_key;
+ appendBFSuffix(ns_key, i, &bf_key);
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ s = getBFMetadata(bf_meta_key, &bf_metadata);
+ if (!s.ok()) return s;
+ bf_metadata_list.push_back(bf_metadata);
+ }
+
+ for (size_t i = 0; i < items.size(); ++i) {
+ const auto &item = items[i];
+ BloomHashval h = bloomCalcHash64(item.data(), static_cast<int>(item.size()));
+ // check
+ auto batch = storage_->GetWriteBatchBase(); // todo: as placeholder, change the function to delete this
+ for (int ii = sb_chain_metadata.n_filters - 1; ii >= 0; --ii) {
+ std::string bf_key;
+ appendBFSuffix(ns_key, ii, &bf_key);
Review Comment:
Can we avoid re-caculate this by `bf_metadata_list`? Seems it's calculated once
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
+ public:
+ uint16_t n_filters;
+ uint16_t options;
Review Comment:
After reviewing RedisBloom's code.
```C
/**
* Common function for adding one or more items to a bloom filter.
* capacity and error rate must not be 0.
*/
static SBChain *bfCreateChain(RedisModuleKey *key, double error_rate, size_t capacity,
unsigned expansion, unsigned scaling) {
SBChain *sb = SB_NewChain(capacity, error_rate, BLOOM_OPT_FORCE64 | scaling | BLOOM_OPT_NOROUND,
expansion);
if (sb != NULL) {
RedisModule_ModuleTypeSetValue(key, BFType, sb);
}
return sb;
}
```
It uses `options` for compability. I guess we don't need to keep all same as them(we don't have existing data). We can just simplify this options and bloom filter.
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
Review Comment:
We'd better add an version to metadata to support future evolution
##########
src/types/redis_bloomfilter.cc:
##########
@@ -0,0 +1,378 @@
+/*
+ * 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_bloomfilter.h"
+
+namespace redis {
+
+double BloomFilter::calcBpe(double error) {
+ static const double denom = 0.480453013918201; // ln(2)^2
+ double num = log(error);
+
+ double bpe = -(num / denom);
+ if (bpe < 0) {
+ bpe = -bpe;
+ }
+ return bpe;
+}
+
+void BloomFilter::appendBFSuffix(const Slice &ns_key, uint16_t filters_index, std::string *output) {
+ output->clear();
+
+ output->append(ns_key.data(), ns_key.size());
+ output->append(kBloomFilterSeparator);
+ PutFixed16(output, filters_index);
+}
+
+void BloomFilter ::appendBFMetaSuffix(const Slice &bf_key, std::string *output) {
+ output->clear();
+
+ output->append(bf_key.data(), bf_key.size());
+ output->append(kBloomFilterSeparator);
+ output->append("meta");
+}
+
+void BloomFilter::bfInit(BFMetadata *bf_metadata, uint64_t entries, double error, unsigned int options) {
+ bf_metadata->error = error;
+ bf_metadata->bf_bits = 0;
+ bf_metadata->entries = entries;
+ bf_metadata->bpe = calcBpe(error);
+ bf_metadata->size = 0;
+
+ uint64_t bits = 0;
+
+ {
+ double bn2 = logb(static_cast<double>(entries) * bf_metadata->bpe);
+ // if (bn2 > 63 || bn2 == INFINITY) {
+ // return 1;
+ // }
+ bf_metadata->n2 = static_cast<uint8_t>(bn2) + 1;
+ bits = 1LLU << bf_metadata->n2;
+
+ // Determine the number of extra bits available for more items. We rounded
+ // up the number of bits to the next-highest power of two. This means we
+ // might have up to 2x the bits available to us.
+ uint64_t bit_diff = bits - static_cast<uint64_t>(static_cast<double>(entries) * bf_metadata->bpe);
+ // The number of additional items we can store is the extra number of bits
+ // divided by bits-per-element
+ auto item_diff = static_cast<uint64_t>(static_cast<double>(bit_diff) / bf_metadata->bpe);
+ bf_metadata->entries += item_diff;
+ }
+
+ if (bits % 64) {
+ bf_metadata->bf_bytes = ((bits / 64) + 1) * 8;
+ } else {
+ bf_metadata->bf_bytes = bits / 8;
+ }
+ bf_metadata->bf_bits = bf_metadata->bf_bytes * 8;
+
+ bf_metadata->hashes = (int)ceil(0.693147180559945 * bf_metadata->bpe); // ln(2)
+}
+
+BloomHashval BloomFilter::bloomCalcHash64(const void *buffer, int len) {
+ BloomHashval rv;
+ rv.a = MurmurHash64A_Bloom(buffer, len, 0xc6a4a7935bd1e995ULL);
+ rv.b = MurmurHash64A_Bloom(buffer, len, rv.a);
+ return rv;
+}
+
+int BloomFilter::testBitSetBit(std::string &value, uint64_t x, ReadWriteMode mode) {
+ uint8_t mask = 1 << (x % 8);
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ uint8_t c = value[byte_index]; // expensive memory access
+
+ if (c & mask) {
+ return 1;
+ } else {
+ if (mode == ReadWriteMode::MODE_WRITE) {
+ uint8_t s = c | mask;
+ value[byte_index] = static_cast<char>(s);
+ }
+ return 0;
+ }
+}
+
+rocksdb::Status BloomFilter::getSBChainMetadata(const Slice &ns_key, SBChainMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, ns_key, metadata);
+}
+
+rocksdb::Status BloomFilter::getBFMetadata(const Slice &bf_meta_key, BFMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, bf_meta_key, metadata);
+}
+
+rocksdb::Status BloomFilter::createSBChain(const Slice &ns_key, double error_rate, uint64_t capacity,
+ uint16_t expansion, uint16_t scaling, SBChainMetadata &sb_chain_metadata) {
+ sb_chain_metadata.n_filters = 1;
+ sb_chain_metadata.growth = expansion;
+ sb_chain_metadata.options = scaling;
+ sb_chain_metadata.size = 0;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createSBChain"});
+ batch->PutLogData(log_data.Encode());
+
+ std::string sb_chain_bytes;
+ sb_chain_metadata.Encode(&sb_chain_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_bytes);
+
+ std::string bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ bfInit(&bf_metadata, capacity, error_rate, scaling);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::createBloomFilter(const Slice &ns_key, SBChainMetadata &sb_chain_metadata,
+ BFMetadata &bf_metadata, std::string &bf_key) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createBloomFilter"});
+ batch->PutLogData(log_data.Encode());
+
+ BFMetadata old_bf_metadata = bf_metadata;
+ sb_chain_metadata.n_filters += 1;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ bfInit(&bf_metadata, sb_chain_metadata.growth * old_bf_metadata.entries,
+ old_bf_metadata.error * kErrorTighteningRatio, sb_chain_metadata.options);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::bloomCheckAdd64(const Slice &bf_key, BFMetadata &bf_metadata, BloomHashval hashval,
+ ReadWriteMode mode, ObserverOrUniquePtr<rocksdb::WriteBatchBase> &batch,
+ int &ret) {
+ int found_unset = 0;
+ const uint64_t mod = (1LLU << bf_metadata.n2);
+ for (uint64_t i = 0; i < bf_metadata.hashes; i++) {
+ uint64_t x = ((hashval.a + i * hashval.b)) % mod;
+ uint32_t index = (x / kBloomFilterSegmentBits) * kBloomFilterSegmentBytes;
+ std::string sub_key, value;
+ InternalKey(bf_key, std::to_string(index), bf_metadata.version, storage_->IsSlotIdEncoded()).Encode(&sub_key);
+ rocksdb::Status s =
+ storage_->Get(rocksdb::ReadOptions(), sub_key, &value); // todo: use bf_key in InternalKey maybe error
+ if (!s.ok() && !s.IsNotFound()) return s;
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ if (byte_index >= value.size()) { // expand the bloomfilter
+ size_t expand_size = 0;
+ if (byte_index >= value.size() * 2) {
+ expand_size = byte_index - value.size() + 1;
+ } else if (value.size() * 2 > kBloomFilterSegmentBytes) {
+ expand_size = kBloomFilterSegmentBytes - value.size();
+ } else {
+ expand_size = value.size();
+ }
+ value.append(expand_size, 0);
Review Comment:
So, currently, we'll try our best to saving memories, avoid too large value?
##########
src/storage/redis_metadata.cc:
##########
@@ -401,3 +401,85 @@ rocksdb::Status StreamMetadata::Decode(const std::string &bytes) {
return rocksdb::Status::OK();
}
+
+void SBChainMetadata::Encode(std::string *dst) {
+ Metadata::Encode(dst);
+
+ PutFixed16(dst, n_filters);
+ PutFixed16(dst, options);
+ PutFixed16(dst, growth);
+}
+
+rocksdb::Status SBChainMetadata::Decode(const std::string &bytes) {
+ Slice input(bytes);
+ if (!GetFixed8(&input, &flags)) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+ if (!GetExpire(&input)) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ if (input.size() < 8 + CommonEncodedSize()) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ GetFixed64(&input, &version);
+ GetFixedCommon(&input, &size);
+
+ if (input.size() < 6) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ GetFixed16(&input, &n_filters);
+ GetFixed16(&input, &options);
+ GetFixed16(&input, &growth);
+
+ return rocksdb::Status::OK();
+}
+
+void BFMetadata::Encode(std::string *dst) {
+ Metadata::Encode(dst);
+
+ PutFixed32(dst, hashes);
+ PutFixed8(dst, n2);
+ PutFixed64(dst, entries);
+
+ PutDouble(dst, error);
+ PutDouble(dst, bpe);
+
+ PutFixed64(dst, bf_bits);
+ PutFixed64(dst, bf_bytes);
+}
+
+rocksdb::Status BFMetadata::Decode(const std::string &bytes) {
+ Slice input(bytes);
+ if (!GetFixed8(&input, &flags)) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+ if (!GetExpire(&input)) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ if (input.size() < 8 + CommonEncodedSize()) {
+ return rocksdb::Status::InvalidArgument(kErrMetadataTooShort);
+ }
+
+ GetFixed64(&input, &version);
+ GetFixedCommon(&input, &size);
+
+ if (input.size() < 45) {
Review Comment:
Prefer constexpr than magic numbers.
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
+ public:
+ uint16_t n_filters;
+ uint16_t options;
+ uint16_t growth;
+
+ explicit SBChainMetadata(bool generate_version = true) : Metadata(kRedisBloomFilter, generate_version) {}
+
+ void Encode(std::string *dst) override;
+ rocksdb::Status Decode(const std::string &bytes) override;
+};
+
+class BFMetadata : public Metadata {
+ public:
Review Comment:
As a simple implemention. We can use an aligned bloom filter. And keep only:
```C
class BFMetadata : public Metadata {
public:
uint32_t hashes;
uint64_t entries;
double error;
double bpe;
uint64_t bf_bytes;
};
```
We can even implement it easier using a BlockSplitBloomFilter.
##########
src/types/redis_bloomfilter.cc:
##########
@@ -0,0 +1,378 @@
+/*
+ * 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_bloomfilter.h"
+
+namespace redis {
+
+double BloomFilter::calcBpe(double error) {
+ static const double denom = 0.480453013918201; // ln(2)^2
+ double num = log(error);
+
+ double bpe = -(num / denom);
+ if (bpe < 0) {
+ bpe = -bpe;
+ }
+ return bpe;
+}
+
+void BloomFilter::appendBFSuffix(const Slice &ns_key, uint16_t filters_index, std::string *output) {
+ output->clear();
+
+ output->append(ns_key.data(), ns_key.size());
+ output->append(kBloomFilterSeparator);
+ PutFixed16(output, filters_index);
+}
+
+void BloomFilter ::appendBFMetaSuffix(const Slice &bf_key, std::string *output) {
+ output->clear();
+
+ output->append(bf_key.data(), bf_key.size());
+ output->append(kBloomFilterSeparator);
+ output->append("meta");
+}
+
+void BloomFilter::bfInit(BFMetadata *bf_metadata, uint64_t entries, double error, unsigned int options) {
+ bf_metadata->error = error;
+ bf_metadata->bf_bits = 0;
+ bf_metadata->entries = entries;
+ bf_metadata->bpe = calcBpe(error);
+ bf_metadata->size = 0;
+
+ uint64_t bits = 0;
+
+ {
+ double bn2 = logb(static_cast<double>(entries) * bf_metadata->bpe);
+ // if (bn2 > 63 || bn2 == INFINITY) {
+ // return 1;
+ // }
+ bf_metadata->n2 = static_cast<uint8_t>(bn2) + 1;
+ bits = 1LLU << bf_metadata->n2;
+
+ // Determine the number of extra bits available for more items. We rounded
+ // up the number of bits to the next-highest power of two. This means we
+ // might have up to 2x the bits available to us.
+ uint64_t bit_diff = bits - static_cast<uint64_t>(static_cast<double>(entries) * bf_metadata->bpe);
+ // The number of additional items we can store is the extra number of bits
+ // divided by bits-per-element
+ auto item_diff = static_cast<uint64_t>(static_cast<double>(bit_diff) / bf_metadata->bpe);
+ bf_metadata->entries += item_diff;
+ }
+
+ if (bits % 64) {
+ bf_metadata->bf_bytes = ((bits / 64) + 1) * 8;
+ } else {
+ bf_metadata->bf_bytes = bits / 8;
+ }
+ bf_metadata->bf_bits = bf_metadata->bf_bytes * 8;
+
+ bf_metadata->hashes = (int)ceil(0.693147180559945 * bf_metadata->bpe); // ln(2)
+}
+
+BloomHashval BloomFilter::bloomCalcHash64(const void *buffer, int len) {
+ BloomHashval rv;
+ rv.a = MurmurHash64A_Bloom(buffer, len, 0xc6a4a7935bd1e995ULL);
+ rv.b = MurmurHash64A_Bloom(buffer, len, rv.a);
+ return rv;
+}
+
+int BloomFilter::testBitSetBit(std::string &value, uint64_t x, ReadWriteMode mode) {
+ uint8_t mask = 1 << (x % 8);
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ uint8_t c = value[byte_index]; // expensive memory access
+
+ if (c & mask) {
+ return 1;
+ } else {
+ if (mode == ReadWriteMode::MODE_WRITE) {
+ uint8_t s = c | mask;
+ value[byte_index] = static_cast<char>(s);
+ }
+ return 0;
+ }
+}
+
+rocksdb::Status BloomFilter::getSBChainMetadata(const Slice &ns_key, SBChainMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, ns_key, metadata);
+}
+
+rocksdb::Status BloomFilter::getBFMetadata(const Slice &bf_meta_key, BFMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, bf_meta_key, metadata);
+}
+
+rocksdb::Status BloomFilter::createSBChain(const Slice &ns_key, double error_rate, uint64_t capacity,
+ uint16_t expansion, uint16_t scaling, SBChainMetadata &sb_chain_metadata) {
+ sb_chain_metadata.n_filters = 1;
+ sb_chain_metadata.growth = expansion;
+ sb_chain_metadata.options = scaling;
+ sb_chain_metadata.size = 0;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createSBChain"});
+ batch->PutLogData(log_data.Encode());
+
+ std::string sb_chain_bytes;
+ sb_chain_metadata.Encode(&sb_chain_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_bytes);
+
+ std::string bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ bfInit(&bf_metadata, capacity, error_rate, scaling);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::createBloomFilter(const Slice &ns_key, SBChainMetadata &sb_chain_metadata,
+ BFMetadata &bf_metadata, std::string &bf_key) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createBloomFilter"});
+ batch->PutLogData(log_data.Encode());
+
+ BFMetadata old_bf_metadata = bf_metadata;
+ sb_chain_metadata.n_filters += 1;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ bfInit(&bf_metadata, sb_chain_metadata.growth * old_bf_metadata.entries,
+ old_bf_metadata.error * kErrorTighteningRatio, sb_chain_metadata.options);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::bloomCheckAdd64(const Slice &bf_key, BFMetadata &bf_metadata, BloomHashval hashval,
+ ReadWriteMode mode, ObserverOrUniquePtr<rocksdb::WriteBatchBase> &batch,
+ int &ret) {
+ int found_unset = 0;
+ const uint64_t mod = (1LLU << bf_metadata.n2);
+ for (uint64_t i = 0; i < bf_metadata.hashes; i++) {
+ uint64_t x = ((hashval.a + i * hashval.b)) % mod;
+ uint32_t index = (x / kBloomFilterSegmentBits) * kBloomFilterSegmentBytes;
+ std::string sub_key, value;
+ InternalKey(bf_key, std::to_string(index), bf_metadata.version, storage_->IsSlotIdEncoded()).Encode(&sub_key);
+ rocksdb::Status s =
+ storage_->Get(rocksdb::ReadOptions(), sub_key, &value); // todo: use bf_key in InternalKey maybe error
+ if (!s.ok() && !s.IsNotFound()) return s;
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ if (byte_index >= value.size()) { // expand the bloomfilter
+ size_t expand_size = 0;
+ if (byte_index >= value.size() * 2) {
+ expand_size = byte_index - value.size() + 1;
+ } else if (value.size() * 2 > kBloomFilterSegmentBytes) {
+ expand_size = kBloomFilterSegmentBytes - value.size();
+ } else {
+ expand_size = value.size();
+ }
+ value.append(expand_size, 0);
+ }
+
+ if (!testBitSetBit(value, x, mode)) {
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 0;
+ return rocksdb::Status::OK();
+ } else {
+ found_unset = 1;
+ batch->Put(sub_key, value);
+ storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+ // todo: if the indexs are same, the value need to be updated immediately, if have other ways to compl it
+ }
+ }
+ }
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 1;
+ } else {
+ ret = found_unset;
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status BloomFilter::Reserve(const Slice &user_key, double error_rate, uint64_t capacity, uint16_t expansion,
+ uint16_t scaling) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("item exists");
+ }
+
+ return createSBChain(ns_key, error_rate, capacity, expansion, scaling, sb_chain_metadata);
+}
+
+rocksdb::Status BloomFilter::Add(const Slice &user_key, const Slice &item, int &ret) {
+ std::vector<int> tmp(1, 0);
+ rocksdb::Status s = MAdd(user_key, {item}, tmp);
+ ret = tmp[0];
+ return s;
+}
+
+rocksdb::Status BloomFilter::MAdd(const Slice &user_key, const std::vector<Slice> &items, std::vector<int> &rets) {
+ BFInsertOptions insert_options = {.capacity = kBFDefaultInitCapacity,
+ .error_rate = kBFDefaultErrorRate,
+ .autocreate = 1,
+ .expansion = kBFDefaultExpansion,
+ .scaling = 1};
+ return InsertCommon(user_key, items, &insert_options, rets);
+}
+
+rocksdb::Status BloomFilter::InsertCommon(const Slice &user_key, const std::vector<Slice> &items,
+ const BFInsertOptions *insert_options, std::vector<int> &rets) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+
+ if (s.IsNotFound() && insert_options->autocreate) {
+ rocksdb::Status create_s = createSBChain(ns_key, insert_options->error_rate, insert_options->capacity,
+ insert_options->expansion, insert_options->scaling, sb_chain_metadata);
+ if (!create_s.ok()) return create_s;
+ } else if (!s.ok()) {
+ return s;
+ }
+
+ std::vector<BFMetadata> bf_metadata_list;
+ for (u_int16_t i = 0; i < sb_chain_metadata.n_filters; ++i) {
+ std::string bf_key, bf_meta_key;
+ appendBFSuffix(ns_key, i, &bf_key);
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ s = getBFMetadata(bf_meta_key, &bf_metadata);
+ if (!s.ok()) return s;
+ bf_metadata_list.push_back(bf_metadata);
+ }
+
+ BFMetadata cur_bf_metadata = bf_metadata_list.back();
+ std::string cur_bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &cur_bf_key);
+ if (!sb_chain_metadata.options && cur_bf_metadata.size + items.size() > cur_bf_metadata.entries) {
+ return rocksdb::Status::Aborted("filter is full and is nonscaling");
+ }
+
+ int suc_insert = 0;
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"insert"});
+ batch->PutLogData(log_data.Encode());
+ for (size_t i = 0; i < items.size(); ++i) {
+ bool item_exist = false;
+ const auto &item = items[i];
+ BloomHashval h = bloomCalcHash64(item.data(), static_cast<int>(item.size()));
+
+ // check
+ for (int ii = sb_chain_metadata.n_filters - 1; ii >= 0; --ii) {
+ std::string bf_key;
+ appendBFSuffix(ns_key, ii, &bf_key);
+ int ret = 0;
+ s = bloomCheckAdd64(bf_key, bf_metadata_list[ii], h, ReadWriteMode::MODE_READ, batch, ret);
+ if (ret == 1) {
+ item_exist = true;
+ break;
+ }
+ }
+
+ // insert
+ if (!item_exist) {
+ if (cur_bf_metadata.size >= cur_bf_metadata.entries) {
+ std::string old_bf_metadata_bytes, cur_bf_meta_key;
+ cur_bf_metadata.Encode(&old_bf_metadata_bytes);
+ appendBFMetaSuffix(cur_bf_key, &cur_bf_meta_key);
+ batch->Put(metadata_cf_handle_, cur_bf_meta_key, old_bf_metadata_bytes);
+ createBloomFilter(ns_key, sb_chain_metadata, cur_bf_metadata,
+ cur_bf_key); // todo: the scaling function need to test
+ }
+ int ret = 0;
+ s = bloomCheckAdd64(cur_bf_key, cur_bf_metadata, h, ReadWriteMode::MODE_WRITE, batch, ret);
+ rets[i] = ret;
+ suc_insert += ret;
+ cur_bf_metadata.size += ret;
+ }
+ }
+ sb_chain_metadata.size += suc_insert;
+
+ std::string cur_bf_metadata_bytes, cur_bf_meta_key, sb_chain_metadata_bytes;
+ cur_bf_metadata.Encode(&cur_bf_metadata_bytes);
+ sb_chain_metadata.Encode(&sb_chain_metadata_bytes);
+ appendBFMetaSuffix(cur_bf_key, &cur_bf_meta_key);
+ batch->Put(metadata_cf_handle_, cur_bf_meta_key, cur_bf_metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_metadata_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::Exist(const Slice &user_key, const Slice &item, int &ret) {
+ std::vector<int> tmp(1, 0);
+ rocksdb::Status s = MExist(user_key, {item}, tmp);
+ ret = tmp[0];
+ return s;
+}
+
+rocksdb::Status BloomFilter::MExist(const Slice &user_key, const std::vector<Slice> &items, std::vector<int> &rets) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok()) return s;
+
+ std::vector<BFMetadata> bf_metadata_list;
+ for (u_int16_t i = 0; i < sb_chain_metadata.n_filters; ++i) {
Review Comment:
```suggestion
for (uint16_t i = 0; i < sb_chain_metadata.n_filters; ++i) {
```
It's this a typo?
##########
src/types/redis_bloomfilter.cc:
##########
@@ -0,0 +1,378 @@
+/*
+ * 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_bloomfilter.h"
+
+namespace redis {
+
+double BloomFilter::calcBpe(double error) {
+ static const double denom = 0.480453013918201; // ln(2)^2
+ double num = log(error);
+
+ double bpe = -(num / denom);
+ if (bpe < 0) {
+ bpe = -bpe;
+ }
+ return bpe;
+}
+
+void BloomFilter::appendBFSuffix(const Slice &ns_key, uint16_t filters_index, std::string *output) {
+ output->clear();
+
+ output->append(ns_key.data(), ns_key.size());
+ output->append(kBloomFilterSeparator);
+ PutFixed16(output, filters_index);
+}
+
+void BloomFilter ::appendBFMetaSuffix(const Slice &bf_key, std::string *output) {
+ output->clear();
+
+ output->append(bf_key.data(), bf_key.size());
+ output->append(kBloomFilterSeparator);
+ output->append("meta");
+}
+
+void BloomFilter::bfInit(BFMetadata *bf_metadata, uint64_t entries, double error, unsigned int options) {
+ bf_metadata->error = error;
+ bf_metadata->bf_bits = 0;
+ bf_metadata->entries = entries;
+ bf_metadata->bpe = calcBpe(error);
+ bf_metadata->size = 0;
+
+ uint64_t bits = 0;
+
+ {
+ double bn2 = logb(static_cast<double>(entries) * bf_metadata->bpe);
+ // if (bn2 > 63 || bn2 == INFINITY) {
+ // return 1;
+ // }
+ bf_metadata->n2 = static_cast<uint8_t>(bn2) + 1;
+ bits = 1LLU << bf_metadata->n2;
+
+ // Determine the number of extra bits available for more items. We rounded
+ // up the number of bits to the next-highest power of two. This means we
+ // might have up to 2x the bits available to us.
+ uint64_t bit_diff = bits - static_cast<uint64_t>(static_cast<double>(entries) * bf_metadata->bpe);
+ // The number of additional items we can store is the extra number of bits
+ // divided by bits-per-element
+ auto item_diff = static_cast<uint64_t>(static_cast<double>(bit_diff) / bf_metadata->bpe);
+ bf_metadata->entries += item_diff;
+ }
+
+ if (bits % 64) {
+ bf_metadata->bf_bytes = ((bits / 64) + 1) * 8;
+ } else {
+ bf_metadata->bf_bytes = bits / 8;
+ }
+ bf_metadata->bf_bits = bf_metadata->bf_bytes * 8;
+
+ bf_metadata->hashes = (int)ceil(0.693147180559945 * bf_metadata->bpe); // ln(2)
+}
+
+BloomHashval BloomFilter::bloomCalcHash64(const void *buffer, int len) {
+ BloomHashval rv;
+ rv.a = MurmurHash64A_Bloom(buffer, len, 0xc6a4a7935bd1e995ULL);
+ rv.b = MurmurHash64A_Bloom(buffer, len, rv.a);
+ return rv;
+}
+
+int BloomFilter::testBitSetBit(std::string &value, uint64_t x, ReadWriteMode mode) {
+ uint8_t mask = 1 << (x % 8);
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ uint8_t c = value[byte_index]; // expensive memory access
+
+ if (c & mask) {
+ return 1;
+ } else {
+ if (mode == ReadWriteMode::MODE_WRITE) {
+ uint8_t s = c | mask;
+ value[byte_index] = static_cast<char>(s);
+ }
+ return 0;
+ }
+}
+
+rocksdb::Status BloomFilter::getSBChainMetadata(const Slice &ns_key, SBChainMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, ns_key, metadata);
+}
+
+rocksdb::Status BloomFilter::getBFMetadata(const Slice &bf_meta_key, BFMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, bf_meta_key, metadata);
+}
+
+rocksdb::Status BloomFilter::createSBChain(const Slice &ns_key, double error_rate, uint64_t capacity,
+ uint16_t expansion, uint16_t scaling, SBChainMetadata &sb_chain_metadata) {
+ sb_chain_metadata.n_filters = 1;
+ sb_chain_metadata.growth = expansion;
+ sb_chain_metadata.options = scaling;
+ sb_chain_metadata.size = 0;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createSBChain"});
+ batch->PutLogData(log_data.Encode());
+
+ std::string sb_chain_bytes;
+ sb_chain_metadata.Encode(&sb_chain_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_bytes);
+
+ std::string bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ bfInit(&bf_metadata, capacity, error_rate, scaling);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::createBloomFilter(const Slice &ns_key, SBChainMetadata &sb_chain_metadata,
+ BFMetadata &bf_metadata, std::string &bf_key) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createBloomFilter"});
+ batch->PutLogData(log_data.Encode());
+
+ BFMetadata old_bf_metadata = bf_metadata;
+ sb_chain_metadata.n_filters += 1;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ bfInit(&bf_metadata, sb_chain_metadata.growth * old_bf_metadata.entries,
+ old_bf_metadata.error * kErrorTighteningRatio, sb_chain_metadata.options);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::bloomCheckAdd64(const Slice &bf_key, BFMetadata &bf_metadata, BloomHashval hashval,
+ ReadWriteMode mode, ObserverOrUniquePtr<rocksdb::WriteBatchBase> &batch,
+ int &ret) {
+ int found_unset = 0;
+ const uint64_t mod = (1LLU << bf_metadata.n2);
+ for (uint64_t i = 0; i < bf_metadata.hashes; i++) {
+ uint64_t x = ((hashval.a + i * hashval.b)) % mod;
+ uint32_t index = (x / kBloomFilterSegmentBits) * kBloomFilterSegmentBytes;
+ std::string sub_key, value;
+ InternalKey(bf_key, std::to_string(index), bf_metadata.version, storage_->IsSlotIdEncoded()).Encode(&sub_key);
+ rocksdb::Status s =
+ storage_->Get(rocksdb::ReadOptions(), sub_key, &value); // todo: use bf_key in InternalKey maybe error
+ if (!s.ok() && !s.IsNotFound()) return s;
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ if (byte_index >= value.size()) { // expand the bloomfilter
+ size_t expand_size = 0;
+ if (byte_index >= value.size() * 2) {
+ expand_size = byte_index - value.size() + 1;
+ } else if (value.size() * 2 > kBloomFilterSegmentBytes) {
+ expand_size = kBloomFilterSegmentBytes - value.size();
+ } else {
+ expand_size = value.size();
+ }
+ value.append(expand_size, 0);
+ }
+
+ if (!testBitSetBit(value, x, mode)) {
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 0;
+ return rocksdb::Status::OK();
+ } else {
+ found_unset = 1;
+ batch->Put(sub_key, value);
+ storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+ // todo: if the indexs are same, the value need to be updated immediately, if have other ways to compl it
+ }
+ }
+ }
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 1;
+ } else {
+ ret = found_unset;
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status BloomFilter::Reserve(const Slice &user_key, double error_rate, uint64_t capacity, uint16_t expansion,
+ uint16_t scaling) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("item exists");
+ }
+
+ return createSBChain(ns_key, error_rate, capacity, expansion, scaling, sb_chain_metadata);
+}
+
+rocksdb::Status BloomFilter::Add(const Slice &user_key, const Slice &item, int &ret) {
+ std::vector<int> tmp(1, 0);
+ rocksdb::Status s = MAdd(user_key, {item}, tmp);
+ ret = tmp[0];
+ return s;
+}
+
+rocksdb::Status BloomFilter::MAdd(const Slice &user_key, const std::vector<Slice> &items, std::vector<int> &rets) {
+ BFInsertOptions insert_options = {.capacity = kBFDefaultInitCapacity,
+ .error_rate = kBFDefaultErrorRate,
+ .autocreate = 1,
+ .expansion = kBFDefaultExpansion,
+ .scaling = 1};
+ return InsertCommon(user_key, items, &insert_options, rets);
+}
+
+rocksdb::Status BloomFilter::InsertCommon(const Slice &user_key, const std::vector<Slice> &items,
+ const BFInsertOptions *insert_options, std::vector<int> &rets) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+
+ if (s.IsNotFound() && insert_options->autocreate) {
+ rocksdb::Status create_s = createSBChain(ns_key, insert_options->error_rate, insert_options->capacity,
+ insert_options->expansion, insert_options->scaling, sb_chain_metadata);
+ if (!create_s.ok()) return create_s;
+ } else if (!s.ok()) {
+ return s;
+ }
+
+ std::vector<BFMetadata> bf_metadata_list;
+ for (u_int16_t i = 0; i < sb_chain_metadata.n_filters; ++i) {
+ std::string bf_key, bf_meta_key;
+ appendBFSuffix(ns_key, i, &bf_key);
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ s = getBFMetadata(bf_meta_key, &bf_metadata);
+ if (!s.ok()) return s;
+ bf_metadata_list.push_back(bf_metadata);
+ }
+
+ BFMetadata cur_bf_metadata = bf_metadata_list.back();
+ std::string cur_bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &cur_bf_key);
+ if (!sb_chain_metadata.options && cur_bf_metadata.size + items.size() > cur_bf_metadata.entries) {
+ return rocksdb::Status::Aborted("filter is full and is nonscaling");
+ }
+
+ int suc_insert = 0;
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"insert"});
+ batch->PutLogData(log_data.Encode());
+ for (size_t i = 0; i < items.size(); ++i) {
+ bool item_exist = false;
+ const auto &item = items[i];
+ BloomHashval h = bloomCalcHash64(item.data(), static_cast<int>(item.size()));
+
+ // check
+ for (int ii = sb_chain_metadata.n_filters - 1; ii >= 0; --ii) {
+ std::string bf_key;
+ appendBFSuffix(ns_key, ii, &bf_key);
+ int ret = 0;
+ s = bloomCheckAdd64(bf_key, bf_metadata_list[ii], h, ReadWriteMode::MODE_READ, batch, ret);
+ if (ret == 1) {
+ item_exist = true;
+ break;
+ }
+ }
+
+ // insert
+ if (!item_exist) {
+ if (cur_bf_metadata.size >= cur_bf_metadata.entries) {
+ std::string old_bf_metadata_bytes, cur_bf_meta_key;
+ cur_bf_metadata.Encode(&old_bf_metadata_bytes);
+ appendBFMetaSuffix(cur_bf_key, &cur_bf_meta_key);
+ batch->Put(metadata_cf_handle_, cur_bf_meta_key, old_bf_metadata_bytes);
+ createBloomFilter(ns_key, sb_chain_metadata, cur_bf_metadata,
+ cur_bf_key); // todo: the scaling function need to test
+ }
+ int ret = 0;
+ s = bloomCheckAdd64(cur_bf_key, cur_bf_metadata, h, ReadWriteMode::MODE_WRITE, batch, ret);
+ rets[i] = ret;
+ suc_insert += ret;
+ cur_bf_metadata.size += ret;
+ }
+ }
+ sb_chain_metadata.size += suc_insert;
+
+ std::string cur_bf_metadata_bytes, cur_bf_meta_key, sb_chain_metadata_bytes;
+ cur_bf_metadata.Encode(&cur_bf_metadata_bytes);
+ sb_chain_metadata.Encode(&sb_chain_metadata_bytes);
+ appendBFMetaSuffix(cur_bf_key, &cur_bf_meta_key);
+ batch->Put(metadata_cf_handle_, cur_bf_meta_key, cur_bf_metadata_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_metadata_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::Exist(const Slice &user_key, const Slice &item, int &ret) {
+ std::vector<int> tmp(1, 0);
+ rocksdb::Status s = MExist(user_key, {item}, tmp);
+ ret = tmp[0];
+ return s;
+}
+
+rocksdb::Status BloomFilter::MExist(const Slice &user_key, const std::vector<Slice> &items, std::vector<int> &rets) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok()) return s;
+
+ std::vector<BFMetadata> bf_metadata_list;
Review Comment:
Can be
```suggestion
std::vector<BFMetadata> bf_metadata_list;
bf_metadata_list.reserve(sb_chain_metadata.n_filters);
```
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] git-hulk commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1297402857
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
Review Comment:
Not sure what's SB short for which words.
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1285158127
##########
src/types/redis_bloomfilter.h:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#pragma once
+
+#include <common/murmurhash2.h>
+
+#include "storage/redis_db.h"
+#include "storage/redis_metadata.h"
+
+namespace redis {
+
+const char kBloomFilterSeparator[] = ":";
+const uint32_t kBloomFilterSegmentBits = 1024 * 8;
+const uint32_t kBloomFilterSegmentBytes = 1024;
+const uint64_t kBFDefaultInitCapacity = 100;
+const double kBFDefaultErrorRate = 0.01;
+const uint16_t kBFDefaultExpansion = 2;
+const double kErrorTighteningRatio = 0.5;
+
+struct BloomHashval {
+ uint64_t a;
+ uint64_t b;
+};
+
+struct BFInsertOptions {
Review Comment:
```
struct BFInsertOptions {
uint64_t capacity = kBFDefaultInitCapacity;
double error_rate = kBFDefaultErrorRate;
...
```
Nit: I guess provide default values would be better here.
##########
src/common/murmurhash2.h:
##########
@@ -0,0 +1,22 @@
+//-----------------------------------------------------------------------------
+// MurmurHash2 was written by Austin Appleby, and is placed in the public
+// domain. The author hereby disclaims copyright to this source code.
Review Comment:
This is nice, but should we vendor this in thirdparty library or vendor at somewhere? @PragmaTwice
##########
src/commands/cmd_bloomfilter.cc:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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 "commander.h"
+#include "error_constants.h"
+#include "server/server.h"
+#include "types/redis_bloomfilter.h"
+
+namespace redis {
+
+class CommandBFReserve : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ auto parse_error_rate = ParseFloat<double>(args[2]);
+ if (!parse_error_rate) {
+ return {Status::RedisParseErr, errValueIsNotFloat};
+ }
+ error_rate_ = *parse_error_rate;
+ if (error_rate_ >= 1 || error_rate_ <= 0) {
+ return {Status::RedisParseErr, "0 < error rate range < 1"};
+ }
+
+ auto parse_capacity = ParseInt<uint64_t>(args[3], 10);
+ if (!parse_capacity) {
+ return {Status::RedisParseErr, errValueNotInteger};
+ }
+ capacity_ = *parse_capacity;
+ if (capacity_ <= 0) {
+ return {Status::RedisParseErr, "capacity should be larger than 0"};
+ }
+
+ size_t expect_size = 4;
+ auto iter = find(args.begin(), args.end(), "NONSCALING");
Review Comment:
```
while (parser.Good()) {
if (parser.EatEqICaseFlag("NONSCALING", ...)) {
...
} else {
return parser.InvalidSyntax();
}
}
```
Using parser like this might helps
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] PragmaTwice commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "PragmaTwice (via GitHub)" <gi...@apache.org>.
PragmaTwice commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1286436864
##########
src/common/murmurhash2.h:
##########
@@ -0,0 +1,22 @@
+//-----------------------------------------------------------------------------
+// MurmurHash2 was written by Austin Appleby, and is placed in the public
+// domain. The author hereby disclaims copyright to this source code.
Review Comment:
I think currently it is ok to put it in `common/`, but we may move it to a vendor directory later.
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#issuecomment-1680767299
This commit ( https://github.com/apache/kvrocks/pull/1632/commits/acf50104fbbecc84e363dfd88f750bf52c46f0e7 ) looks ok, but would you mind create an single patch for it?
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#issuecomment-1695427871
Close as doing in other patch
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1285158127
##########
src/types/redis_bloomfilter.h:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#pragma once
+
+#include <common/murmurhash2.h>
+
+#include "storage/redis_db.h"
+#include "storage/redis_metadata.h"
+
+namespace redis {
+
+const char kBloomFilterSeparator[] = ":";
+const uint32_t kBloomFilterSegmentBits = 1024 * 8;
+const uint32_t kBloomFilterSegmentBytes = 1024;
+const uint64_t kBFDefaultInitCapacity = 100;
+const double kBFDefaultErrorRate = 0.01;
+const uint16_t kBFDefaultExpansion = 2;
+const double kErrorTighteningRatio = 0.5;
+
+struct BloomHashval {
+ uint64_t a;
+ uint64_t b;
+};
+
+struct BFInsertOptions {
Review Comment:
```
struct BFInsertOptions {
uint64_t capacity = kBFDefaultInitCapacity;
double error_rate = kBFDefaultErrorRate;
...
```
Nit: I guess provide default values would be better here. C might not have default argument. But we can make full use of it.
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1286585226
##########
src/server/redis_reply.h:
##########
@@ -40,6 +40,15 @@ std::string Integer(T data) {
return ":" + std::to_string(data) + CRLF;
}
+template <typename T, std::enable_if_t<std::is_integral_v<T>, int> = 0>
+std::string MultiInteger(std::vector<T> &multi_data) {
Review Comment:
```suggestion
std::string MultiInteger(const std::vector<T> &multi_data) {
```
##########
src/types/redis_bloomfilter.cc:
##########
@@ -0,0 +1,378 @@
+/*
+ * 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_bloomfilter.h"
+
+namespace redis {
+
+double BloomFilter::calcBpe(double error) {
+ static const double denom = 0.480453013918201; // ln(2)^2
+ double num = log(error);
+
+ double bpe = -(num / denom);
+ if (bpe < 0) {
+ bpe = -bpe;
+ }
+ return bpe;
+}
+
+void BloomFilter::appendBFSuffix(const Slice &ns_key, uint16_t filters_index, std::string *output) {
+ output->clear();
+
+ output->append(ns_key.data(), ns_key.size());
+ output->append(kBloomFilterSeparator);
+ PutFixed16(output, filters_index);
+}
+
+void BloomFilter ::appendBFMetaSuffix(const Slice &bf_key, std::string *output) {
+ output->clear();
+
+ output->append(bf_key.data(), bf_key.size());
+ output->append(kBloomFilterSeparator);
+ output->append("meta");
+}
+
+void BloomFilter::bfInit(BFMetadata *bf_metadata, uint64_t entries, double error, unsigned int options) {
Review Comment:
`unsigned int` would better be `uint32_t`
##########
src/storage/redis_metadata.h:
##########
@@ -143,6 +144,7 @@ class Metadata {
virtual void Encode(std::string *dst);
virtual rocksdb::Status Decode(const std::string &bytes);
bool operator==(const Metadata &that) const;
+ virtual ~Metadata() = default;
Review Comment:
Nice catch! cc @git-hulk seems we have this for a long time?
##########
src/types/redis_bloomfilter.cc:
##########
@@ -0,0 +1,378 @@
+/*
+ * 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_bloomfilter.h"
+
+namespace redis {
+
+double BloomFilter::calcBpe(double error) {
+ static const double denom = 0.480453013918201; // ln(2)^2
+ double num = log(error);
+
+ double bpe = -(num / denom);
+ if (bpe < 0) {
+ bpe = -bpe;
+ }
+ return bpe;
+}
+
+void BloomFilter::appendBFSuffix(const Slice &ns_key, uint16_t filters_index, std::string *output) {
+ output->clear();
+
+ output->append(ns_key.data(), ns_key.size());
+ output->append(kBloomFilterSeparator);
+ PutFixed16(output, filters_index);
+}
+
+void BloomFilter ::appendBFMetaSuffix(const Slice &bf_key, std::string *output) {
+ output->clear();
+
+ output->append(bf_key.data(), bf_key.size());
+ output->append(kBloomFilterSeparator);
+ output->append("meta");
+}
+
+void BloomFilter::bfInit(BFMetadata *bf_metadata, uint64_t entries, double error, unsigned int options) {
+ bf_metadata->error = error;
+ bf_metadata->bf_bits = 0;
+ bf_metadata->entries = entries;
+ bf_metadata->bpe = calcBpe(error);
+ bf_metadata->size = 0;
+
+ uint64_t bits = 0;
+
+ {
+ double bn2 = logb(static_cast<double>(entries) * bf_metadata->bpe);
+ // if (bn2 > 63 || bn2 == INFINITY) {
+ // return 1;
+ // }
+ bf_metadata->n2 = static_cast<uint8_t>(bn2) + 1;
+ bits = 1LLU << bf_metadata->n2;
+
+ // Determine the number of extra bits available for more items. We rounded
+ // up the number of bits to the next-highest power of two. This means we
+ // might have up to 2x the bits available to us.
+ uint64_t bit_diff = bits - static_cast<uint64_t>(static_cast<double>(entries) * bf_metadata->bpe);
+ // The number of additional items we can store is the extra number of bits
+ // divided by bits-per-element
+ auto item_diff = static_cast<uint64_t>(static_cast<double>(bit_diff) / bf_metadata->bpe);
+ bf_metadata->entries += item_diff;
+ }
+
+ if (bits % 64) {
+ bf_metadata->bf_bytes = ((bits / 64) + 1) * 8;
+ } else {
+ bf_metadata->bf_bytes = bits / 8;
+ }
+ bf_metadata->bf_bits = bf_metadata->bf_bytes * 8;
+
+ bf_metadata->hashes = (int)ceil(0.693147180559945 * bf_metadata->bpe); // ln(2)
+}
+
+BloomHashval BloomFilter::bloomCalcHash64(const void *buffer, int len) {
+ BloomHashval rv;
+ rv.a = MurmurHash64A_Bloom(buffer, len, 0xc6a4a7935bd1e995ULL);
+ rv.b = MurmurHash64A_Bloom(buffer, len, rv.a);
+ return rv;
+}
+
+int BloomFilter::testBitSetBit(std::string &value, uint64_t x, ReadWriteMode mode) {
+ uint8_t mask = 1 << (x % 8);
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ uint8_t c = value[byte_index]; // expensive memory access
+
+ if (c & mask) {
+ return 1;
+ } else {
+ if (mode == ReadWriteMode::MODE_WRITE) {
+ uint8_t s = c | mask;
+ value[byte_index] = static_cast<char>(s);
+ }
+ return 0;
+ }
+}
+
+rocksdb::Status BloomFilter::getSBChainMetadata(const Slice &ns_key, SBChainMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, ns_key, metadata);
+}
+
+rocksdb::Status BloomFilter::getBFMetadata(const Slice &bf_meta_key, BFMetadata *metadata) {
+ return Database::GetMetadata(kRedisBloomFilter, bf_meta_key, metadata);
+}
+
+rocksdb::Status BloomFilter::createSBChain(const Slice &ns_key, double error_rate, uint64_t capacity,
+ uint16_t expansion, uint16_t scaling, SBChainMetadata &sb_chain_metadata) {
+ sb_chain_metadata.n_filters = 1;
+ sb_chain_metadata.growth = expansion;
+ sb_chain_metadata.options = scaling;
+ sb_chain_metadata.size = 0;
+
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createSBChain"});
+ batch->PutLogData(log_data.Encode());
+
+ std::string sb_chain_bytes;
+ sb_chain_metadata.Encode(&sb_chain_bytes);
+ batch->Put(metadata_cf_handle_, ns_key, sb_chain_bytes);
+
+ std::string bf_key;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ BFMetadata bf_metadata;
+ bfInit(&bf_metadata, capacity, error_rate, scaling);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::createBloomFilter(const Slice &ns_key, SBChainMetadata &sb_chain_metadata,
+ BFMetadata &bf_metadata, std::string &bf_key) {
+ auto batch = storage_->GetWriteBatchBase();
+ WriteBatchLogData log_data(kRedisBloomFilter, {"createBloomFilter"});
+ batch->PutLogData(log_data.Encode());
+
+ BFMetadata old_bf_metadata = bf_metadata;
+ sb_chain_metadata.n_filters += 1;
+ appendBFSuffix(ns_key, sb_chain_metadata.n_filters - 1, &bf_key);
+ std::string bf_meta_key;
+ appendBFMetaSuffix(bf_key, &bf_meta_key);
+ bfInit(&bf_metadata, sb_chain_metadata.growth * old_bf_metadata.entries,
+ old_bf_metadata.error * kErrorTighteningRatio, sb_chain_metadata.options);
+ std::string bf_meta_bytes;
+ bf_metadata.Encode(&bf_meta_bytes);
+ batch->Put(metadata_cf_handle_, bf_meta_key, bf_meta_bytes);
+
+ return storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+}
+
+rocksdb::Status BloomFilter::bloomCheckAdd64(const Slice &bf_key, BFMetadata &bf_metadata, BloomHashval hashval,
+ ReadWriteMode mode, ObserverOrUniquePtr<rocksdb::WriteBatchBase> &batch,
+ int &ret) {
+ int found_unset = 0;
+ const uint64_t mod = (1LLU << bf_metadata.n2);
+ for (uint64_t i = 0; i < bf_metadata.hashes; i++) {
+ uint64_t x = ((hashval.a + i * hashval.b)) % mod;
+ uint32_t index = (x / kBloomFilterSegmentBits) * kBloomFilterSegmentBytes;
+ std::string sub_key, value;
+ InternalKey(bf_key, std::to_string(index), bf_metadata.version, storage_->IsSlotIdEncoded()).Encode(&sub_key);
+ rocksdb::Status s =
+ storage_->Get(rocksdb::ReadOptions(), sub_key, &value); // todo: use bf_key in InternalKey maybe error
+ if (!s.ok() && !s.IsNotFound()) return s;
+ uint32_t byte_index = (x / 8) % kBloomFilterSegmentBytes;
+ if (byte_index >= value.size()) { // expand the bloomfilter
+ size_t expand_size = 0;
+ if (byte_index >= value.size() * 2) {
+ expand_size = byte_index - value.size() + 1;
+ } else if (value.size() * 2 > kBloomFilterSegmentBytes) {
+ expand_size = kBloomFilterSegmentBytes - value.size();
+ } else {
+ expand_size = value.size();
+ }
+ value.append(expand_size, 0);
+ }
+
+ if (!testBitSetBit(value, x, mode)) {
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 0;
+ return rocksdb::Status::OK();
+ } else {
+ found_unset = 1;
+ batch->Put(sub_key, value);
+ storage_->Write(storage_->DefaultWriteOptions(), batch->GetWriteBatch());
+ // todo: if the indexs are same, the value need to be updated immediately, if have other ways to compl it
+ }
+ }
+ }
+ if (mode == ReadWriteMode::MODE_READ) {
+ ret = 1;
+ } else {
+ ret = found_unset;
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status BloomFilter::Reserve(const Slice &user_key, double error_rate, uint64_t capacity, uint16_t expansion,
+ uint16_t scaling) {
+ std::string ns_key;
+ AppendNamespacePrefix(user_key, &ns_key);
+
+ LockGuard guard(storage_->GetLockManager(), ns_key);
+ SBChainMetadata sb_chain_metadata;
+ rocksdb::Status s = getSBChainMetadata(ns_key, &sb_chain_metadata);
+ if (!s.ok() && !s.IsNotFound()) return s;
+ if (!s.IsNotFound()) {
+ return rocksdb::Status::InvalidArgument("item exists");
+ }
+
+ return createSBChain(ns_key, error_rate, capacity, expansion, scaling, sb_chain_metadata);
+}
+
+rocksdb::Status BloomFilter::Add(const Slice &user_key, const Slice &item, int &ret) {
Review Comment:
```suggestion
rocksdb::Status BloomFilter::Add(const Slice &user_key, const Slice &item, int* ret) {
```
Personally, I perfer `pointer` for output argument.
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
+ public:
+ uint16_t n_filters;
+ uint16_t options;
Review Comment:
RedisBloom uses `options`. it use `BLOOM_OPT_FORCE64` and `BLOOM_OPT_NO_SCALING` etc. We can not compability with them. And if we'd like to keep compatible, maybe we need some comments here.
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU closed pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU closed pull request #1632: [WIP] BloomFilter command Draft and implemention
URL: https://github.com/apache/kvrocks/pull/1632
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] zncleon commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "zncleon (via GitHub)" <gi...@apache.org>.
zncleon commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1298382071
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
Review Comment:
Sure, SB is short for Scalable BloomFilter as RedisBloom do.
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] git-hulk commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "git-hulk (via GitHub)" <gi...@apache.org>.
git-hulk commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1297402857
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
Review Comment:
Not sure what's SB short for which words, Scalable BloomFilter?
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1286605546
##########
src/storage/redis_metadata.h:
##########
@@ -143,6 +144,7 @@ class Metadata {
virtual void Encode(std::string *dst);
virtual rocksdb::Status Decode(const std::string &bytes);
bool operator==(const Metadata &that) const;
+ virtual ~Metadata() = default;
Review Comment:
Oh previously this might work, because seems we didn't `delete Metadata` when subclass like `ListMetadata` is used. However I think we'd better has this.
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kvrocks] mapleFU commented on a diff in pull request #1632: [WIP] BloomFilter command Draft and implemention
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #1632:
URL: https://github.com/apache/kvrocks/pull/1632#discussion_r1286599825
##########
src/storage/redis_metadata.h:
##########
@@ -197,3 +199,33 @@ class StreamMetadata : public Metadata {
void Encode(std::string *dst) override;
rocksdb::Status Decode(const std::string &bytes) override;
};
+
+class SBChainMetadata : public Metadata {
+ public:
+ uint16_t n_filters;
+ uint16_t options;
+ uint16_t growth;
+
+ explicit SBChainMetadata(bool generate_version = true) : Metadata(kRedisBloomFilter, generate_version) {}
+
+ void Encode(std::string *dst) override;
+ rocksdb::Status Decode(const std::string &bytes) override;
+};
+
+class BFMetadata : public Metadata {
+ public:
+ uint32_t hashes;
+ uint8_t n2;
+ uint64_t entries;
+
+ double error;
+ double bpe;
+
+ uint64_t bf_bits;
+ uint64_t bf_bytes;
+
+ explicit BFMetadata(bool generate_version = true) : Metadata(kRedisBloomFilter, generate_version) {}
Review Comment:
Should this better to be a String or something?
--
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: issues-unsubscribe@kvrocks.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org