Expose log_data blobs in WriteBatchIterator and WriteBatchIteratorCf

Author: lucasvuillier

CHANGELOG.md

@@ -1,5 +1,15 @@
 # Changelog
 
+## Unreleased
+
+- feat: expose `log_data` blobs during write-batch iteration.
+  `WriteBatchIterator` and `WriteBatchIteratorCf` both grow an optional
+  `log_data(&mut self, blob: &[u8])` method (default: no-op) that is
+  invoked for each blob written via `WriteBatchWithTransaction::put_log_data`
+  when iterating with `iterate()` / `iterate_cf()`.
+  Previously these blobs were silently dropped at the Rust layer even though
+  the underlying C API forwarded them. (lucas.vuillier)
+
 ## 0.50.0 (2026-05-23)
 
 - breaking: bump MSRV to 1.91.0 per the rolling 6-month policy. 1.91.0

librocksdb-sys/c-api-extensions/c_api_extensions.cc

@@ -17,6 +17,7 @@
 #include "rocksdb/listener.h"
 #include "rocksdb/options.h"
 #include "rocksdb/table.h"
+#include "rocksdb/write_batch.h"
 
 using ROCKSDB_NAMESPACE::BackgroundErrorRecoveryInfo;
 using ROCKSDB_NAMESPACE::BlockBasedTableOptions;
@@ -28,8 +29,10 @@ using ROCKSDB_NAMESPACE::ExternalFileIngestionInfo;
 using ROCKSDB_NAMESPACE::FlushJobInfo;
 using ROCKSDB_NAMESPACE::Options;
 using ROCKSDB_NAMESPACE::ReadOptions;
+using ROCKSDB_NAMESPACE::Slice;
 using ROCKSDB_NAMESPACE::Status;
 using ROCKSDB_NAMESPACE::SubcompactionJobInfo;
+using ROCKSDB_NAMESPACE::WriteBatch;
 using ROCKSDB_NAMESPACE::WriteStallInfo;
 using ROCKSDB_NAMESPACE::MemTableInfo;
 
@@ -350,3 +353,109 @@ extern "C" double rocksdb_compactoptions_get_blob_garbage_collection_age_cutoff(
     rocksdb_compactoptions_t* opt) {
   return reinterpret_cast<CompactRangeOptions*>(opt)->blob_garbage_collection_age_cutoff;
 }
+
+// -----------------------------------------------------------------------------
+// WriteBatch iteration with log_data support
+//
+// rocksdb_writebatch_t is defined in rocksdb/db/c.cc as:
+//   struct rocksdb_writebatch_t { WriteBatch rep; };
+// The `rep` field is the first (and only) member, so a pointer to the opaque
+// C type is also a valid pointer to its embedded WriteBatch — same layout
+// trick used for ReadOptions, Options, and BlockBasedTableOptions above.
+// -----------------------------------------------------------------------------
+
+namespace {
+
+class RustWriteBatchHandler : public WriteBatch::Handler {
+ public:
+  void* state;
+  void (*put_fn)(void*, const char*, size_t, const char*, size_t);
+  void (*delete_fn)(void*, const char*, size_t);
+  void (*log_data_fn)(void*, const char*, size_t);
+
+  Status PutCF(uint32_t cf, const Slice& key, const Slice& value) override {
+    if (cf == 0 && put_fn != nullptr) {
+      put_fn(state, key.data(), key.size(), value.data(), value.size());
+    }
+    return Status::OK();
+  }
+
+  Status DeleteCF(uint32_t cf, const Slice& key) override {
+    if (cf == 0 && delete_fn != nullptr) {
+      delete_fn(state, key.data(), key.size());
+    }
+    return Status::OK();
+  }
+
+  void LogData(const Slice& blob) override {
+    if (log_data_fn != nullptr) {
+      log_data_fn(state, blob.data(), blob.size());
+    }
+  }
+};
+
+class RustWriteBatchCfHandler : public WriteBatch::Handler {
+ public:
+  void* state;
+  void (*put_cf_fn)(void*, uint32_t, const char*, size_t, const char*, size_t);
+  void (*delete_cf_fn)(void*, uint32_t, const char*, size_t);
+  void (*merge_cf_fn)(void*, uint32_t, const char*, size_t, const char*, size_t);
+  void (*log_data_fn)(void*, const char*, size_t);
+
+  Status PutCF(uint32_t cf, const Slice& key, const Slice& value) override {
+    if (put_cf_fn != nullptr) {
+      put_cf_fn(state, cf, key.data(), key.size(), value.data(), value.size());
+    }
+    return Status::OK();
+  }
+
+  Status DeleteCF(uint32_t cf, const Slice& key) override {
+    if (delete_cf_fn != nullptr) {
+      delete_cf_fn(state, cf, key.data(), key.size());
+    }
+    return Status::OK();
+  }
+
+  Status MergeCF(uint32_t cf, const Slice& key, const Slice& value) override {
+    if (merge_cf_fn != nullptr) {
+      merge_cf_fn(state, cf, key.data(), key.size(), value.data(), value.size());
+    }
+    return Status::OK();
+  }
+
+  void LogData(const Slice& blob) override {
+    if (log_data_fn != nullptr) {
+      log_data_fn(state, blob.data(), blob.size());
+    }
+  }
+};
+
+}  // namespace
+
+extern "C" void rust_rocksdb_writebatch_iterate(
+    rocksdb_writebatch_t* b, void* state,
+    void (*put)(void*, const char*, size_t, const char*, size_t),
+    void (*deleted)(void*, const char*, size_t),
+    void (*log_data)(void*, const char*, size_t)) {
+  RustWriteBatchHandler handler;
+  handler.state = state;
+  handler.put_fn = put;
+  handler.delete_fn = deleted;
+  handler.log_data_fn = log_data;
+  reinterpret_cast<WriteBatch*>(b)->Iterate(&handler);
+}
+
+extern "C" void rust_rocksdb_writebatch_iterate_cf(
+    rocksdb_writebatch_t* b, void* state,
+    void (*put_cf)(void*, uint32_t, const char*, size_t, const char*, size_t),
+    void (*deleted_cf)(void*, uint32_t, const char*, size_t),
+    void (*merge_cf)(void*, uint32_t, const char*, size_t, const char*, size_t),
+    void (*log_data)(void*, const char*, size_t)) {
+  RustWriteBatchCfHandler handler;
+  handler.state = state;
+  handler.put_cf_fn = put_cf;
+  handler.delete_cf_fn = deleted_cf;
+  handler.merge_cf_fn = merge_cf;
+  handler.log_data_fn = log_data;
+  reinterpret_cast<WriteBatch*>(b)->Iterate(&handler);
+}

librocksdb-sys/c-api-extensions/c_api_extensions.h

@@ -164,6 +164,29 @@ extern ROCKSDB_LIBRARY_API void rust_rocksdb_eventlistener_destroy(
 extern ROCKSDB_LIBRARY_API void rust_rocksdb_options_add_eventlistener(
     rocksdb_options_t*, rust_rocksdb_eventlistener_t*);
 
+/* -------------------------------------------------------------------------
+ * WriteBatch iteration with log_data support
+ *
+ * The upstream rocksdb_writebatch_iterate / rocksdb_writebatch_iterate_cf
+ * do not expose the WriteBatch::Handler::LogData callback, so blobs written
+ * via rocksdb_writebatch_put_log_data are silently dropped during iteration.
+ * These wrappers add a log_data parameter so callers can observe those blobs.
+ * ------------------------------------------------------------------------- */
+extern ROCKSDB_LIBRARY_API void rust_rocksdb_writebatch_iterate(
+    rocksdb_writebatch_t*, void* state,
+    void (*put)(void*, const char* k, size_t klen, const char* v, size_t vlen),
+    void (*deleted)(void*, const char* k, size_t klen),
+    void (*log_data)(void*, const char* blob, size_t blen));
+
+extern ROCKSDB_LIBRARY_API void rust_rocksdb_writebatch_iterate_cf(
+    rocksdb_writebatch_t*, void* state,
+    void (*put_cf)(void*, uint32_t cfid, const char* k, size_t klen,
+                   const char* v, size_t vlen),
+    void (*deleted_cf)(void*, uint32_t cfid, const char* k, size_t klen),
+    void (*merge_cf)(void*, uint32_t cfid, const char* k, size_t klen,
+                     const char* v, size_t vlen),
+    void (*log_data)(void*, const char* blob, size_t blen));
+
 #ifdef __cplusplus
 }
 #endif

librocksdb-sys/tests/ffi.rs

@@ -192,6 +192,14 @@ unsafe extern "C" fn CheckDel(ptr: *mut c_void, k: *const c_char, klen: size_t)
     *state += 1;
 }
 
+// Callback from rocksdb_writebatch_iterate()
+unsafe extern "C" fn CheckLogData(ptr: *mut c_void, blob: *const c_char, blen: size_t) {
+    let mut state: *mut c_int = ptr as *mut c_int;
+    CheckCondition!(*state == 3);
+    CheckEqual(cstrp!("blob"), blob, blen);
+    *state += 1;
+}
+
 unsafe extern "C" fn CmpDestroy(arg: *mut c_void) {}
 
 unsafe extern "C" fn CmpCompare(
@@ -566,6 +574,25 @@ fn ffi() {
             rocksdb_writebatch_destroy(wb);
         }
 
+        StartPhase("writebatch_log_data");
+        {
+            let wb = rocksdb_writebatch_create();
+            rocksdb_writebatch_put(wb, cstrp!("bar"), 3, cstrp!("b"), 1);
+            rocksdb_writebatch_put(wb, cstrp!("box"), 3, cstrp!("c"), 1);
+            rocksdb_writebatch_put_log_data(wb, cstrp!("blob"), 4);
+            rocksdb_writebatch_delete(wb, cstrp!("bar"), 3);
+            let mut pos: c_int = 0;
+            rust_rocksdb_writebatch_iterate(
+                wb,
+                (&mut pos as *mut c_int).cast::<c_void>(),
+                Some(CheckPut),
+                Some(CheckDel),
+                Some(CheckLogData),
+            );
+            CheckCondition!(pos == 4);
+            rocksdb_writebatch_destroy(wb);
+        }
+
         StartPhase("writebatch_vectors");
         {
             let wb = rocksdb_writebatch_create();

src/write_batch.rs

@@ -53,7 +53,7 @@ pub struct WriteBatchWithTransaction<const TRANSACTION: bool> {
     pub(crate) inner: *mut ffi::rocksdb_writebatch_t,
 }
 
-/// Receives the puts and deletes of a write batch.
+/// Receives the puts, deletes, and log data of a write batch.
 ///
 /// The application must provide an implementation of this trait when
 /// iterating the operations within a `WriteBatch`
@@ -62,6 +62,10 @@ pub trait WriteBatchIterator {
     fn put(&mut self, key: &[u8], value: &[u8]);
     /// Called with a key that was `delete`d from the batch.
     fn delete(&mut self, key: &[u8]);
+    /// Called with a blob written via [`WriteBatchWithTransaction::put_log_data`].
+    ///
+    /// The default implementation is a no-op; override to observe log data.
+    fn log_data(&mut self, _blob: &[u8]) {}
 }
 
 /// Receives the puts, deletes, and merges of a write batch with column family
@@ -84,6 +88,10 @@ pub trait WriteBatchIteratorCf {
     /// Merge operations combine the provided value with the existing value at
     /// the key using a database-defined merge operator.
     fn merge_cf(&mut self, cf_id: u32, key: &[u8], value: &[u8]);
+    /// Called with a blob written via [`WriteBatchWithTransaction::put_log_data`].
+    ///
+    /// The default implementation is a no-op; override to observe log data.
+    fn log_data(&mut self, _blob: &[u8]) {}
 }
 
 unsafe extern "C" fn writebatch_put_callback<T: WriteBatchIterator>(
@@ -158,6 +166,30 @@ unsafe extern "C" fn writebatch_merge_cf_callback<T: WriteBatchIteratorCf>(
     }
 }
 
+unsafe extern "C" fn writebatch_log_data_callback<T: WriteBatchIterator>(
+    state: *mut c_void,
+    blob: *const c_char,
+    blen: usize,
+) {
+    unsafe {
+        let callbacks = &mut *(state as *mut T);
+        let blob = slice::from_raw_parts(blob as *const u8, blen);
+        callbacks.log_data(blob);
+    }
+}
+
+unsafe extern "C" fn writebatch_log_data_cf_callback<T: WriteBatchIteratorCf>(
+    state: *mut c_void,
+    blob: *const c_char,
+    blen: usize,
+) {
+    unsafe {
+        let callbacks = &mut *(state as *mut T);
+        let blob = slice::from_raw_parts(blob as *const u8, blen);
+        callbacks.log_data(blob);
+    }
+}
+
 impl<const TRANSACTION: bool> WriteBatchWithTransaction<TRANSACTION> {
     /// Create a new `WriteBatch` without allocating memory.
     pub fn new() -> Self {
@@ -215,39 +247,41 @@ impl<const TRANSACTION: bool> WriteBatchWithTransaction<TRANSACTION> {
         self.len() == 0
     }
 
-    /// Iterate the put and delete operations within this write batch. Note that
-    /// this does _not_ return an `Iterator` but instead will invoke the `put()`
-    /// and `delete()` member functions of the provided `WriteBatchIterator`
-    /// trait implementation.
+    /// Iterate the put, delete, and log-data operations within this write batch.
+    /// Note that this does _not_ return an `Iterator` but instead will invoke
+    /// the `put()`, `delete()`, and `log_data()` member functions of the
+    /// provided `WriteBatchIterator` trait implementation.
     pub fn iterate<T: WriteBatchIterator>(&self, callbacks: &mut T) {
         let state = std::ptr::from_mut::<T>(callbacks) as *mut c_void;
         unsafe {
-            ffi::rocksdb_writebatch_iterate(
+            ffi::rust_rocksdb_writebatch_iterate(
                 self.inner,
                 state,
                 Some(writebatch_put_callback::<T>),
                 Some(writebatch_delete_callback::<T>),
+                Some(writebatch_log_data_callback::<T>),
             );
         }
     }
 
-    /// Iterate the put, delete, and merge operations within this write batch with column family
-    /// information. Note that this does _not_ return an `Iterator` but instead will invoke the
-    /// `put_cf()`, `delete_cf()`, and `merge_cf()` member functions of the provided
-    /// `WriteBatchIteratorCf` trait implementation.
+    /// Iterate the put, delete, merge, and log-data operations within this write batch with
+    /// column family information. Note that this does _not_ return an `Iterator` but instead
+    /// will invoke the `put_cf()`, `delete_cf()`, `merge_cf()`, and `log_data()` member
+    /// functions of the provided `WriteBatchIteratorCf` trait implementation.
     ///
     /// # Notes
     /// - For operations on the default column family ("default"), the `cf_id` parameter passed to
     ///   the callbacks will be 0
     pub fn iterate_cf<T: WriteBatchIteratorCf>(&self, callbacks: &mut T) {
         let state = std::ptr::from_mut::<T>(callbacks) as *mut c_void;
         unsafe {
-            ffi::rocksdb_writebatch_iterate_cf(
+            ffi::rust_rocksdb_writebatch_iterate_cf(
                 self.inner,
                 state,
                 Some(writebatch_put_cf_callback::<T>),
                 Some(writebatch_delete_cf_callback::<T>),
                 Some(writebatch_merge_cf_callback::<T>),
+                Some(writebatch_log_data_cf_callback::<T>),
             );
         }
     }