feat(storage): use roaring bitmaps for history tables (#5463)

bin/reth/src/commands/stage/run.rs

@@ -18,14 +18,14 @@ use reth_config::Config;
 use reth_db::init_db;
 use reth_downloaders::bodies::bodies::BodiesDownloaderBuilder;
 use reth_primitives::ChainSpec;
-use reth_provider::{ProviderFactory, StageCheckpointReader};
+use reth_provider::{ProviderFactory, StageCheckpointReader, StageCheckpointWriter};
 use reth_stages::{
     stages::{
         AccountHashingStage, BodyStage, ExecutionStage, ExecutionStageThresholds,
         IndexAccountHistoryStage, IndexStorageHistoryStage, MerkleStage, SenderRecoveryStage,
         StorageHashingStage, TransactionLookupStage,
     },
-    ExecInput, Stage, StageExt, UnwindInput,
+    ExecInput, ExecOutput, Stage, StageExt, UnwindInput, UnwindOutput,
 };
 use std::{any::Any, net::SocketAddr, path::PathBuf, sync::Arc};
 use tracing::*;
@@ -102,6 +102,10 @@ pub struct Command {
     // e.g. query the DB size, or any table data.
     #[arg(long, short)]
     commit: bool,
+
+    /// Save stage checkpoints
+    #[arg(long)]
+    checkpoints: bool,
 }
 
 impl Command {
@@ -244,8 +248,12 @@ impl Command {
 
         if !self.skip_unwind {
             while unwind.checkpoint.block_number > self.from {
-                let unwind_output = unwind_stage.unwind(&provider_rw, unwind)?;
-                unwind.checkpoint = unwind_output.checkpoint;
+                let UnwindOutput { checkpoint } = unwind_stage.unwind(&provider_rw, unwind)?;
+                unwind.checkpoint = checkpoint;
+
+                if self.checkpoints {
+                    provider_rw.save_stage_checkpoint(unwind_stage.id(), checkpoint)?;
+                }
 
                 if self.commit {
                     provider_rw.commit()?;
@@ -261,16 +269,19 @@ impl Command {
 
         loop {
             exec_stage.execute_ready(input).await?;
-            let output = exec_stage.execute(&provider_rw, input)?;
+            let ExecOutput { checkpoint, done } = exec_stage.execute(&provider_rw, input)?;
 
-            input.checkpoint = Some(output.checkpoint);
+            input.checkpoint = Some(checkpoint);
 
+            if self.checkpoints {
+                provider_rw.save_stage_checkpoint(exec_stage.id(), checkpoint)?;
+            }
             if self.commit {
                 provider_rw.commit()?;
                 provider_rw = factory.provider_rw()?;
             }
 
-            if output.done {
+            if done {
                 break
             }
         }

crates/interfaces/src/test_utils/generators.rs

@@ -324,12 +324,9 @@ pub fn random_eoa_account<R: Rng>(rng: &mut R) -> (Address, Account) {
 }
 
 /// Generate random Externally Owned Accounts
-pub fn random_eoa_account_range<R: Rng>(
-    rng: &mut R,
-    acc_range: Range<u64>,
-) -> Vec<(Address, Account)> {
-    let mut accounts = Vec::with_capacity(acc_range.end.saturating_sub(acc_range.start) as usize);
-    for _ in acc_range {
+pub fn random_eoa_accounts<R: Rng>(rng: &mut R, accounts_num: usize) -> Vec<(Address, Account)> {
+    let mut accounts = Vec::with_capacity(accounts_num);
+    for _ in 0..accounts_num {
         accounts.push(random_eoa_account(rng))
     }
     accounts

crates/primitives/Cargo.toml

@@ -49,11 +49,11 @@ rayon.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 sha2 = "0.10.7"
-sucds = "~0.6"
 tempfile.workspace = true
 thiserror.workspace = true
 zstd = { version = "0.12", features = ["experimental"] }
 ahash.workspace = true
+roaring = "0.10.2"
 
 # `test-utils` feature
 hash-db = { version = "~0.15", optional = true }
@@ -82,6 +82,9 @@ triehash = "0.8"
 hash-db = "~0.15"
 plain_hasher = "0.2"
 
+sucds = "0.8.1"
+anyhow = "1.0.75"
+
 # necessary so we don't hit a "undeclared 'std'":
 # https://github.com/paradigmxyz/reth/pull/177#discussion_r1021172198
 criterion.workspace = true
@@ -122,3 +125,7 @@ harness = false
 [[bench]]
 name = "nibbles"
 harness = false
+
+[[bench]]
+name = "integer_list"
+harness = false