feat: move ZstdDictionary inside NippyJar and create a snapshot manager

bin/reth/src/db/snapshots/headers.rs

@@ -5,13 +5,14 @@ use super::{
 use rand::{seq::SliceRandom, Rng};
 use reth_db::{database::Database, open_db_read_only, table::Decompress};
 use reth_interfaces::db::LogLevel;
-use reth_nippy_jar::NippyJar;
 use reth_primitives::{
     snapshot::{Compression, Filters, InclusionFilter, PerfectHashingFunction},
     ChainSpec, Header, SnapshotSegment,
 };
-use reth_provider::{DatabaseProviderRO, HeaderProvider, ProviderError, ProviderFactory};
-use reth_snapshot::segments::{get_snapshot_segment_file_name, Headers, Segment};
+use reth_provider::{
+    providers::SnapshotProvider, DatabaseProviderRO, HeaderProvider, ProviderError, ProviderFactory,
+};
+use reth_snapshot::segments::{Headers, Segment};
 use std::{path::Path, sync::Arc};
 
 impl Command {
@@ -54,20 +55,12 @@ impl Command {
 
         let mut row_indexes = range.clone().collect::<Vec<_>>();
         let mut rng = rand::thread_rng();
-        let mut dictionaries = None;
-        let mut jar = NippyJar::load(&get_snapshot_segment_file_name(
-            SnapshotSegment::Headers,
-            filters,
-            compression,
-            &range,
-        ))?;
-
-        let (provider, decompressors) = self.prepare_jar_provider(&mut jar, &mut dictionaries)?;
-        let mut cursor = if !decompressors.is_empty() {
-            provider.cursor_with_decompressors(decompressors)
-        } else {
-            provider.cursor()
-        };
+        let path =
+            SnapshotSegment::Headers.filename_with_configuration(filters, compression, &range);
+        let provider = SnapshotProvider::default();
+        let jar_provider =
+            provider.get_segment_provider(SnapshotSegment::Headers, self.from, Some(path))?;
+        let mut cursor = jar_provider.cursor()?;
 
         for bench_kind in [BenchKind::Walk, BenchKind::RandomAll] {
             bench(

bin/reth/src/db/snapshots/mod.rs

@@ -2,15 +2,11 @@ use clap::Parser;
 use itertools::Itertools;
 use reth_db::{open_db_read_only, DatabaseEnvRO};
 use reth_interfaces::db::LogLevel;
-use reth_nippy_jar::{
-    compression::{DecoderDictionary, Decompressor},
-    NippyJar,
-};
 use reth_primitives::{
-    snapshot::{Compression, InclusionFilter, PerfectHashingFunction, SegmentHeader},
+    snapshot::{Compression, InclusionFilter, PerfectHashingFunction},
     BlockNumber, ChainSpec, SnapshotSegment,
 };
-use reth_provider::{providers::SnapshotProvider, ProviderFactory};
+use reth_provider::ProviderFactory;
 use std::{path::Path, sync::Arc};
 
 mod bench;
@@ -134,22 +130,4 @@ impl Command {
 
         Ok(())
     }
-
-    /// Returns a [`SnapshotProvider`] of the provided [`NippyJar`], alongside a list of
-    /// [`DecoderDictionary`] and [`Decompressor`] if necessary.
-    fn prepare_jar_provider<'a>(
-        &self,
-        jar: &'a mut NippyJar<SegmentHeader>,
-        dictionaries: &'a mut Option<Vec<DecoderDictionary<'_>>>,
-    ) -> eyre::Result<(SnapshotProvider<'a>, Vec<Decompressor<'a>>)> {
-        let mut decompressors: Vec<Decompressor<'_>> = vec![];
-        if let Some(reth_nippy_jar::compression::Compressors::Zstd(zstd)) = jar.compressor_mut() {
-            if zstd.use_dict {
-                *dictionaries = zstd.generate_decompress_dictionaries();
-                decompressors = zstd.generate_decompressors(dictionaries.as_ref().expect("qed"))?;
-            }
-        }
-
-        Ok((SnapshotProvider { jar: &*jar }, decompressors))
-    }
 }

bin/reth/src/db/snapshots/receipts.rs

@@ -5,19 +5,15 @@ use super::{
 use rand::{seq::SliceRandom, Rng};
 use reth_db::{database::Database, open_db_read_only, table::Decompress};
 use reth_interfaces::db::LogLevel;
-use reth_nippy_jar::NippyJar;
 use reth_primitives::{
     snapshot::{Filters, InclusionFilter},
     ChainSpec, Receipt, SnapshotSegment,
 };
 use reth_provider::{
-    DatabaseProviderRO, ProviderError, ProviderFactory, ReceiptProvider, TransactionsProvider,
-    TransactionsProviderExt,
-};
-use reth_snapshot::{
-    segments,
-    segments::{get_snapshot_segment_file_name, Segment},
+    providers::SnapshotProvider, DatabaseProviderRO, ProviderError, ProviderFactory,
+    ReceiptProvider, TransactionsProvider, TransactionsProviderExt,
 };
+use reth_snapshot::{segments, segments::Segment};
 use std::{path::Path, sync::Arc};
 
 impl Command {
@@ -59,26 +55,22 @@ impl Command {
         let block_range = self.from..=(self.from + self.block_interval - 1);
 
         let mut rng = rand::thread_rng();
-        let mut dictionaries = None;
-        let mut jar = NippyJar::load(&get_snapshot_segment_file_name(
-            SnapshotSegment::Receipts,
-            filters,
-            compression,
-            &block_range,
-        ))?;
 
         let tx_range = ProviderFactory::new(open_db_read_only(db_path, log_level)?, chain.clone())
             .provider()?
-            .transaction_range_by_block_range(block_range)?;
+            .transaction_range_by_block_range(block_range.clone())?;
 
         let mut row_indexes = tx_range.clone().collect::<Vec<_>>();
 
-        let (provider, decompressors) = self.prepare_jar_provider(&mut jar, &mut dictionaries)?;
-        let mut cursor = if !decompressors.is_empty() {
-            provider.cursor_with_decompressors(decompressors)
-        } else {
-            provider.cursor()
-        };
+        let path = SnapshotSegment::Receipts.filename_with_configuration(
+            filters,
+            compression,
+            &block_range,
+        );
+        let provider = SnapshotProvider::default();
+        let jar_provider =
+            provider.get_segment_provider(SnapshotSegment::Receipts, self.from, Some(path))?;
+        let mut cursor = jar_provider.cursor()?;
 
         for bench_kind in [BenchKind::Walk, BenchKind::RandomAll] {
             bench(

bin/reth/src/db/snapshots/transactions.rs

@@ -5,19 +5,15 @@ use super::{
 use rand::{seq::SliceRandom, Rng};
 use reth_db::{database::Database, open_db_read_only, table::Decompress};
 use reth_interfaces::db::LogLevel;
-use reth_nippy_jar::NippyJar;
 use reth_primitives::{
     snapshot::{Filters, InclusionFilter},
     ChainSpec, SnapshotSegment, TransactionSignedNoHash,
 };
 use reth_provider::{
-    DatabaseProviderRO, ProviderError, ProviderFactory, TransactionsProvider,
-    TransactionsProviderExt,
-};
-use reth_snapshot::{
-    segments,
-    segments::{get_snapshot_segment_file_name, Segment},
+    providers::SnapshotProvider, DatabaseProviderRO, ProviderError, ProviderFactory,
+    TransactionsProvider, TransactionsProviderExt,
 };
+use reth_snapshot::{segments, segments::Segment};
 use std::{path::Path, sync::Arc};
 
 impl Command {
@@ -59,26 +55,22 @@ impl Command {
         let block_range = self.from..=(self.from + self.block_interval - 1);
 
         let mut rng = rand::thread_rng();
-        let mut dictionaries = None;
-        let mut jar = NippyJar::load(&get_snapshot_segment_file_name(
-            SnapshotSegment::Transactions,
-            filters,
-            compression,
-            &block_range,
-        ))?;
 
         let tx_range = ProviderFactory::new(open_db_read_only(db_path, log_level)?, chain.clone())
             .provider()?
-            .transaction_range_by_block_range(block_range)?;
+            .transaction_range_by_block_range(block_range.clone())?;
 
         let mut row_indexes = tx_range.clone().collect::<Vec<_>>();
 
-        let (provider, decompressors) = self.prepare_jar_provider(&mut jar, &mut dictionaries)?;
-        let mut cursor = if !decompressors.is_empty() {
-            provider.cursor_with_decompressors(decompressors)
-        } else {
-            provider.cursor()
-        };
+        let path = SnapshotSegment::Transactions.filename_with_configuration(
+            filters,
+            compression,
+            &block_range,
+        );
+        let provider = SnapshotProvider::default();
+        let jar_provider =
+            provider.get_segment_provider(SnapshotSegment::Transactions, self.from, Some(path))?;
+        let mut cursor = jar_provider.cursor()?;
 
         for bench_kind in [BenchKind::Walk, BenchKind::RandomAll] {
             bench(

crates/primitives/src/snapshot/mod.rs

@@ -7,3 +7,6 @@ mod segment;
 pub use compression::Compression;
 pub use filters::{Filters, InclusionFilter, PerfectHashingFunction};
 pub use segment::{SegmentHeader, SnapshotSegment};
+
+/// Default snapshot block count.
+pub const BLOCKS_PER_SNAPSHOT: u64 = 500_000;

crates/primitives/src/snapshot/segment.rs

@@ -1,6 +1,8 @@
-use crate::{BlockNumber, TxNumber};
+use crate::{snapshot::PerfectHashingFunction, BlockNumber, TxNumber};
 use serde::{Deserialize, Serialize};
-use std::ops::RangeInclusive;
+use std::{ops::RangeInclusive, path::PathBuf};
+
+use super::{Compression, Filters, InclusionFilter};
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, Serialize)]
 #[cfg_attr(feature = "clap", derive(clap::ValueEnum))]
@@ -14,8 +16,70 @@ pub enum SnapshotSegment {
     Receipts,
 }
 
+impl SnapshotSegment {
+    /// Returns the default configuration of the segment.
+    const fn config(&self) -> (Filters, Compression) {
+        let default_config = (
+            Filters::WithFilters(InclusionFilter::Cuckoo, super::PerfectHashingFunction::Fmph),
+            Compression::Lz4,
+        );
+
+        match self {
+            SnapshotSegment::Headers => default_config,
+            SnapshotSegment::Transactions => default_config,
+            SnapshotSegment::Receipts => default_config,
+        }
+    }
+
+    /// Returns the default file name for the provided segment and range.
+    pub fn filename(&self, range: &RangeInclusive<BlockNumber>) -> PathBuf {
+        let (filters, compression) = self.config();
+        self.filename_with_configuration(filters, compression, range)
+    }
+
+    /// Returns file name for the provided segment, filters, compression and range.
+    pub fn filename_with_configuration(
+        &self,
+        filters: Filters,
+        compression: Compression,
+        range: &RangeInclusive<BlockNumber>,
+    ) -> PathBuf {
+        let segment_name = match self {
+            SnapshotSegment::Headers => "headers",
+            SnapshotSegment::Transactions => "transactions",
+            SnapshotSegment::Receipts => "receipts",
+        };
+        let filters_name = match filters {
+            Filters::WithFilters(inclusion_filter, phf) => {
+                let inclusion_filter = match inclusion_filter {
+                    InclusionFilter::Cuckoo => "cuckoo",
+                };
+                let phf = match phf {
+                    PerfectHashingFunction::Fmph => "fmph",
+                    PerfectHashingFunction::GoFmph => "gofmph",
+                };
+                format!("{inclusion_filter}-{phf}")
+            }
+            Filters::WithoutFilters => "none".to_string(),
+        };
+        let compression_name = match compression {
+            Compression::Lz4 => "lz4",
+            Compression::Zstd => "zstd",
+            Compression::ZstdWithDictionary => "zstd-dict",
+            Compression::Uncompressed => "uncompressed",
+        };
+
+        format!(
+            "snapshot_{segment_name}_{}_{}_{filters_name}_{compression_name}",
+            range.start(),
+            range.end(),
+        )
+        .into()
+    }
+}
+
 /// A segment header that contains information common to all segments. Used for storage.
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Eq, PartialEq, Hash)]
 pub struct SegmentHeader {
     block_range: RangeInclusive<BlockNumber>,
     tx_range: RangeInclusive<TxNumber>,

crates/snapshot/src/segments/mod.rs

@@ -19,7 +19,7 @@ use reth_primitives::{
     BlockNumber, SnapshotSegment,
 };
 use reth_provider::{DatabaseProviderRO, TransactionsProviderExt};
-use std::{ops::RangeInclusive, path::PathBuf};
+use std::ops::RangeInclusive;
 
 pub(crate) type Rows<const COLUMNS: usize> = [Vec<Vec<u8>>; COLUMNS];
 
@@ -61,7 +61,7 @@ pub(crate) fn prepare_jar<DB: Database, const COLUMNS: usize>(
     let tx_range = provider.transaction_range_by_block_range(block_range.clone())?;
     let mut nippy_jar = NippyJar::new(
         COLUMNS,
-        &get_snapshot_segment_file_name(segment, filters, compression, &block_range),
+        &segment.filename_with_configuration(filters, compression, &block_range),
         SegmentHeader::new(block_range, tx_range),
     );
 
@@ -90,43 +90,3 @@ pub(crate) fn prepare_jar<DB: Database, const COLUMNS: usize>(
 
     Ok(nippy_jar)
 }
-
-/// Returns file name for the provided segment, filters, compression and range.
-pub fn get_snapshot_segment_file_name(
-    segment: SnapshotSegment,
-    filters: Filters,
-    compression: Compression,
-    range: &RangeInclusive<BlockNumber>,
-) -> PathBuf {
-    let segment_name = match segment {
-        SnapshotSegment::Headers => "headers",
-        SnapshotSegment::Transactions => "transactions",
-        SnapshotSegment::Receipts => "receipts",
-    };
-    let filters_name = match filters {
-        Filters::WithFilters(inclusion_filter, phf) => {
-            let inclusion_filter = match inclusion_filter {
-                InclusionFilter::Cuckoo => "cuckoo",
-            };
-            let phf = match phf {
-                PerfectHashingFunction::Fmph => "fmph",
-                PerfectHashingFunction::GoFmph => "gofmph",
-            };
-            format!("{inclusion_filter}-{phf}")
-        }
-        Filters::WithoutFilters => "none".to_string(),
-    };
-    let compression_name = match compression {
-        Compression::Lz4 => "lz4",
-        Compression::Zstd => "zstd",
-        Compression::ZstdWithDictionary => "zstd-dict",
-        Compression::Uncompressed => "uncompressed",
-    };
-
-    format!(
-        "snapshot_{segment_name}_{}_{}_{filters_name}_{compression_name}",
-        range.start(),
-        range.end(),
-    )
-    .into()
-}

crates/storage/nippy-jar/Cargo.toml

@@ -34,6 +34,7 @@ tracing-appender = "0.2"
 anyhow = "1.0"
 thiserror.workspace = true
 hex = "*"
+derive_more = "0.99"
 
 [dev-dependencies]
 rand = { version = "0.8", features = ["small_rng"] }