feat(storage): custom tables

bin/reth/src/commands/db/diff.rs

@@ -171,7 +171,7 @@ where
     T::Key: Hash,
     T::Value: PartialEq,
 {
-    let table = T::TABLE;
+    let table = T::NAME;
 
     info!("Analyzing table {table}...");
     let result = find_diffs_advanced::<T>(&primary_tx, &secondary_tx)?;

bin/reth/src/commands/db/stats.rs

@@ -78,7 +78,7 @@ impl Command {
         ]);
 
         tool.provider_factory.db_ref().view(|tx| {
-            let mut db_tables = Tables::ALL.iter().map(|table| table.name()).collect::<Vec<_>>();
+            let mut db_tables = Tables::ALL.iter().map(|table| table.name()).collect::<Vec<_>>(); // TODO: custom tables
             db_tables.sort();
             let mut total_size = 0;
             for db_table in db_tables {
@@ -318,7 +318,7 @@ impl Command {
         table.load_preset(comfy_table::presets::ASCII_MARKDOWN);
         table.set_header(vec![Cell::new("Table"), Cell::new("Checksum"), Cell::new("Elapsed")]);
 
-        let db_tables = Tables::ALL;
+        let db_tables = Tables::ALL; // TODO: custom tables
         let mut total_elapsed = Duration::default();
 
         for db_table in db_tables {

bin/reth/src/commands/node/mod.rs

@@ -10,7 +10,7 @@ use crate::{
 };
 use clap::{value_parser, Args, Parser};
 use reth_cli_runner::CliContext;
-use reth_db::{init_db, DatabaseEnv};
+use reth_db::{init_db, DatabaseEnv, Tables};
 use reth_node_builder::{NodeBuilder, WithLaunchContext};
 use reth_node_core::{node_config::NodeConfig, version};
 use reth_primitives::ChainSpec;
@@ -183,7 +183,9 @@ impl<Ext: clap::Args + fmt::Debug> NodeCommand<Ext> {
         let db_path = data_dir.db_path();
 
         tracing::info!(target: "reth::cli", path = ?db_path, "Opening database");
-        let database = Arc::new(init_db(db_path.clone(), self.db.database_args())?.with_metrics());
+        let tables = Tables::ALL.iter().map(|table| table.name()).collect::<Vec<_>>();
+        let database =
+            Arc::new(init_db(db_path.clone(), self.db.database_args())?.with_metrics(tables));
 
         if with_unused_ports {
             node_config = node_config.with_unused_ports();

crates/storage/db/src/abstraction/table.rs

@@ -80,11 +80,14 @@ impl<T> Value for T where T: Compress + Decompress + Serialize {}
 /// It allows for the use of codecs. See [`crate::models::ShardedKey`] for a custom
 /// implementation.
 pub trait Table: Send + Sync + Debug + 'static {
-    /// The dynamic type of the table.
-    const TABLE: crate::Tables;
-
     /// The table's name.
-    const NAME: &'static str = Self::TABLE.name();
+    const NAME: &'static str;
+
+    /// Index of the table.
+    const INDEX: usize;
+
+    /// If the table allows for duplicate keys.
+    const DUPSORT: bool;
 
     /// Key element of `Table`.
     ///

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

@@ -52,7 +52,7 @@ impl<K: TransactionKind, T: Table> Cursor<K, T> {
         f: impl FnOnce(&mut Self) -> R,
     ) -> R {
         if let Some(metrics) = self.metrics.as_ref().cloned() {
-            metrics.record_operation(T::TABLE, operation, value_size, || f(self))
+            metrics.record_operation(T::NAME, operation, value_size, || f(self))
         } else {
             f(self)
         }

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

@@ -41,6 +41,9 @@ const DEFAULT_MAX_READERS: u64 = 32_000;
 #[cfg(not(windows))]
 const MAX_SAFE_READER_SPACE: usize = 10 * GIGABYTE;
 
+/// Maximum number of allowed databases/tables.
+const MAX_DBS: usize = 256;
+
 /// Environment used when opening a MDBX environment. RO/RW.
 #[derive(Debug)]
 pub enum DatabaseEnvKind {
@@ -58,7 +61,7 @@ impl DatabaseEnvKind {
 }
 
 /// Arguments for database initialization.
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Debug)]
 pub struct DatabaseArguments {
     /// Client version that accesses the database.
     client_version: ClientVersion,
@@ -87,6 +90,26 @@ pub struct DatabaseArguments {
     ///
     /// This flag affects only at environment opening but can't be changed after.
     exclusive: Option<bool>,
+    /// Creates default tables.
+    ///
+    /// If 'true', the default tables are created when the database is opened. If [None], the
+    /// default value is used.
+    ///
+    /// If 'false', the default tables are not created when the database is opened.
+    /// This is useful when using the database only for custom tables.
+    default_tables: Option<bool>,
+}
+
+impl Default for DatabaseArguments {
+    fn default() -> Self {
+        Self {
+            client_version: ClientVersion::default(),
+            log_level: None,
+            max_read_transaction_duration: None,
+            exclusive: None,
+            default_tables: Some(true),
+        }
+    }
 }
 
 impl DatabaseArguments {
@@ -97,6 +120,7 @@ impl DatabaseArguments {
             log_level: None,
             max_read_transaction_duration: None,
             exclusive: None,
+            default_tables: Some(true),
         }
     }
 
@@ -121,10 +145,21 @@ impl DatabaseArguments {
         self
     }
 
+    /// Set the default tables flag.
+    pub fn with_default_tables(mut self, default_tables: Option<bool>) -> Self {
+        self.default_tables = default_tables;
+        self
+    }
+
     /// Returns the client version if any.
     pub fn client_version(&self) -> &ClientVersion {
         &self.client_version
     }
+
+    /// Returns the default tables flag if any.
+    pub fn default_tables(&self) -> Option<bool> {
+        self.default_tables
+    }
 }
 
 /// Wrapper for the libmdbx environment: [Environment]
@@ -169,6 +204,7 @@ impl DatabaseMetrics for DatabaseEnv {
 
         let _ = self
             .view(|tx| {
+                // TODO: custom tables
                 for table in Tables::ALL.iter().map(Tables::name) {
                     let table_db = tx.inner.open_db(Some(table)).wrap_err("Could not open db.")?;
 
@@ -245,7 +281,7 @@ impl DatabaseMetadata for DatabaseEnv {
 impl DatabaseEnv {
     /// Opens the database at the specified path with the given `EnvKind`.
     ///
-    /// It does not create the tables, for that call [`DatabaseEnv::create_tables`].
+    /// It does not create the tables, for that call [`DatabaseEnv::create_default_tables`].
     pub fn open(
         path: &Path,
         kind: DatabaseEnvKind,
@@ -264,8 +300,7 @@ impl DatabaseEnv {
 
         // Note: We set max dbs to 256 here to allow for custom tables. This needs to be set on
         // environment creation.
-        debug_assert!(Tables::ALL.len() <= 256, "number of tables exceed max dbs");
-        inner_env.set_max_dbs(256);
+        inner_env.set_max_dbs(MAX_DBS);
         inner_env.set_geometry(Geometry {
             // Maximum database size of 4 terabytes
             size: Some(0..(4 * TERABYTE)),
@@ -371,7 +406,7 @@ impl DatabaseEnv {
                     LogLevel::Extra => 7,
                 });
             } else {
-                return Err(DatabaseError::LogLevelUnavailable(log_level))
+                return Err(DatabaseError::LogLevelUnavailable(log_level));
             }
         }
 
@@ -388,13 +423,24 @@ impl DatabaseEnv {
     }
 
     /// Enables metrics on the database.
-    pub fn with_metrics(mut self) -> Self {
-        self.metrics = Some(DatabaseEnvMetrics::new().into());
+    pub fn with_metrics(mut self, tables: Vec<&'static str>) -> Self {
+        self.metrics = Some(DatabaseEnvMetrics::new(tables).into());
         self
     }
 
-    /// Creates all the defined tables, if necessary.
-    pub fn create_tables(&self) -> Result<(), DatabaseError> {
+    /// Creates a table.
+    pub fn create_table(&self, name: &'static str, dupsort: bool) -> Result<(), DatabaseError> {
+        let tx = self.inner.begin_rw_txn().map_err(|e| DatabaseError::InitTx(e.into()))?;
+
+        let flags = if dupsort { DatabaseFlags::DUP_SORT } else { DatabaseFlags::default() };
+        tx.create_db(Some(name), flags).map_err(|e| DatabaseError::CreateTable(e.into()))?;
+        tx.commit().map_err(|e| DatabaseError::Commit(e.into()))?;
+
+        Ok(())
+    }
+
+    /// Creates all the defined default tables, if necessary.
+    pub fn create_default_tables(&self) -> Result<(), DatabaseError> {
         let tx = self.inner.begin_rw_txn().map_err(|e| DatabaseError::InitTx(e.into()))?;
 
         for table in Tables::ALL {
@@ -415,7 +461,7 @@ impl DatabaseEnv {
     /// Records version that accesses the database with write privileges.
     pub fn record_client_version(&self, version: ClientVersion) -> Result<(), DatabaseError> {
         if version.is_empty() {
-            return Ok(())
+            return Ok(());
         }
 
         let tx = self.tx_mut()?;
@@ -473,7 +519,7 @@ mod tests {
     fn create_test_db_with_path(kind: DatabaseEnvKind, path: &Path) -> DatabaseEnv {
         let env = DatabaseEnv::open(path, kind, DatabaseArguments::new(ClientVersion::default()))
             .expect(ERROR_DB_CREATION);
-        env.create_tables().expect(ERROR_TABLE_CREATION);
+        env.create_default_tables().expect(ERROR_TABLE_CREATION);
         env
     }
 

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

@@ -1,10 +1,10 @@
 //! Transaction wrapper for libmdbx-sys.
 
-use super::cursor::Cursor;
+use super::{cursor::Cursor, MAX_DBS};
 use crate::{
     metrics::{DatabaseEnvMetrics, Operation, TransactionMode, TransactionOutcome},
     table::{Compress, DupSort, Encode, Table, TableImporter},
-    tables::{utils::decode_one, Tables},
+    tables::utils::decode_one,
     transaction::{DbTx, DbTxMut},
     DatabaseError,
 };
@@ -39,7 +39,7 @@ pub struct Tx<K: TransactionKind> {
 
     /// Database table handle cache.
     // TODO: Use `std::sync::OnceLock` once `get_or_try_init` is stable.
-    db_handles: [OnceCell<DBI>; Tables::COUNT],
+    db_handles: [OnceCell<DBI>; MAX_DBS],
 }
 
 impl<K: TransactionKind> Tx<K> {
@@ -75,7 +75,7 @@ impl<K: TransactionKind> Tx<K> {
         #[allow(clippy::declare_interior_mutable_const)]
         const ONCECELL_DBI_NEW: OnceCell<DBI> = OnceCell::new();
         #[allow(clippy::declare_interior_mutable_const)]
-        const DB_HANDLES: [OnceCell<DBI>; Tables::COUNT] = [ONCECELL_DBI_NEW; Tables::COUNT];
+        const DB_HANDLES: [OnceCell<DBI>; MAX_DBS] = [ONCECELL_DBI_NEW; MAX_DBS];
         Self { inner, db_handles: DB_HANDLES, metrics_handler }
     }
 
@@ -86,7 +86,7 @@ impl<K: TransactionKind> Tx<K> {
 
     /// Gets a table database handle if it exists, otherwise creates it.
     pub fn get_dbi<T: Table>(&self) -> Result<DBI, DatabaseError> {
-        self.db_handles[T::TABLE as usize]
+        self.db_handles[T::INDEX]
             .get_or_try_init(|| {
                 self.inner
                     .open_db(Some(T::NAME))
@@ -170,7 +170,7 @@ impl<K: TransactionKind> Tx<K> {
             metrics_handler.log_backtrace_on_long_read_transaction();
             metrics_handler
                 .env_metrics
-                .record_operation(T::TABLE, operation, value_size, || f(&self.inner))
+                .record_operation(T::NAME, operation, value_size, || f(&self.inner))
         } else {
             f(&self.inner)
         }
@@ -393,7 +393,7 @@ impl DbTxMut for Tx<RW> {
 mod tests {
     use crate::{
         database::Database, mdbx::DatabaseArguments, models::client_version::ClientVersion, tables,
-        transaction::DbTx, DatabaseEnv, DatabaseEnvKind,
+        transaction::DbTx, DatabaseEnv, DatabaseEnvKind, Tables,
     };
     use reth_interfaces::db::DatabaseError;
     use reth_libmdbx::MaxReadTransactionDuration;
@@ -409,7 +409,9 @@ mod tests {
             .with_max_read_transaction_duration(Some(MaxReadTransactionDuration::Set(
                 MAX_DURATION,
             )));
-        let db = DatabaseEnv::open(dir.path(), DatabaseEnvKind::RW, args).unwrap().with_metrics();
+        let tables = Tables::ALL.iter().map(|table| table.name()).collect::<Vec<_>>();
+        let db =
+            DatabaseEnv::open(dir.path(), DatabaseEnvKind::RW, args).unwrap().with_metrics(tables);
 
         let mut tx = db.tx().unwrap();
         tx.metrics_handler.as_mut().unwrap().long_transaction_duration = MAX_DURATION;
@@ -435,7 +437,9 @@ mod tests {
             .with_max_read_transaction_duration(Some(MaxReadTransactionDuration::Set(
                 MAX_DURATION,
             )));
-        let db = DatabaseEnv::open(dir.path(), DatabaseEnvKind::RW, args).unwrap().with_metrics();
+        let tables = Tables::ALL.iter().map(|table| table.name()).collect::<Vec<_>>();
+        let db =
+            DatabaseEnv::open(dir.path(), DatabaseEnvKind::RW, args).unwrap().with_metrics(tables);
 
         let mut tx = db.tx().unwrap();
         tx.metrics_handler.as_mut().unwrap().long_transaction_duration = MAX_DURATION;

crates/storage/db/src/lib.rs

@@ -126,8 +126,11 @@ pub fn init_db<P: AsRef<Path>>(path: P, args: DatabaseArguments) -> eyre::Result
     #[cfg(feature = "mdbx")]
     {
         let client_version = args.client_version().clone();
+        let default_tables = args.default_tables().unwrap_or(true);
         let db = create_db(path, args)?;
-        db.create_tables()?;
+        if default_tables {
+            db.create_default_tables()?;
+        }
         db.record_client_version(client_version)?;
         Ok(db)
     }

crates/storage/db/src/metrics.rs

@@ -1,4 +1,3 @@
-use crate::Tables;
 use metrics::{Gauge, Histogram};
 use reth_libmdbx::CommitLatency;
 use reth_metrics::{metrics::Counter, Metrics};
@@ -20,7 +19,7 @@ const LARGE_VALUE_THRESHOLD_BYTES: usize = 4096;
 #[derive(Debug)]
 pub struct DatabaseEnvMetrics {
     /// Caches OperationMetrics handles for each table and operation tuple.
-    operations: FxHashMap<(Tables, Operation), OperationMetrics>,
+    operations: FxHashMap<(String, Operation), OperationMetrics>,
     /// Caches TransactionMetrics handles for counters grouped by only transaction mode.
     /// Updated both at tx open and close.
     transactions: FxHashMap<TransactionMode, TransactionMetrics>,
@@ -31,29 +30,31 @@ pub struct DatabaseEnvMetrics {
 }
 
 impl DatabaseEnvMetrics {
-    pub(crate) fn new() -> Self {
+    pub(crate) fn new(tables: Vec<&'static str>) -> Self {
         // Pre-populate metric handle maps with all possible combinations of labels
         // to avoid runtime locks on the map when recording metrics.
         Self {
-            operations: Self::generate_operation_handles(),
+            operations: Self::generate_operation_handles(tables),
             transactions: Self::generate_transaction_handles(),
             transaction_outcomes: Self::generate_transaction_outcome_handles(),
         }
     }
 
     /// Generate a map of all possible operation handles for each table and operation tuple.
     /// Used for tracking all operation metrics.
-    fn generate_operation_handles() -> FxHashMap<(Tables, Operation), OperationMetrics> {
+    fn generate_operation_handles(
+        tables: Vec<&'static str>,
+    ) -> FxHashMap<(String, Operation), OperationMetrics> {
         let mut operations = FxHashMap::with_capacity_and_hasher(
-            Tables::COUNT * Operation::COUNT,
+            tables.len() * Operation::COUNT,
             BuildHasherDefault::<FxHasher>::default(),
         );
-        for table in Tables::ALL {
+        for table in tables {
             for operation in Operation::iter() {
                 operations.insert(
-                    (*table, operation),
+                    (table.into(), operation),
                     OperationMetrics::new_with_labels(&[
-                        (Labels::Table.as_str(), table.name()),
+                        (Labels::Table.as_str(), table),
                         (Labels::Operation.as_str(), operation.as_str()),
                     ]),
                 );
@@ -104,13 +105,13 @@ impl DatabaseEnvMetrics {
     /// Panics if a metric recorder is not found for the given table and operation.
     pub(crate) fn record_operation<R>(
         &self,
-        table: Tables,
+        table: &'static str,
         operation: Operation,
         value_size: Option<usize>,
         f: impl FnOnce() -> R,
     ) -> R {
         self.operations
-            .get(&(table, operation))
+            .get(&(table.into(), operation))
             .expect("operation & table metric handle not found")
             .record(value_size, f)
     }