feat(db): add MDBX put-append for fast ordered puts (#18603)

crates/storage/db-api/src/transaction.rs

@@ -50,6 +50,12 @@ pub trait DbTxMut: Send + Sync {
 
     /// Put value to database
     fn put<T: Table>(&self, key: T::Key, value: T::Value) -> Result<(), DatabaseError>;
+    /// Append value with the largest key to database. This should have the same
+    /// outcome as `put`, but databases like MDBX provide dedicated modes to make
+    /// it much faster, typically from O(logN) down to O(1) thanks to no lookup.
+    fn append<T: Table>(&self, key: T::Key, value: T::Value) -> Result<(), DatabaseError> {
+        self.put::<T>(key, value)
+    }
     /// Delete value from database
     fn delete<T: Table>(&self, key: T::Key, value: Option<T::Value>)
         -> Result<bool, DatabaseError>;

crates/storage/db/Cargo.toml

@@ -112,3 +112,8 @@ harness = false
 name = "get"
 required-features = ["test-utils"]
 harness = false
+
+[[bench]]
+name = "put"
+required-features = ["test-utils"]
+harness = false

crates/storage/db/benches/put.rs

@@ -0,0 +1,44 @@
+#![allow(missing_docs)]
+
+use alloy_primitives::B256;
+use criterion::{criterion_group, criterion_main, Criterion};
+use reth_db::{test_utils::create_test_rw_db_with_path, CanonicalHeaders, Database};
+use reth_db_api::transaction::DbTxMut;
+
+mod utils;
+use utils::BENCH_DB_PATH;
+
+const NUM_BLOCKS: u64 = 1_000_000;
+
+criterion_group! {
+    name = benches;
+    config = Criterion::default();
+    targets = put
+}
+criterion_main!(benches);
+
+// Small benchmark showing that `append` is much faster than `put` when keys are put in order
+fn put(c: &mut Criterion) {
+    let mut group = c.benchmark_group("Put");
+
+    let setup = || {
+        let _ = std::fs::remove_dir_all(BENCH_DB_PATH);
+        create_test_rw_db_with_path(BENCH_DB_PATH).tx_mut().expect("tx")
+    };
+
+    group.bench_function("put", |b| {
+        b.iter_with_setup(setup, |tx| {
+            for i in 0..NUM_BLOCKS {
+                tx.put::<CanonicalHeaders>(i, B256::ZERO).unwrap();
+            }
+        })
+    });
+
+    group.bench_function("append", |b| {
+        b.iter_with_setup(setup, |tx| {
+            for i in 0..NUM_BLOCKS {
+                tx.append::<CanonicalHeaders>(i, B256::ZERO).unwrap();
+            }
+        })
+    });
+}

crates/storage/db/src/implementation/mdbx/tx.rs

@@ -340,28 +340,64 @@ impl<K: TransactionKind> DbTx for Tx<K> {
     }
 }
 
+#[derive(Clone, Copy)]
+enum PutKind {
+    /// Default kind that inserts a new key-value or overwrites an existed key.
+    Upsert,
+    /// Append the key-value to the end of the table -- fast path when the new
+    /// key is the highest so far, like the latest block number.
+    Append,
+}
+
+impl PutKind {
+    const fn into_operation_and_flags(self) -> (Operation, DatabaseWriteOperation, WriteFlags) {
+        match self {
+            Self::Upsert => {
+                (Operation::PutUpsert, DatabaseWriteOperation::PutUpsert, WriteFlags::UPSERT)
+            }
+            Self::Append => {
+                (Operation::PutAppend, DatabaseWriteOperation::PutAppend, WriteFlags::APPEND)
+            }
+        }
+    }
+}
+
+impl Tx<RW> {
+    /// The inner implementation mapping to `mdbx_put` that supports different
+    /// put kinds like upserting and appending.
+    fn put<T: Table>(
+        &self,
+        kind: PutKind,
+        key: T::Key,
+        value: T::Value,
+    ) -> Result<(), DatabaseError> {
+        let key = key.encode();
+        let value = value.compress();
+        let (operation, write_operation, flags) = kind.into_operation_and_flags();
+        self.execute_with_operation_metric::<T, _>(operation, Some(value.as_ref().len()), |tx| {
+            tx.put(self.get_dbi::<T>()?, key.as_ref(), value, flags).map_err(|e| {
+                DatabaseWriteError {
+                    info: e.into(),
+                    operation: write_operation,
+                    table_name: T::NAME,
+                    key: key.into(),
+                }
+                .into()
+            })
+        })
+    }
+}
+
 impl DbTxMut for Tx<RW> {
     type CursorMut<T: Table> = Cursor<RW, T>;
     type DupCursorMut<T: DupSort> = Cursor<RW, T>;
 
     fn put<T: Table>(&self, key: T::Key, value: T::Value) -> Result<(), DatabaseError> {
-        let key = key.encode();
-        let value = value.compress();
-        self.execute_with_operation_metric::<T, _>(
-            Operation::Put,
-            Some(value.as_ref().len()),
-            |tx| {
-                tx.put(self.get_dbi::<T>()?, key.as_ref(), value, WriteFlags::UPSERT).map_err(|e| {
-                    DatabaseWriteError {
-                        info: e.into(),
-                        operation: DatabaseWriteOperation::Put,
-                        table_name: T::NAME,
-                        key: key.into(),
-                    }
-                    .into()
-                })
-            },
-        )
+        self.put::<T>(PutKind::Upsert, key, value)
+    }
+
+    fn append<T: Table>(&self, key: T::Key, value: T::Value) -> Result<(), DatabaseError> {
+        self.put::<T>(PutKind::Append, key, value)
     }
 
     fn delete<T: Table>(

crates/storage/db/src/metrics.rs

@@ -197,8 +197,10 @@ impl TransactionOutcome {
 pub(crate) enum Operation {
     /// Database get operation.
     Get,
-    /// Database put operation.
-    Put,
+    /// Database put upsert operation.
+    PutUpsert,
+    /// Database put append operation.
+    PutAppend,
     /// Database delete operation.
     Delete,
     /// Database cursor upsert operation.
@@ -220,7 +222,8 @@ impl Operation {
     pub(crate) const fn as_str(&self) -> &'static str {
         match self {
             Self::Get => "get",
-            Self::Put => "put",
+            Self::PutUpsert => "put-upsert",
+            Self::PutAppend => "put-append",
             Self::Delete => "delete",
             Self::CursorUpsert => "cursor-upsert",
             Self::CursorInsert => "cursor-insert",

crates/storage/errors/src/db.rs

@@ -106,8 +106,10 @@ pub enum DatabaseWriteOperation {
     CursorInsert,
     /// Append duplicate cursor.
     CursorAppendDup,
-    /// Put.
-    Put,
+    /// Put upsert.
+    PutUpsert,
+    /// Put append.
+    PutAppend,
 }
 
 /// Database log level.