From c6fea40236c7b1ab67eab6ac83a61e4e6786a344 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 06:48:50 +0200
Subject: [PATCH 01/10] refactor: unify merge resolution via bloom-filtered
 iterator pipeline
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Replace the hand-rolled resolve_merge_get() with a bloom-filtered
single-key iterator pipeline that reuses MvccStream for merge/RT/
Indirection resolution. This eliminates the duplicated operand
collection logic between point reads and range scans.

- Add Table::bloom_may_contain_key() for standard bloom pre-filtering
- Extract bloom_may_contain_hash() as shared base for prefix and key bloom
- Add key_hash field to IterState for single-key bloom filtering
- Refactor create_range bloom checks into bloom_passes() helper
- Replace resolve_merge_get with resolve_merge_via_pipeline using
  Merger → MvccStream on a key..=key range
- Remove unused Memtable::get_all_for_key()

Closes #46
---
 src/memtable/mod.rs     |  68 -------------
 src/range.rs            |  77 ++++++++-------
 src/table/mod.rs        |  46 ++++++---
 src/tree/mod.rs         | 212 ++++++++++------------------------------
 tests/merge_operator.rs |  12 +--
 5 files changed, 138 insertions(+), 277 deletions(-)

diff --git a/src/memtable/mod.rs b/src/memtable/mod.rs
index 19c3505fb..433ee24e6 100644
--- a/src/memtable/mod.rs
+++ b/src/memtable/mod.rs
@@ -144,31 +144,6 @@ impl Memtable {
         })
     }
 
-    /// Collects all entries for a given key with seqno < `seqno`,
-    /// ordered by descending sequence number (newest first).
-    ///
-    /// Used by the merge operator read path to collect all operands for a key.
-    // Allocates a Vec and clones entries — acceptable for the merge slow-path.
-    // A zero-copy iterator API would avoid this but changes the skiplist contract.
-    pub(crate) fn get_all_for_key(&self, key: &[u8], seqno: SeqNo) -> Vec<InternalValue> {
-        if seqno == 0 {
-            return Vec::new();
-        }
-
-        // ValueType is not part of InternalKey ordering (only user_key + Reverse(seqno)),
-        // so the value type here is arbitrary — it does not affect seek position.
-        let lower_bound = InternalKey::new(key, seqno - 1, ValueType::Value);
-
-        self.items
-            .range(lower_bound..)
-            .take_while(|entry| &*entry.key().user_key == key)
-            .map(|entry| InternalValue {
-                key: entry.key().clone(),
-                value: entry.value().clone(),
-            })
-            .collect()
-    }
-
     /// Gets approximate size of memtable in bytes.
     pub fn size(&self) -> u64 {
         self.approximate_size
@@ -693,47 +668,4 @@ mod tests {
             memtable.get(b"abc", 50)
         );
     }
-
-    #[test]
-    fn get_all_for_key_seqno_zero_returns_empty() {
-        let memtable = Memtable::new(0);
-        memtable.insert(crate::InternalValue::from_components(
-            "key",
-            "val",
-            1,
-            ValueType::Value,
-        ));
-
-        // seqno=0 means nothing is visible — early return
-        assert!(memtable.get_all_for_key(b"key", 0).is_empty());
-    }
-
-    #[test]
-    fn get_all_for_key_returns_all_versions() {
-        let memtable = Memtable::new(0);
-        memtable.insert(crate::InternalValue::from_components(
-            "key",
-            "op2",
-            3,
-            ValueType::MergeOperand,
-        ));
-        memtable.insert(crate::InternalValue::from_components(
-            "key",
-            "op1",
-            2,
-            ValueType::MergeOperand,
-        ));
-        memtable.insert(crate::InternalValue::from_components(
-            "key",
-            "base",
-            1,
-            ValueType::Value,
-        ));
-
-        let entries = memtable.get_all_for_key(b"key", 4);
-        assert_eq!(entries.len(), 3);
-        assert_eq!(entries[0].key.seqno, 3);
-        assert_eq!(entries[1].key.seqno, 2);
-        assert_eq!(entries[2].key.seqno, 1);
-    }
 }
diff --git a/src/range.rs b/src/range.rs
index 2b12a2297..4aa16cb7b 100644
--- a/src/range.rs
+++ b/src/range.rs
@@ -79,6 +79,12 @@ pub struct IterState {
     /// hash will be skipped entirely during the scan.
     pub(crate) prefix_hash: Option<u64>,
 
+    /// Optional key hash for standard bloom filter pre-filtering.
+    ///
+    /// When set (typically for single-key point-read pipelines), segments
+    /// whose bloom filter reports no match for this hash will be skipped.
+    pub(crate) key_hash: Option<u64>,
+
     /// Optional metrics handle for recording prefix-related statistics (e.g. bloom skips).
     ///
     /// `None` when the caller does not wish to record metrics; this is
@@ -130,6 +136,41 @@ fn range_tombstone_overlaps_bounds(
     overlaps_lo && overlaps_hi
 }
 
+/// Checks prefix and key bloom filters for a table.
+///
+/// Returns `true` if the table should be included (bloom says "maybe" or no
+/// filter available), `false` if it can be safely skipped.
+fn bloom_passes(state: &IterState, table: &crate::table::Table) -> bool {
+    if let Some(prefix_hash) = state.prefix_hash {
+        match table.maybe_contains_prefix(prefix_hash) {
+            Ok(false) => {
+                #[cfg(feature = "metrics")]
+                if let Some(m) = &state.metrics {
+                    m.prefix_bloom_skips
+                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                }
+                return false;
+            }
+            Err(e) => {
+                log::debug!("prefix bloom check failed for table {:?}: {e}", table.id(),);
+            }
+            _ => {}
+        }
+    }
+
+    if let Some(key_hash) = state.key_hash {
+        match table.bloom_may_contain_key(key_hash) {
+            Ok(false) => return false,
+            Err(e) => {
+                log::debug!("key bloom check failed for table {:?}: {e}", table.id(),);
+            }
+            _ => {}
+        }
+    }
+
+    true
+}
+
 impl TreeIter {
     #[expect(
         clippy::too_many_lines,
@@ -231,39 +272,9 @@ impl TreeIter {
                         if table.check_key_range_overlap(&(
                             user_range.0.as_ref().map(std::convert::AsRef::as_ref),
                             user_range.1.as_ref().map(std::convert::AsRef::as_ref),
-                        )) {
-                            // If a prefix hash is available (prefix scan with prefix bloom
-                            // filters configured), check the bloom filter for the prefix.
-                            // Skip the segment if the prefix is definitely absent.
-                            if let Some(prefix_hash) = lock.prefix_hash {
-                                match table.maybe_contains_prefix(prefix_hash) {
-                                    Ok(false) => {
-                                        // Prefix bloom says this segment has no matching keys
-                                        // — skip it entirely.
-                                        #[cfg(feature = "metrics")]
-                                        if let Some(m) = &lock.metrics {
-                                            m.prefix_bloom_skips
-                                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-                                        }
-                                    }
-                                    Ok(true) => {
-                                        single_tables.push(table.clone());
-                                    }
-                                    Err(e) => {
-                                        // On I/O error reading the filter, include the segment
-                                        // conservatively to avoid missing data. Use debug level
-                                        // to avoid log noise during transient I/O issues in
-                                        // prefix-heavy workloads.
-                                        log::debug!(
-                                            "prefix bloom check failed for table {:?}: {e}",
-                                            table.id(),
-                                        );
-                                        single_tables.push(table.clone());
-                                    }
-                                }
-                            } else {
-                                single_tables.push(table.clone());
-                            }
+                        )) && bloom_passes(lock, table)
+                        {
+                            single_tables.push(table.clone());
                         }
                     }
                     _ => {
diff --git a/src/table/mod.rs b/src/table/mod.rs
index 674ea5b6d..6bb482854 100644
--- a/src/table/mod.rs
+++ b/src/table/mod.rs
@@ -780,25 +780,24 @@ impl Table {
         self.metadata.key_range.overlaps_with_bounds(bounds)
     }
 
-    /// Checks the bloom filter for a prefix hash.
+    /// Checks the full-table bloom filter for a hash value.
     ///
-    /// Returns `Ok(true)` if the prefix may exist in this table (or if no
-    /// filter is available), `Ok(false)` if the prefix is definitely absent.
+    /// Returns `Ok(true)` if the hash may exist in the filter (or if no full
+    /// filter is available), `Ok(false)` if the hash is definitely absent.
     ///
-    /// This is used by prefix scans to skip segments that contain no keys
-    /// with a matching prefix. The prefix must have been indexed at write
-    /// time via a [`PrefixExtractor`](crate::PrefixExtractor).
-    pub(crate) fn maybe_contains_prefix(&self, prefix_hash: u64) -> crate::Result<bool> {
+    /// Handles full (non-partitioned) filters directly. Partitioned / TLI
+    /// filters are keyed by user key, not raw hash, so this method returns
+    /// `Ok(true)` conservatively for those types.
+    fn bloom_may_contain_hash(&self, hash: u64) -> crate::Result<bool> {
         // Full (non-partitioned) filter — single bloom covers the entire table
         if let Some(block) = &self.pinned_filter_block {
-            return block.maybe_contains_hash(prefix_hash);
+            return block.maybe_contains_hash(hash);
         }
 
         // Partitioned / TLI filters: partition index is keyed by user key, not
-        // prefix hash — we would need to scan ALL partitions to check the prefix,
+        // raw hash — we would need to scan ALL partitions to check,
         // which is O(partitions) I/O and defeats the purpose of bloom skip.
         // Returning Ok(true) is correct (conservative: segment is NOT skipped).
-        // Future: accept prefix bounds to seek overlapping partitions only.
         if self.pinned_filter_index.is_some() || self.regions.filter_tli.is_some() {
             return Ok(true);
         }
@@ -813,13 +812,36 @@ impl Table {
                 CompressionType::None, // NOTE: Filter blocks are never compressed (crate invariant)
             )?;
             let block = FilterBlock::new(block);
-            return block.maybe_contains_hash(prefix_hash);
+            return block.maybe_contains_hash(hash);
         }
 
-        // No filter available — cannot rule out the prefix
+        // No filter available — cannot rule out the hash
         Ok(true)
     }
 
+    /// Checks the bloom filter for a prefix hash.
+    ///
+    /// Returns `Ok(true)` if the prefix may exist in this table (or if no
+    /// filter is available), `Ok(false)` if the prefix is definitely absent.
+    ///
+    /// This is used by prefix scans to skip segments that contain no keys
+    /// with a matching prefix. The prefix must have been indexed at write
+    /// time via a [`PrefixExtractor`](crate::PrefixExtractor).
+    pub(crate) fn maybe_contains_prefix(&self, prefix_hash: u64) -> crate::Result<bool> {
+        self.bloom_may_contain_hash(prefix_hash)
+    }
+
+    /// Checks the bloom filter for a key hash.
+    ///
+    /// Returns `Ok(true)` if the key may exist in this table (or if no
+    /// filter is available), `Ok(false)` if the key is definitely absent.
+    ///
+    /// Used by the point-read merge pipeline to pre-filter disk tables
+    /// before building range iterators.
+    pub(crate) fn bloom_may_contain_key(&self, key_hash: u64) -> crate::Result<bool> {
+        self.bloom_may_contain_hash(key_hash)
+    }
+
     /// Returns the highest effective sequence number in the table.
     ///
     /// For tables produced by flush/compaction (`global_seqno == 0`), this
diff --git a/src/tree/mod.rs b/src/tree/mod.rs
index ec7a9fbf8..0bf2779dc 100644
--- a/src/tree/mod.rs
+++ b/src/tree/mod.rs
@@ -666,7 +666,7 @@ impl AbstractTree for Tree {
             &super_version,
             key,
             seqno,
-            self.config.merge_operator.as_deref(),
+            self.config.merge_operator.as_ref(),
         )
     }
 
@@ -683,7 +683,7 @@ impl AbstractTree for Tree {
                     &super_version,
                     key.as_ref(),
                     seqno,
-                    self.config.merge_operator.as_deref(),
+                    self.config.merge_operator.as_ref(),
                 )
             })
             .collect()
@@ -741,16 +741,22 @@ impl Tree {
         super_version: &SuperVersion,
         key: &[u8],
         seqno: SeqNo,
-        merge_operator: Option<&dyn crate::merge_operator::MergeOperator>,
+        merge_operator: Option<&Arc<dyn crate::merge_operator::MergeOperator>>,
     ) -> crate::Result<Option<UserValue>> {
         let entry = Self::get_internal_entry_from_version(super_version, key, seqno)?;
 
         match entry {
             Some(entry) if entry.key.value_type == ValueType::MergeOperand => {
                 if let Some(merge_op) = merge_operator {
-                    // Always resolve even for a single operand: there may be
-                    // older operands or a base value in lower storage layers.
-                    Self::resolve_merge_get(super_version, key, seqno, merge_op)
+                    // Build a bloom-filtered single-key iterator pipeline that
+                    // reuses MvccStream for merge/RT/Indirection resolution,
+                    // eliminating the previous hand-rolled merge collection.
+                    Self::resolve_merge_via_pipeline(
+                        super_version.clone(),
+                        key,
+                        seqno,
+                        Arc::clone(merge_op),
+                    )
                 } else if Self::is_suppressed_by_range_tombstones(
                     super_version,
                     key,
@@ -767,6 +773,46 @@ impl Tree {
         }
     }
 
+    /// Resolves merge operands for a point read via a bloom-filtered iterator pipeline.
+    ///
+    /// Builds a single-key range (`key..=key`) with bloom pre-filtering, wraps
+    /// all sources in `Merger → MvccStream`, and takes the first result. This
+    /// reuses the unified merge/RT/Indirection resolution logic from `MvccStream`
+    /// instead of duplicating it in a hand-rolled collection loop.
+    ///
+    /// Bloom pre-filtering ensures that 95%+ of disk tables are rejected at the
+    /// filter level, preserving O(1) reject performance on deep LSM trees.
+    fn resolve_merge_via_pipeline(
+        version: SuperVersion,
+        key: &[u8],
+        seqno: SeqNo,
+        merge_operator: Arc<dyn crate::merge_operator::MergeOperator>,
+    ) -> crate::Result<Option<UserValue>> {
+        use crate::range::{IterState, TreeIter};
+
+        let key_hash = crate::table::filter::standard_bloom::Builder::get_hash(key);
+        let key_slice = crate::Slice::from(key);
+        let range = key_slice.clone()..=key_slice;
+
+        let iter_state = IterState {
+            version,
+            ephemeral: None,
+            merge_operator: Some(merge_operator),
+            prefix_hash: None,
+            key_hash: Some(key_hash),
+            #[cfg(feature = "metrics")]
+            metrics: None,
+        };
+
+        let mut iter = TreeIter::create_range(iter_state, range, seqno);
+
+        match iter.next() {
+            Some(Ok(entry)) => Ok(Some(entry.value)),
+            Some(Err(e)) => Err(e),
+            None => Ok(None),
+        }
+    }
+
     #[doc(hidden)]
     pub fn create_internal_range<'a, K: AsRef<[u8]> + 'a, R: RangeBounds<K> + 'a>(
         version: SuperVersion,
@@ -822,6 +868,7 @@ impl Tree {
             ephemeral,
             merge_operator,
             prefix_hash,
+            key_hash: None,
             #[cfg(feature = "metrics")]
             metrics: None,
         };
@@ -829,158 +876,6 @@ impl Tree {
         TreeIter::create_range(iter_state, bounds, seqno)
     }
 
-    /// Resolves merge operands for a point read.
-    ///
-    /// Collects ALL entries for the key across all storage layers (active memtable,
-    /// sealed memtables, disk tables), identifies the base value, and applies the
-    /// merge operator. Entries are processed from newest to oldest (descending seqno).
-    ///
-    /// This intentionally duplicates merge-collection logic from `MvccStream`
-    /// because point reads access storage layers directly (memtable, sealed,
-    /// disk) rather than through a merged iterator stream, and need per-entry
-    /// RT suppression via `is_suppressed_by_range_tombstones`.
-    fn resolve_merge_get(
-        super_version: &SuperVersion,
-        key: &[u8],
-        seqno: SeqNo,
-        merge_op: &dyn crate::merge_operator::MergeOperator,
-    ) -> crate::Result<Option<UserValue>> {
-        let mut operands: Vec<UserValue> = Vec::new();
-        let mut base_value: Option<UserValue> = None;
-        let mut found_base = false;
-
-        let mut has_indirection_base = false;
-
-        // Process a single entry. Returns true if search should stop.
-        let mut process_entry = |entry: &InternalValue| -> bool {
-            match entry.key.value_type {
-                ValueType::Value => {
-                    base_value = Some(entry.value.clone());
-                    true
-                }
-                ValueType::Indirection => {
-                    // Indirection entries point to blob-stored values and must
-                    // not be forwarded as raw bytes to the merge operator.
-                    has_indirection_base = true;
-                    true
-                }
-                ValueType::Tombstone | ValueType::WeakTombstone => true,
-                ValueType::MergeOperand => {
-                    operands.push(entry.value.clone());
-                    false
-                }
-            }
-        };
-
-        // Check if an entry is suppressed by a range tombstone. RT-suppressed
-        // entries are logically deleted and must not participate in merge
-        // resolution — treat them as a tombstone boundary.
-        let is_rt_suppressed = |entry: &InternalValue| -> bool {
-            Self::is_suppressed_by_range_tombstones(super_version, key, entry.key.seqno, seqno)
-        };
-
-        // 1. Scan active memtable — returns all entries for key in desc seqno order
-        for entry in &super_version.active_memtable.get_all_for_key(key, seqno) {
-            if is_rt_suppressed(entry) {
-                found_base = true;
-                break;
-            }
-            if process_entry(entry) {
-                found_base = true;
-                break;
-            }
-        }
-
-        // 2. Scan sealed memtables (newest first)
-        if !found_base {
-            'sealed: for mt in super_version.sealed_memtables.iter().rev() {
-                for entry in &mt.get_all_for_key(key, seqno) {
-                    if is_rt_suppressed(entry) {
-                        found_base = true;
-                        break 'sealed;
-                    }
-                    if process_entry(entry) {
-                        found_base = true;
-                        break 'sealed;
-                    }
-                }
-            }
-        }
-
-        // 3. Scan tables on disk
-        //
-        // L0 runs can overlap and iter_levels()/run ordering is not
-        // guaranteed newest-first. Collect all matching on-disk entries for
-        // this key and process them in descending seqno order so that newer
-        // MergeOperands are seen before older bases/tombstones.
-        if !found_base {
-            let key_slice = crate::Slice::from(key);
-
-            let mut disk_entries: Vec<InternalValue> = Vec::new();
-
-            for run in super_version
-                .version
-                .iter_levels()
-                .flat_map(|lvl| lvl.iter())
-            {
-                if let Some(table) = run.get_for_key(key) {
-                    let range = key_slice.clone()..=key_slice.clone();
-                    for item in table.range(range) {
-                        let item = item?;
-                        if item.key.seqno >= seqno {
-                            continue;
-                        }
-                        disk_entries.push(item);
-                    }
-                }
-            }
-
-            // Newest-first by seqno
-            disk_entries.sort_by(|a, b| b.key.seqno.cmp(&a.key.seqno));
-
-            for entry in &disk_entries {
-                if is_rt_suppressed(entry) {
-                    break;
-                }
-                if process_entry(entry) {
-                    break;
-                }
-            }
-        }
-
-        if has_indirection_base {
-            // We encountered an indirection as the would-be base value.
-            // Indirection payloads are internal blob pointers and must not be
-            // passed to the merge operator as user data.
-            //
-            // Fall back to the raw newest merge operand (if any), instead of
-            // reporting the key as missing. This preserves backward-compatible
-            // behavior for callers that expect at least the latest operand when
-            // merge resolution over an indirection base is not supported.
-            if let Some(latest_operand) = operands.first() {
-                return Ok(Some(latest_operand.clone()));
-            }
-
-            // No visible merge operands; nothing user-visible to return.
-            return Ok(None);
-        }
-
-        if operands.is_empty() {
-            return Ok(base_value);
-        }
-
-        // Build operand refs in chronological order (ascending seqno)
-        let mut operand_refs: Vec<&[u8]> = Vec::with_capacity(operands.len());
-        for op in operands.iter().rev() {
-            operand_refs.push(op.as_ref());
-        }
-        // MergeOperator::merge is user code — panics propagate to the caller.
-        // The RefUnwindSafe bound ensures safety if caught externally.
-        let merged = merge_op.merge(key, base_value.as_deref(), &operand_refs)?;
-
-        Ok(Some(merged))
-    }
-
     pub(crate) fn get_internal_entry_from_version(
         super_version: &SuperVersion,
         key: &[u8],
@@ -1329,6 +1224,7 @@ impl Tree {
             ephemeral,
             merge_operator: self.config.merge_operator.clone(),
             prefix_hash,
+            key_hash: None,
             #[cfg(feature = "metrics")]
             metrics: Some(self.0.metrics.clone()),
         };
diff --git a/tests/merge_operator.rs b/tests/merge_operator.rs
index 63d1acd82..1b78b03c5 100644
--- a/tests/merge_operator.rs
+++ b/tests/merge_operator.rs
@@ -560,7 +560,7 @@ fn merge_multiple_operands_in_single_table() -> lsm_tree::Result<()> {
     tree.flush_active_memtable(0)?;
 
     // All 4 entries are in the same table. table.get() returns only
-    // the newest (MergeOperand@3), but resolve_merge_get must collect
+    // the newest (MergeOperand@3), but resolve_merge_via_pipeline must collect
     // all entries via range scan to produce the correct result.
     assert_eq!(Some(160), get_counter(&tree, "counter", 4));
 
@@ -898,7 +898,7 @@ fn merge_rt_no_operator_get_and_multi_get_agree() -> lsm_tree::Result<()> {
 
 /// RT suppresses operand in disk range scan during merge resolution.
 /// Exercises the is_rt_suppressed path inside the table.range() fallback
-/// in resolve_merge_get (line ~909).
+/// in resolve_merge_via_pipeline (line ~909).
 #[test]
 fn merge_rt_suppresses_operand_in_disk_range_scan() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;
@@ -936,7 +936,7 @@ fn merge_disk_base_via_point_lookup() -> lsm_tree::Result<()> {
     tree.merge("counter", 10_i64.to_le_bytes(), 1);
     tree.merge("counter", 20_i64.to_le_bytes(), 2);
 
-    // resolve_merge_get: active memtable has op@2, op@1
+    // resolve_merge_via_pipeline: active memtable has op@2, op@1
     // Then scans disk: table.get() returns base@0 (Value, not MergeOperand)
     // → process_entry sets base_value, found_base=true
     assert_eq!(Some(130), get_counter(&tree, "counter", 3));
@@ -945,7 +945,7 @@ fn merge_disk_base_via_point_lookup() -> lsm_tree::Result<()> {
 }
 
 /// Merge with Tombstone base in sealed memtable — exercises sealed memtable
-/// scan path in resolve_merge_get (lines ~868-877).
+/// scan path in resolve_merge_via_pipeline (lines ~868-877).
 #[test]
 fn merge_tombstone_in_sealed_memtable() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;
@@ -961,7 +961,7 @@ fn merge_tombstone_in_sealed_memtable() -> lsm_tree::Result<()> {
     // New operands in active memtable
     tree.merge("counter", 42_i64.to_le_bytes(), 2);
 
-    // resolve_merge_get scans active (finds op@2), then disk (finds tombstone@1)
+    // resolve_merge_via_pipeline scans active (finds op@2), then disk (finds tombstone@1)
     // tombstone stops scan, merge with no base: merge(None, [42]) = 42
     assert_eq!(Some(42), get_counter(&tree, "counter", 3));
 
@@ -969,7 +969,7 @@ fn merge_tombstone_in_sealed_memtable() -> lsm_tree::Result<()> {
 }
 
 /// Merge where operands span active memtable and disk — tests that
-/// resolve_merge_get correctly collects from all layers.
+/// resolve_merge_via_pipeline correctly collects from all layers.
 #[test]
 fn merge_operands_across_active_and_disk() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;

From 0c8f22e1ad48195f7def21dd1907a96a9adda798 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 10:44:31 +0200
Subject: [PATCH 02/10] test: cover bloom_passes key_hash and prefix_hash
 rejection paths

Add integration tests that exercise the Ok(false) branch in
bloom_passes for both key bloom and prefix bloom pre-filtering:

- merge_bloom_skips_non_matching_tables: L0 table with wide key range
  [aaa,zzz] is skipped by key bloom when reading "counter" via the
  merge pipeline, while the table containing "counter" is included
- prefix_bloom_rejects_in_l0_single_table_run: L0 single-table run
  with key range spanning [aaa:*,zzz:*] is skipped by prefix bloom
  when scanning for "mmm:", while the table with "mmm:" keys passes
---
 tests/merge_operator.rs    | 30 +++++++++++++++++++++++++++++
 tests/tree_prefix_bloom.rs | 39 ++++++++++++++++++++++++++++++++++++++
 2 files changed, 69 insertions(+)

diff --git a/tests/merge_operator.rs b/tests/merge_operator.rs
index 1b78b03c5..84631b38c 100644
--- a/tests/merge_operator.rs
+++ b/tests/merge_operator.rs
@@ -993,3 +993,33 @@ fn merge_operands_across_active_and_disk() -> lsm_tree::Result<()> {
 
     Ok(())
 }
+
+/// Bloom pre-filter skips tables whose key_range overlaps but whose bloom
+/// filter reports the key absent. Exercises the Ok(false) → skip path in
+/// the bloom-filtered iterator pipeline (bloom_passes + key_hash).
+#[test]
+fn merge_bloom_skips_non_matching_tables() -> lsm_tree::Result<()> {
+    let folder = tempfile::tempdir()?;
+    let tree = open_tree_with_counter(&folder);
+
+    // Table 1: wide key range [aaa, zzz] that does NOT contain "counter".
+    // Its key_range overlaps "counter" but bloom will reject it.
+    tree.insert("aaa", 0_i64.to_le_bytes(), 0);
+    tree.insert("zzz", 0_i64.to_le_bytes(), 1);
+    tree.flush_active_memtable(0)?;
+
+    // Table 2: contains "counter" base value — bloom will accept it.
+    tree.insert("counter", 100_i64.to_le_bytes(), 2);
+    tree.flush_active_memtable(0)?;
+
+    // Merge operand in active memtable
+    tree.merge("counter", 10_i64.to_le_bytes(), 3);
+
+    // resolve_merge_via_pipeline builds a key..=key range with bloom hash:
+    //   Table 1: key_range [aaa,zzz] overlaps "counter" ✓, bloom → Ok(false) → SKIP
+    //   Table 2: key_range [counter,counter] overlaps ✓, bloom → Ok(true) → INCLUDE
+    //   merge(Some(100), [10]) = 110
+    assert_eq!(Some(110), get_counter(&tree, "counter", 4));
+
+    Ok(())
+}
diff --git a/tests/tree_prefix_bloom.rs b/tests/tree_prefix_bloom.rs
index 64a419d01..5b6b1cf1b 100644
--- a/tests/tree_prefix_bloom.rs
+++ b/tests/tree_prefix_bloom.rs
@@ -865,3 +865,42 @@ fn prefix_bloom_skip_metrics_zero_without_extractor() -> lsm_tree::Result<()> {
 
     Ok(())
 }
+
+/// Prefix scan correctness when an L0 single-table run has a wide key range
+/// that overlaps the scanned prefix but does not contain any matching keys.
+/// Ensures the non-matching table does not affect scan results, independent of
+/// whether the prefix bloom rejects the scanned prefix.
+#[test]
+fn prefix_bloom_rejects_in_l0_single_table_run() -> lsm_tree::Result<()> {
+    let folder = tempfile::tempdir()?;
+    let tree = tree_with_prefix_bloom(&folder)?;
+
+    // L0 table 1: keys with prefixes "aaa:" and "zzz:" — wide key range.
+    // Bloom contains "aaa:" and "zzz:" but NOT "mmm:".
+    for i in 0..5 {
+        tree.insert(format!("aaa:{i}"), "v", i);
+    }
+    for i in 5..10 {
+        tree.insert(format!("zzz:{i}"), "v", i);
+    }
+    tree.flush_active_memtable(0)?;
+
+    // L0 table 2: keys with prefix "mmm:" — bloom contains "mmm:".
+    for i in 10..15 {
+        tree.insert(format!("mmm:{i}"), "v", i);
+    }
+    tree.flush_active_memtable(0)?;
+
+    let results: Vec<_> = tree
+        .create_prefix("mmm:", 15, None)
+        .collect::<Result<Vec<_>, _>>()?;
+    assert_eq!(results.len(), 5);
+
+    // Verify existing prefixes still work
+    let results: Vec<_> = tree
+        .create_prefix("aaa:", 15, None)
+        .collect::<Result<Vec<_>, _>>()?;
+    assert_eq!(results.len(), 5);
+
+    Ok(())
+}

From 318fe328f628bcd39c27e91d169c813221ec4205 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 11:08:33 +0200
Subject: [PATCH 03/10] refactor(bloom): rename bloom_may_contain_key to
 bloom_may_contain_key_hash

- Clarify that the parameter is a precomputed u64 hash, not a user key
- Soften doc comment: replace hard "95%+" guarantee with qualified
  "can reject many" wording appropriate for all filter types
- Reword test names/docstrings to describe correctness validation
  rather than claiming direct bloom skip observation
---
 src/range.rs               | 2 +-
 src/table/mod.rs           | 4 ++--
 src/tree/mod.rs            | 4 ++--
 tests/merge_operator.rs    | 8 ++++----
 tests/tree_prefix_bloom.rs | 1 +
 5 files changed, 10 insertions(+), 9 deletions(-)

diff --git a/src/range.rs b/src/range.rs
index 4aa16cb7b..8cfe9aef6 100644
--- a/src/range.rs
+++ b/src/range.rs
@@ -159,7 +159,7 @@ fn bloom_passes(state: &IterState, table: &crate::table::Table) -> bool {
     }
 
     if let Some(key_hash) = state.key_hash {
-        match table.bloom_may_contain_key(key_hash) {
+        match table.bloom_may_contain_key_hash(key_hash) {
             Ok(false) => return false,
             Err(e) => {
                 log::debug!("key bloom check failed for table {:?}: {e}", table.id(),);
diff --git a/src/table/mod.rs b/src/table/mod.rs
index 6bb482854..1cff81603 100644
--- a/src/table/mod.rs
+++ b/src/table/mod.rs
@@ -831,14 +831,14 @@ impl Table {
         self.bloom_may_contain_hash(prefix_hash)
     }
 
-    /// Checks the bloom filter for a key hash.
+    /// Checks the bloom filter for a precomputed key hash.
     ///
     /// Returns `Ok(true)` if the key may exist in this table (or if no
     /// filter is available), `Ok(false)` if the key is definitely absent.
     ///
     /// Used by the point-read merge pipeline to pre-filter disk tables
     /// before building range iterators.
-    pub(crate) fn bloom_may_contain_key(&self, key_hash: u64) -> crate::Result<bool> {
+    pub(crate) fn bloom_may_contain_key_hash(&self, key_hash: u64) -> crate::Result<bool> {
         self.bloom_may_contain_hash(key_hash)
     }
 
diff --git a/src/tree/mod.rs b/src/tree/mod.rs
index 0bf2779dc..98d9aef36 100644
--- a/src/tree/mod.rs
+++ b/src/tree/mod.rs
@@ -780,8 +780,8 @@ impl Tree {
     /// reuses the unified merge/RT/Indirection resolution logic from `MvccStream`
     /// instead of duplicating it in a hand-rolled collection loop.
     ///
-    /// Bloom pre-filtering ensures that 95%+ of disk tables are rejected at the
-    /// filter level, preserving O(1) reject performance on deep LSM trees.
+    /// Bloom pre-filtering can reject many disk tables at the filter level,
+    /// which typically improves point-read performance on deep LSM trees.
     fn resolve_merge_via_pipeline(
         version: SuperVersion,
         key: &[u8],
diff --git a/tests/merge_operator.rs b/tests/merge_operator.rs
index 84631b38c..73f54c5e5 100644
--- a/tests/merge_operator.rs
+++ b/tests/merge_operator.rs
@@ -994,11 +994,11 @@ fn merge_operands_across_active_and_disk() -> lsm_tree::Result<()> {
     Ok(())
 }
 
-/// Bloom pre-filter skips tables whose key_range overlaps but whose bloom
-/// filter reports the key absent. Exercises the Ok(false) → skip path in
-/// the bloom-filtered iterator pipeline (bloom_passes + key_hash).
+/// Merge correctness when bloom pre-filtering is enabled and there exists an
+/// overlapping table whose bloom filter reports the key absent. This ensures
+/// the extra overlapping table does not affect the merged result.
 #[test]
-fn merge_bloom_skips_non_matching_tables() -> lsm_tree::Result<()> {
+fn merge_bloom_with_overlapping_non_matching_table() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;
     let tree = open_tree_with_counter(&folder);
 
diff --git a/tests/tree_prefix_bloom.rs b/tests/tree_prefix_bloom.rs
index 5b6b1cf1b..f11ce6a28 100644
--- a/tests/tree_prefix_bloom.rs
+++ b/tests/tree_prefix_bloom.rs
@@ -536,6 +536,7 @@ fn prefix_bloom_negative_lookup_in_key_range_gap() -> lsm_tree::Result<()> {
     Ok(())
 }
 
+<<<<<<< HEAD
 /// Multi-table runs (typically L0) now support per-table prefix bloom
 /// skipping. This test creates multiple flushes WITHOUT compaction so
 /// the tables remain in a single multi-table L0 run, then verifies

From 7f36218f920e98de0aef5bab41970c007efb6959 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 12:15:49 +0200
Subject: [PATCH 04/10] docs: remove stale line references and clarify test
 docstrings

- Remove stale line number reference from merge_rt test comment
- Reword prefix bloom test docstring to describe correctness
  validation independent of bloom filter behavior
- Add comment explaining why multi-table runs skip bloom filtering
---
 tests/merge_operator.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/merge_operator.rs b/tests/merge_operator.rs
index 73f54c5e5..a7661ea41 100644
--- a/tests/merge_operator.rs
+++ b/tests/merge_operator.rs
@@ -898,7 +898,7 @@ fn merge_rt_no_operator_get_and_multi_get_agree() -> lsm_tree::Result<()> {
 
 /// RT suppresses operand in disk range scan during merge resolution.
 /// Exercises the is_rt_suppressed path inside the table.range() fallback
-/// in resolve_merge_via_pipeline (line ~909).
+/// in resolve_merge_via_pipeline.
 #[test]
 fn merge_rt_suppresses_operand_in_disk_range_scan() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;

From 8e6ba0f78b7ac5e511e087c440c8bc7528268d20 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 12:33:05 +0200
Subject: [PATCH 05/10] fix: remove leftover conflict marker in
 tree_prefix_bloom test

---
 tests/tree_prefix_bloom.rs | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/tree_prefix_bloom.rs b/tests/tree_prefix_bloom.rs
index f11ce6a28..5b6b1cf1b 100644
--- a/tests/tree_prefix_bloom.rs
+++ b/tests/tree_prefix_bloom.rs
@@ -536,7 +536,6 @@ fn prefix_bloom_negative_lookup_in_key_range_gap() -> lsm_tree::Result<()> {
     Ok(())
 }
 
-<<<<<<< HEAD
 /// Multi-table runs (typically L0) now support per-table prefix bloom
 /// skipping. This test creates multiple flushes WITHOUT compaction so
 /// the tables remain in a single multi-table L0 run, then verifies

From 52deaef76b72a96b9234556cedb7f6d71e4d8014 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 13:08:58 +0200
Subject: [PATCH 06/10] docs: remove stale line ref and clarify pipeline design
 choice

- Remove brittle line number reference from merge_tombstone test
- Add code comment explaining intentional TreeIter reuse in
  resolve_merge_via_pipeline (bloom pre-filter + shared logic)
---
 src/tree/mod.rs         | 3 +++
 tests/merge_operator.rs | 2 +-
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/src/tree/mod.rs b/src/tree/mod.rs
index 98d9aef36..5b9b58338 100644
--- a/src/tree/mod.rs
+++ b/src/tree/mod.rs
@@ -804,6 +804,9 @@ impl Tree {
             metrics: None,
         };
 
+        // Intentionally reuses the full TreeIter pipeline (with bloom pre-filter)
+        // rather than a hand-rolled loop, to share merge/RT/Indirection logic
+        // with range scans. The bloom hash skips most tables at the filter level.
         let mut iter = TreeIter::create_range(iter_state, range, seqno);
 
         match iter.next() {
diff --git a/tests/merge_operator.rs b/tests/merge_operator.rs
index a7661ea41..0bb39df9f 100644
--- a/tests/merge_operator.rs
+++ b/tests/merge_operator.rs
@@ -945,7 +945,7 @@ fn merge_disk_base_via_point_lookup() -> lsm_tree::Result<()> {
 }
 
 /// Merge with Tombstone base in sealed memtable — exercises sealed memtable
-/// scan path in resolve_merge_via_pipeline (lines ~868-877).
+/// scan path in resolve_merge_via_pipeline.
 #[test]
 fn merge_tombstone_in_sealed_memtable() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;

From 94f767cecc723a2a5a87dc3bb53e9a8581649305 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 13:52:08 +0200
Subject: [PATCH 07/10] docs(bloom): note conservative fallback for
 partitioned/TLI filters

---
 src/table/mod.rs | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/src/table/mod.rs b/src/table/mod.rs
index 1cff81603..09969e3c5 100644
--- a/src/table/mod.rs
+++ b/src/table/mod.rs
@@ -837,7 +837,9 @@ impl Table {
     /// filter is available), `Ok(false)` if the key is definitely absent.
     ///
     /// Used by the point-read merge pipeline to pre-filter disk tables
-    /// before building range iterators.
+    /// before building range iterators. For partitioned or TLI filter
+    /// configurations, the underlying check returns `Ok(true)` conservatively,
+    /// so pre-filtering is best-effort and configuration-dependent.
     pub(crate) fn bloom_may_contain_key_hash(&self, key_hash: u64) -> crate::Result<bool> {
         self.bloom_may_contain_hash(key_hash)
     }

From 56aafdbf54d31fcc311df08fe81dbf5abfbfc93e Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 15:02:44 +0200
Subject: [PATCH 08/10] fix(bloom): apply key_hash filtering to multi-table
 runs

- Broaden multi-table run bloom guard to trigger on key_hash too
  (was prefix_hash-only, so point-read merges skipped bloom pruning)
- Replace inline prefix bloom check with bloom_passes() for
  consistent prefix+key bloom handling and metrics
- Soften test comments to not assert definite bloom behavior
- Rename prefix bloom test to describe correctness scenario
---
 src/range.rs               | 30 ++++++------------------------
 tests/merge_operator.rs    | 11 +++++------
 tests/tree_prefix_bloom.rs |  4 ++--
 3 files changed, 13 insertions(+), 32 deletions(-)

diff --git a/src/range.rs b/src/range.rs
index 8cfe9aef6..d77f9b90d 100644
--- a/src/range.rs
+++ b/src/range.rs
@@ -291,9 +291,11 @@ impl TreeIter {
                             );
                         }
 
-                        // If a prefix hash is available, filter individual tables
-                        // within the multi-table run using their bloom filters.
-                        if let Some(prefix_hash) = lock.prefix_hash {
+                        // If a prefix or key hash is available, filter individual
+                        // tables within the multi-table run using their bloom
+                        // filters. This covers both prefix scans (prefix_hash)
+                        // and point-read merge pipelines (key_hash).
+                        if lock.prefix_hash.is_some() || lock.key_hash.is_some() {
                             let bounds = (
                                 user_range.0.as_ref().map(std::convert::AsRef::as_ref),
                                 user_range.1.as_ref().map(std::convert::AsRef::as_ref),
@@ -308,27 +310,7 @@ impl TreeIter {
                                         return false;
                                     }
 
-                                    // On I/O error reading the filter, include the
-                                    // table conservatively to avoid missing data.
-                                    let contains = table
-                                        .maybe_contains_prefix(prefix_hash)
-                                        .inspect_err(|e| {
-                                            log::debug!(
-                                                "prefix bloom check failed for table {:?}: {e}",
-                                                table.id(),
-                                            );
-                                        })
-                                        .unwrap_or(true);
-
-                                    #[cfg(feature = "metrics")]
-                                    if !contains {
-                                        if let Some(m) = &lock.metrics {
-                                            m.prefix_bloom_skips
-                                                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-                                        }
-                                    }
-
-                                    contains
+                                    bloom_passes(lock, table)
                                 })
                                 .cloned()
                                 .collect();
diff --git a/tests/merge_operator.rs b/tests/merge_operator.rs
index 0bb39df9f..a5a3de9be 100644
--- a/tests/merge_operator.rs
+++ b/tests/merge_operator.rs
@@ -1003,22 +1003,21 @@ fn merge_bloom_with_overlapping_non_matching_table() -> lsm_tree::Result<()> {
     let tree = open_tree_with_counter(&folder);
 
     // Table 1: wide key range [aaa, zzz] that does NOT contain "counter".
-    // Its key_range overlaps "counter" but bloom will reject it.
+    // Its key_range overlaps "counter"; bloom may reject it (best-effort).
     tree.insert("aaa", 0_i64.to_le_bytes(), 0);
     tree.insert("zzz", 0_i64.to_le_bytes(), 1);
     tree.flush_active_memtable(0)?;
 
-    // Table 2: contains "counter" base value — bloom will accept it.
+    // Table 2: contains "counter" base value.
     tree.insert("counter", 100_i64.to_le_bytes(), 2);
     tree.flush_active_memtable(0)?;
 
     // Merge operand in active memtable
     tree.merge("counter", 10_i64.to_le_bytes(), 3);
 
-    // resolve_merge_via_pipeline builds a key..=key range with bloom hash:
-    //   Table 1: key_range [aaa,zzz] overlaps "counter" ✓, bloom → Ok(false) → SKIP
-    //   Table 2: key_range [counter,counter] overlaps ✓, bloom → Ok(true) → INCLUDE
-    //   merge(Some(100), [10]) = 110
+    // resolve_merge_via_pipeline builds a key..=key range with bloom hash.
+    // Table 1 does not contain "counter" so it contributes nothing.
+    // merge(Some(100), [10]) = 110
     assert_eq!(Some(110), get_counter(&tree, "counter", 4));
 
     Ok(())
diff --git a/tests/tree_prefix_bloom.rs b/tests/tree_prefix_bloom.rs
index 5b6b1cf1b..5ceea27ad 100644
--- a/tests/tree_prefix_bloom.rs
+++ b/tests/tree_prefix_bloom.rs
@@ -869,9 +869,9 @@ fn prefix_bloom_skip_metrics_zero_without_extractor() -> lsm_tree::Result<()> {
 /// Prefix scan correctness when an L0 single-table run has a wide key range
 /// that overlaps the scanned prefix but does not contain any matching keys.
 /// Ensures the non-matching table does not affect scan results, independent of
-/// whether the prefix bloom rejects the scanned prefix.
+/// whether the prefix bloom happens to reject the scanned prefix.
 #[test]
-fn prefix_bloom_rejects_in_l0_single_table_run() -> lsm_tree::Result<()> {
+fn prefix_scan_l0_wide_non_matching_table_does_not_affect_results() -> lsm_tree::Result<()> {
     let folder = tempfile::tempdir()?;
     let tree = tree_with_prefix_bloom(&folder)?;
 

From 4836e82886258bfced22e144b71b86bceb0f2af4 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 16:06:58 +0200
Subject: [PATCH 09/10] perf(bench): add merge point-read latency benchmark for
 deep L0

Criterion benchmark measuring point-read merge resolution latency
on trees with 10/50/100 L0 tables, both cached and uncached.
Exercises the bloom-filtered iterator pipeline.
---
 Cargo.toml                  |   6 ++
 benches/merge_point_read.rs | 116 ++++++++++++++++++++++++++++++++++++
 2 files changed, 122 insertions(+)
 create mode 100644 benches/merge_point_read.rs

diff --git a/Cargo.toml b/Cargo.toml
index 32784f5c6..3f02c72bf 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -117,3 +117,9 @@ name = "prefix_bloom"
 harness = false
 path = "benches/prefix_bloom.rs"
 required-features = []
+
+[[bench]]
+name = "merge_point_read"
+harness = false
+path = "benches/merge_point_read.rs"
+required-features = []
diff --git a/benches/merge_point_read.rs b/benches/merge_point_read.rs
new file mode 100644
index 000000000..517069de1
--- /dev/null
+++ b/benches/merge_point_read.rs
@@ -0,0 +1,116 @@
+use criterion::{criterion_group, criterion_main, Criterion};
+use lsm_tree::{AbstractTree, Cache, Config, MergeOperator, SequenceNumberCounter, UserValue};
+use std::sync::Arc;
+use tempfile::tempdir;
+
+/// Simple counter merge operator for benchmarks.
+struct CounterMerge;
+
+impl MergeOperator for CounterMerge {
+    fn merge(
+        &self,
+        _key: &[u8],
+        base_value: Option<&[u8]>,
+        operands: &[&[u8]],
+    ) -> lsm_tree::Result<UserValue> {
+        let mut counter: i64 = match base_value {
+            Some(bytes) if bytes.len() == 8 => {
+                i64::from_le_bytes(bytes.try_into().expect("checked"))
+            }
+            _ => 0,
+        };
+        for op in operands {
+            if op.len() == 8 {
+                counter += i64::from_le_bytes((*op).try_into().expect("checked"));
+            }
+        }
+        Ok(counter.to_le_bytes().to_vec().into())
+    }
+}
+
+fn merge_point_read_deep_tree(c: &mut Criterion) {
+    let mut group = c.benchmark_group("merge point read");
+    group.sample_size(100);
+
+    for table_count in [10, 50, 100] {
+        // --- Uncached: cold disk reads ---
+        let folder = tempdir().unwrap();
+        let tree = Config::new(
+            &folder,
+            SequenceNumberCounter::default(),
+            SequenceNumberCounter::default(),
+        )
+        .use_cache(Arc::new(Cache::with_capacity_bytes(0)))
+        .with_merge_operator(Some(Arc::new(CounterMerge)))
+        .open()
+        .unwrap();
+
+        let mut seqno = 0u64;
+
+        // Base value on disk
+        tree.insert("counter", 100_i64.to_le_bytes(), seqno);
+        seqno += 1;
+        tree.flush_active_memtable(0).unwrap();
+
+        // Create many tables with unrelated keys (bloom should reject these)
+        for i in 1..table_count {
+            let key = format!("other_{i:04}");
+            tree.insert(key, 0_i64.to_le_bytes(), seqno);
+            seqno += 1;
+            tree.flush_active_memtable(0).unwrap();
+        }
+
+        // Merge operand in active memtable
+        tree.merge("counter", 1_i64.to_le_bytes(), seqno);
+        seqno += 1;
+
+        group.bench_function(format!("merge get, {table_count} tables (uncached)"), |b| {
+            b.iter(|| {
+                let val = tree.get("counter", seqno).unwrap().unwrap();
+                let n = i64::from_le_bytes((*val).try_into().unwrap());
+                assert_eq!(n, 101);
+            });
+        });
+
+        // --- Cached: warm block cache ---
+        let folder2 = tempdir().unwrap();
+        let tree_cached = Config::new(
+            &folder2,
+            SequenceNumberCounter::default(),
+            SequenceNumberCounter::default(),
+        )
+        .use_cache(Arc::new(Cache::with_capacity_bytes(64 * 1_024 * 1_024)))
+        .with_merge_operator(Some(Arc::new(CounterMerge)))
+        .open()
+        .unwrap();
+
+        let mut s = 0u64;
+        tree_cached.insert("counter", 100_i64.to_le_bytes(), s);
+        s += 1;
+        tree_cached.flush_active_memtable(0).unwrap();
+
+        for i in 1..table_count {
+            let key = format!("other_{i:04}");
+            tree_cached.insert(key, 0_i64.to_le_bytes(), s);
+            s += 1;
+            tree_cached.flush_active_memtable(0).unwrap();
+        }
+
+        tree_cached.merge("counter", 1_i64.to_le_bytes(), s);
+        s += 1;
+
+        // Warm the cache
+        let _ = tree_cached.get("counter", s).unwrap();
+
+        group.bench_function(format!("merge get, {table_count} tables (cached)"), |b| {
+            b.iter(|| {
+                let val = tree_cached.get("counter", s).unwrap().unwrap();
+                let n = i64::from_le_bytes((*val).try_into().unwrap());
+                assert_eq!(n, 101);
+            });
+        });
+    }
+}
+
+criterion_group!(benches, merge_point_read_deep_tree);
+criterion_main!(benches);

From 3d97a409cbf9e2c6ea93ae1e0623c61128defa76 Mon Sep 17 00:00:00 2001
From: Dmitry Prudnikov <mail@polaz.com>
Date: Sun, 22 Mar 2026 16:21:07 +0200
Subject: [PATCH 10/10] refactor(bench): extract populate_merge_tree helper

---
 benches/merge_point_read.rs | 69 ++++++++++++++++++-------------------
 1 file changed, 33 insertions(+), 36 deletions(-)

diff --git a/benches/merge_point_read.rs b/benches/merge_point_read.rs
index 517069de1..b91f3a7fe 100644
--- a/benches/merge_point_read.rs
+++ b/benches/merge_point_read.rs
@@ -1,5 +1,7 @@
 use criterion::{criterion_group, criterion_main, Criterion};
-use lsm_tree::{AbstractTree, Cache, Config, MergeOperator, SequenceNumberCounter, UserValue};
+use lsm_tree::{
+    AbstractTree, AnyTree, Cache, Config, MergeOperator, SequenceNumberCounter, UserValue,
+};
 use std::sync::Arc;
 use tempfile::tempdir;
 
@@ -28,11 +30,36 @@ impl MergeOperator for CounterMerge {
     }
 }
 
+/// Populates a tree with a base value + N-1 unrelated tables + 1 merge operand.
+/// Returns the seqno to use for reads.
+fn populate_merge_tree(tree: &AnyTree, table_count: u64) -> u64 {
+    let mut seqno = 0u64;
+
+    // Base value on disk
+    tree.insert("counter", 100_i64.to_le_bytes(), seqno);
+    seqno += 1;
+    tree.flush_active_memtable(0).unwrap();
+
+    // Create many tables with unrelated keys (bloom should reject these)
+    for i in 1..table_count {
+        let key = format!("other_{i:04}");
+        tree.insert(key, 0_i64.to_le_bytes(), seqno);
+        seqno += 1;
+        tree.flush_active_memtable(0).unwrap();
+    }
+
+    // Merge operand in active memtable
+    tree.merge("counter", 1_i64.to_le_bytes(), seqno);
+    seqno += 1;
+
+    seqno
+}
+
 fn merge_point_read_deep_tree(c: &mut Criterion) {
     let mut group = c.benchmark_group("merge point read");
     group.sample_size(100);
 
-    for table_count in [10, 50, 100] {
+    for table_count in [10u64, 50, 100] {
         // --- Uncached: cold disk reads ---
         let folder = tempdir().unwrap();
         let tree = Config::new(
@@ -45,24 +72,7 @@ fn merge_point_read_deep_tree(c: &mut Criterion) {
         .open()
         .unwrap();
 
-        let mut seqno = 0u64;
-
-        // Base value on disk
-        tree.insert("counter", 100_i64.to_le_bytes(), seqno);
-        seqno += 1;
-        tree.flush_active_memtable(0).unwrap();
-
-        // Create many tables with unrelated keys (bloom should reject these)
-        for i in 1..table_count {
-            let key = format!("other_{i:04}");
-            tree.insert(key, 0_i64.to_le_bytes(), seqno);
-            seqno += 1;
-            tree.flush_active_memtable(0).unwrap();
-        }
-
-        // Merge operand in active memtable
-        tree.merge("counter", 1_i64.to_le_bytes(), seqno);
-        seqno += 1;
+        let seqno = populate_merge_tree(&tree, table_count);
 
         group.bench_function(format!("merge get, {table_count} tables (uncached)"), |b| {
             b.iter(|| {
@@ -84,27 +94,14 @@ fn merge_point_read_deep_tree(c: &mut Criterion) {
         .open()
         .unwrap();
 
-        let mut s = 0u64;
-        tree_cached.insert("counter", 100_i64.to_le_bytes(), s);
-        s += 1;
-        tree_cached.flush_active_memtable(0).unwrap();
-
-        for i in 1..table_count {
-            let key = format!("other_{i:04}");
-            tree_cached.insert(key, 0_i64.to_le_bytes(), s);
-            s += 1;
-            tree_cached.flush_active_memtable(0).unwrap();
-        }
-
-        tree_cached.merge("counter", 1_i64.to_le_bytes(), s);
-        s += 1;
+        let seqno_cached = populate_merge_tree(&tree_cached, table_count);
 
         // Warm the cache
-        let _ = tree_cached.get("counter", s).unwrap();
+        let _ = tree_cached.get("counter", seqno_cached).unwrap();
 
         group.bench_function(format!("merge get, {table_count} tables (cached)"), |b| {
             b.iter(|| {
-                let val = tree_cached.get("counter", s).unwrap().unwrap();
+                let val = tree_cached.get("counter", seqno_cached).unwrap().unwrap();
                 let n = i64::from_le_bytes((*val).try_into().unwrap());
                 assert_eq!(n, 101);
             });