[#25600] DocDB: Implement variable bloom filtering for RocksDB iterator

Summary:
Currently we check bloom filter when RocksDB iterator is being created.
So it could be used only when we plan to use this iterator for particular row lookup.

Implemented variable bloom filters check and used it during conflict resolution.
TServer insert time for PgSingleTServerTest.Scan test improved from 12.46s to 11.99s.

When bloom filter is used in variable mode, we defer checking whether iterator should be created for the particular sst file.
I.e. iterators are always created, but user could change bloom filter key on the fly.
`MergingIteratorBase` is responsible for managing bloom filter key updates.
When key is updated it checks whether particular iterator matches the provided key and could add/remove entries to iterators heap.
Jira: DB-14846

Test Plan: Jenkins

Reviewers: timur

Reviewed By: timur

Subscribers: kannan, ybase, yql

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D41176

Author: spolitov

src/yb/client/ql-tablet-test.cc

@@ -2031,7 +2031,7 @@ Status CalcKeysDistributionAcrossWorkers(
 
   auto iter = CreateRocksDBIterator(
       tablet->doc_db().regular, tablet->doc_db().key_bounds,
-      docdb::BloomFilterMode::DONT_USE_BLOOM_FILTER, boost::none, rocksdb::kDefaultQueryId,
+      docdb::BloomFilterOptions::Inactive(), rocksdb::kDefaultQueryId,
       /* file_filter = */ nullptr, /* iterate_upper_bound = */ nullptr,
       rocksdb::CacheRestartBlockKeys { direction == tablet::Direction::kBackward });
   if (direction == tablet::Direction::kForward) {

src/yb/docdb/bounded_rocksdb_iterator.cc

@@ -55,15 +55,29 @@ const rocksdb::KeyValueEntry& BoundedRocksDbIterator::SeekToLast() {
 }
 
 const rocksdb::KeyValueEntry& BoundedRocksDbIterator::Seek(Slice target) {
+  return DoSeek(target, nullptr);
+}
+
+template <class Filter>
+const rocksdb::KeyValueEntry& BoundedRocksDbIterator::DoSeek(Slice target, Filter filter_user_key) {
   if (!key_bounds_->lower.empty() && target.compare(key_bounds_->lower) < 0) {
-    return FilterEntry(iterator_->Seek(key_bounds_->lower));
+    return FilterEntry(DoSeekImpl(key_bounds_->lower, filter_user_key));
   }
 
   if (!key_bounds_->upper.empty() && target.compare(key_bounds_->upper) > 0) {
-    return FilterEntry(iterator_->Seek(key_bounds_->upper));
+    return FilterEntry(DoSeekImpl(key_bounds_->upper, filter_user_key));
   }
 
-  return FilterEntry(iterator_->Seek(target));
+  return FilterEntry(DoSeekImpl(target, filter_user_key));
+}
+
+const rocksdb::KeyValueEntry& BoundedRocksDbIterator::DoSeekImpl(Slice target, std::nullptr_t) {
+  return iterator_->Seek(target);
+}
+
+const rocksdb::KeyValueEntry& BoundedRocksDbIterator::DoSeekImpl(
+    Slice target, Slice filter_user_key) {
+  return iterator_->SeekWithNewFilter(target, filter_user_key);
 }
 
 const rocksdb::KeyValueEntry& BoundedRocksDbIterator::Next() {
@@ -97,5 +111,10 @@ void BoundedRocksDbIterator::UseFastNext(bool value) {
   iterator_->UseFastNext(value);
 }
 
+const rocksdb::KeyValueEntry& BoundedRocksDbIterator::DoSeekWithNewFilter(
+    Slice target, Slice filter_user_key) {
+  return DoSeek(target, filter_user_key);
+}
+
 }  // namespace docdb
 }  // namespace yb

src/yb/docdb/bounded_rocksdb_iterator.h

@@ -75,6 +75,11 @@ class BoundedRocksDbIterator final : public rocksdb::Iterator {
 
  private:
   const rocksdb::KeyValueEntry& FilterEntry(const rocksdb::KeyValueEntry& entry) const;
+  template <class Filter>
+  const rocksdb::KeyValueEntry& DoSeek(Slice target, Filter filter_user_key);
+  const rocksdb::KeyValueEntry& DoSeekImpl(Slice target, std::nullptr_t);
+  const rocksdb::KeyValueEntry& DoSeekImpl(Slice target, Slice filter_user_key);
+  const rocksdb::KeyValueEntry& DoSeekWithNewFilter(Slice target, Slice filter_user_key) override;
 
   std::unique_ptr<rocksdb::Iterator> iterator_;
   const KeyBounds* key_bounds_;

src/yb/docdb/conflict_resolution.cc

@@ -896,23 +896,20 @@ class StrongConflictChecker {
 
   Status Check(
       Slice intent_key, bool strong, ConflictManagementPolicy conflict_management_policy) {
-    const auto bloom_filter_prefix = VERIFY_RESULT(ExtractFilterPrefixFromKey(intent_key));
-    if (!value_iter_.Initialized() || bloom_filter_prefix != value_iter_bloom_filter_prefix_) {
+    if (!value_iter_.Initialized()) {
       auto hybrid_time_file_filter =
           FLAGS_docdb_ht_filter_conflict_with_committed ? CreateHybridTimeFileFilter(read_time_)
                                                         : nullptr;
       value_iter_ = CreateRocksDBIterator(
           resolver_.doc_db().regular,
           resolver_.doc_db().key_bounds,
-          BloomFilterMode::USE_BLOOM_FILTER,
-          intent_key,
+          BloomFilterOptions::Variable(),
           rocksdb::kDefaultQueryId,
           hybrid_time_file_filter,
           /* iterate_upper_bound = */ nullptr,
           rocksdb::CacheRestartBlockKeys::kFalse);
-      value_iter_bloom_filter_prefix_ = bloom_filter_prefix;
     }
-    value_iter_.Seek(intent_key);
+    value_iter_.SeekWithNewFilter(intent_key);
 
     VLOG_WITH_PREFIX_AND_FUNC(4)
         << "Overwrite; Seek: " << intent_key.ToDebugString() << " ("
@@ -994,7 +991,6 @@ class StrongConflictChecker {
 
   // RocksDb iterator with bloom filter can be reused in case keys has same hash component.
   BoundedRocksDbIterator value_iter_;
-  Slice value_iter_bloom_filter_prefix_;
 };
 
 class ConflictResolverContextBase : public ConflictResolverContext {

src/yb/docdb/cql_operation.cc

@@ -637,8 +637,7 @@ Result<bool> QLWriteOperation::HasDuplicateUniqueIndexValueBackward(
 
   auto iter = CreateIntentAwareIterator(
       data.doc_write_batch->doc_db(),
-      BloomFilterMode::USE_BLOOM_FILTER,
-      pk_doc_key_->Encode().AsSlice(),
+      BloomFilterOptions::Fixed(pk_doc_key_->Encode().AsSlice()),
       request_.query_id(),
       txn_op_context_,
       data.read_operation_data);

src/yb/docdb/doc_reader.cc

@@ -220,7 +220,7 @@ Result<DocHybridTime> GetTableTombstoneTime(
   table_id = table_id_buf.AsSlice();
 
   auto iter = CreateIntentAwareIterator(
-      doc_db, BloomFilterMode::USE_BLOOM_FILTER, table_id, rocksdb::kDefaultQueryId, txn_op_context,
+      doc_db, BloomFilterOptions::Fixed(table_id), rocksdb::kDefaultQueryId, txn_op_context,
       read_operation_data.WithStatistics(nullptr));
   iter->Seek(table_id, SeekFilter::kAll);
   const auto& entry_data = VERIFY_RESULT_REF(iter->Fetch());
@@ -250,7 +250,7 @@ Result<std::optional<SubDocument>> TEST_GetSubDocument(
     const ReadOperationData& read_operation_data,
     const dockv::ReaderProjection* projection) {
   auto iter = CreateIntentAwareIterator(
-      doc_db, BloomFilterMode::USE_BLOOM_FILTER, sub_doc_key, query_id,
+      doc_db, BloomFilterOptions::Fixed(sub_doc_key), query_id,
       txn_op_context, read_operation_data);
   DOCDB_DEBUG_LOG("GetSubDocument for key $0 @ $1", sub_doc_key.ToDebugHexString(),
                   iter->read_time().ToString());

src/yb/docdb/doc_reader_redis.cc

@@ -386,7 +386,7 @@ Status GetRedisSubDocument(
     const TransactionOperationContext& txn_op_context,
     const ReadOperationData& read_operation_data) {
   auto iter = CreateIntentAwareIterator(
-      doc_db, BloomFilterMode::USE_BLOOM_FILTER, data.subdocument_key, query_id, txn_op_context,
+      doc_db, BloomFilterOptions::Fixed(data.subdocument_key), query_id, txn_op_context,
       read_operation_data);
   return GetRedisSubDocument(iter.get(), data, nullptr /* projection */, SeekFwdSuffices::kFalse);
 }

src/yb/docdb/doc_rowwise_iterator.cc

@@ -83,8 +83,7 @@ DocRowwiseIterator::DocRowwiseIterator(
 DocRowwiseIterator::~DocRowwiseIterator() = default;
 
 void DocRowwiseIterator::InitIterator(
-    BloomFilterMode bloom_filter_mode,
-    const boost::optional<const Slice>& user_key_for_filter,
+    const BloomFilterOptions& bloom_filter,
     const rocksdb::QueryId query_id,
     std::shared_ptr<rocksdb::ReadFileFilter> file_filter) {
   if (table_type_ == TableType::PGSQL_TABLE_TYPE) {
@@ -103,8 +102,7 @@ void DocRowwiseIterator::InitIterator(
 
   db_iter_ = CreateIntentAwareIterator(
       doc_db_,
-      bloom_filter_mode,
-      user_key_for_filter,
+      bloom_filter,
       query_id,
       txn_op_context_,
       read_operation_data_,

src/yb/docdb/doc_rowwise_iterator.h

@@ -108,8 +108,7 @@ class DocRowwiseIterator final : public DocRowwiseIteratorBase {
 
  private:
   void InitIterator(
-      BloomFilterMode bloom_filter_mode = BloomFilterMode::DONT_USE_BLOOM_FILTER,
-      const boost::optional<const Slice>& user_key_for_filter = boost::none,
+      const BloomFilterOptions& bloom_filter = BloomFilterOptions::Inactive(),
       const rocksdb::QueryId query_id = rocksdb::kDefaultQueryId,
       std::shared_ptr<rocksdb::ReadFileFilter> file_filter = nullptr) override;
 

src/yb/docdb/doc_rowwise_iterator_base.cc

@@ -172,8 +172,8 @@ Status DocRowwiseIteratorBase::Init(const qlexpr::YQLScanSpec& doc_spec, SkipSee
   const bool is_fixed_point_get =
       !bounds.lower.empty() &&
       VERIFY_RESULT(HashedOrFirstRangeComponentsEqual(bounds.lower, bounds.upper));
-  const auto mode = is_fixed_point_get ? BloomFilterMode::USE_BLOOM_FILTER
-                                       : BloomFilterMode::DONT_USE_BLOOM_FILTER;
+  const auto bloom_filter = is_fixed_point_get ? BloomFilterOptions::Fixed(bounds.lower.AsSlice())
+                                               : BloomFilterOptions::Inactive();
 
   if (is_forward_scan_) {
     has_bound_key_ = !bounds.upper.empty();
@@ -194,7 +194,7 @@ Status DocRowwiseIteratorBase::Init(const qlexpr::YQLScanSpec& doc_spec, SkipSee
     }
   }
 
-  InitIterator(mode, bounds.lower.AsSlice(), doc_spec.QueryId(), CreateFileFilter(doc_spec));
+  InitIterator(bloom_filter, doc_spec.QueryId(), CreateFileFilter(doc_spec));
 
   if (has_bound_key_) {
     if (is_forward_scan_) {