feat: merge operators for commutative LSM operations

src/abstract_tree.rs

@@ -123,7 +123,8 @@ pub trait AbstractTree: sealed::Sealed {
                 .map(|mt| mt.iter().map(Ok))
                 .collect::<Vec<_>>(),
         );
-        let stream = CompactionStream::new(merger, seqno_threshold);
+        let stream = CompactionStream::new(merger, seqno_threshold)
+            .with_merge_operator(self.tree_config().merge_operator.clone());
 
         drop(version_history);
 
@@ -700,6 +701,41 @@ pub trait AbstractTree: sealed::Sealed {
     /// Will return `Err` if an IO error occurs.
     fn remove<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> (u64, u64);
 
+    /// Writes a merge operand for a key.
+    ///
+    /// The operand is stored as a partial update that will be combined with
+    /// other operands and/or a base value via the configured [`crate::MergeOperator`]
+    /// during reads and compaction.
+    ///
+    /// Returns the added item's size and new size of the memtable.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # let folder = tempfile::tempdir()?;
+    /// # use lsm_tree::{AbstractTree, Config, MergeOperator, UserValue};
+    /// # use std::sync::Arc;
+    /// # struct SumMerge;
+    /// # impl MergeOperator for SumMerge {
+    /// #     fn merge(&self, _key: &[u8], base: Option<&[u8]>, operands: &[&[u8]]) -> lsm_tree::Result<UserValue> {
+    /// #         let mut sum: i64 = base.map_or(0, |b| i64::from_le_bytes(b.try_into().unwrap()));
+    /// #         for op in operands { sum += i64::from_le_bytes((*op).try_into().unwrap()); }
+    /// #         Ok(sum.to_le_bytes().to_vec().into())
+    /// #     }
+    /// # }
+    /// # let tree = Config::new(folder, Default::default(), Default::default())
+    /// #     .with_merge_operator(Some(Arc::new(SumMerge)))
+    /// #     .open()?;
+    /// tree.merge("counter", 1_i64.to_le_bytes(), 0);
+    /// # Ok::<(), lsm_tree::Error>(())
+    /// ```
+    fn merge<K: Into<UserKey>, V: Into<UserValue>>(
+        &self,
+        key: K,
+        operand: V,
+        seqno: SeqNo,
+    ) -> (u64, u64);
+
     /// Removes an item from the tree.
     ///
     /// The tombstone marker of this delete operation will vanish when it

src/blob_tree/mod.rs

@@ -255,6 +255,7 @@ impl AbstractTree for BlobTree {
                 &range,
                 seqno,
                 index,
+                None, // BlobTree does not use merge operators for prefix scans
                 prefix_hash,
             )
             .map(move |kv| {
@@ -277,15 +278,14 @@ impl AbstractTree for BlobTree {
         let tree = self.clone();
 
         Box::new(
-            crate::Tree::create_internal_range(super_version.clone(), &range, seqno, index).map(
-                move |kv| {
+            crate::Tree::create_internal_range(super_version.clone(), &range, seqno, index, None)
+                .map(move |kv| {
                     IterGuardImpl::Blob(Guard {
                         tree: tree.clone(),
                         version: super_version.version.clone(),
                         kv,
                     })
-                },
-            ),
+                }),
         )
     }
 
@@ -661,6 +661,15 @@ impl AbstractTree for BlobTree {
             .collect()
     }
 
+    fn merge<K: Into<UserKey>, V: Into<UserValue>>(
+        &self,
+        key: K,
+        operand: V,
+        seqno: SeqNo,
+    ) -> (u64, u64) {
+        self.index.merge(key, operand, seqno)
+    }
+
     fn remove<K: Into<UserKey>>(&self, key: K, seqno: SeqNo) -> (u64, u64) {
         self.index.remove(key, seqno)
     }

src/compaction/filter.rs

@@ -141,7 +141,9 @@ impl<'a> ItemAccessor<'a> {
     /// This method will return an error if blob retrieval fails.
     pub fn value(&self) -> crate::Result<UserValue> {
         match self.item.key.value_type {
-            crate::ValueType::Value => Ok(self.item.value.clone()),
+            // MergeOperand: return raw operand bytes so filters can inspect them.
+            // The compaction pipeline preserves MergeOperand type on Replace.
+            crate::ValueType::Value | crate::ValueType::MergeOperand => Ok(self.item.value.clone()),
             crate::ValueType::Indirection => {
                 // Resolve and read the value from a blob file.
                 let mut reader = &self.item.value[..];