Reduce Turbopack cache size with per-family compression

@@ -10030,6 +10030,7 @@ dependencies = [  "turbo-tasks-malloc",  "xxhash-rust",  "zerocopy",+ "zstd", ]  [[package]]

@@ -358,6 +358,7 @@ tracing = "0.1.44" tracing-subscriber = "0.3.16" triomphe = { git = "https://github.com/sokra/triomphe", branch = "sokra/unstable" } xxhash-rust = { version = "0.8.12", features = ["xxh3"] }+zstd = "0.13.2" unsize = "1.1.0" unty = "0.0.4" url = "2.2.2"

@@ -38,6 +38,7 @@ smallvec = { workspace = true } thread_local = { workspace = true } tracing = { workspace = true } xxhash-rust = { workspace = true }+zstd = { workspace = true }  [dev-dependencies] criterion = { workspace = true }

@@ -56,6 +56,7 @@ A meta file can contain metadata about multiple SST files. The metadata is store - Header   - 4 bytes magic number (0xFE4ADA4A)   - 4 bytes key family+  - 1 byte compression algorithm, which must match the configuration used to open the database   - 4 bytes count of obsolete SST files   - foreach obsolete SST file     - 4 bytes sequence number of the obsolete SST file@@ -88,9 +89,11 @@ The SST file contains only data without any header.  #### Block Compression -Blocks can be stored compressed (LZ4) or uncompressed. The 4-byte header distinguishes them:+Blocks can be stored compressed or uncompressed. The compression algorithm is specified in the meta file. -- **Header > 0**: Block is LZ4 compressed. Header value is the uncompressed length.+The 4-byte header distinguishes compressed from uncompressed storage:++- **Header > 0**: Block is compressed with the family's configured algorithm. Header value is the uncompressed length. - **Header = 0**: Block is stored uncompressed. Actual length is derived from block offsets.  #### Block Checksum@@ -223,7 +226,7 @@ The plain value compressed with dynamic compression. Each blob file has an 8-byt  - 4 bytes: uncompressed length (u32 big-endian) - 4 bytes: CRC32 checksum of the compressed data (u32 big-endian)-- remaining bytes: LZ4-compressed value data+- remaining bytes: value data compressed with the blob's key-family configuration  The checksum is verified on the compressed data **before** decompression when the blob is read. 

@@ -10,9 +10,9 @@ use quick_cache::sync::GuardResult; use rand::{RngExt, SeedableRng, rngs::SmallRng, seq::SliceRandom}; use tempfile::TempDir; use turbo_persistence::{-    ArcBytes, BlockCache, CompactConfig, DbConfig as TpDbConfig, Entry, EntryValue, FamilyConfig,-    FamilyKind, MetaEntryFlags, SerialScheduler, StaticSortedFile, StaticSortedFileMetaData,-    TurboPersistence, hash_key, write_static_stored_file,+    ArcBytes, BlockCache, CompactConfig, Compression, DbConfig as TpDbConfig, Entry, EntryValue,+    FamilyConfig, FamilyKind, MetaEntryFlags, SerialScheduler, StaticSortedFile,+    StaticSortedFileMetaData, TurboPersistence, hash_key, write_static_stored_file, }; use turbo_tasks_malloc::TurboMalloc; @@ -622,6 +622,7 @@ fn prefill_multi_value_database(         family_configs: [FamilyConfig {             name: "test",             kind: FamilyKind::MultiValue,+            compression: Compression::Lz4,         }],     };     let db =@@ -696,6 +697,7 @@ fn open_multi_value_db(path: &Path) -> TurboPersistence<SerialScheduler, 1> {         family_configs: [FamilyConfig {             name: "test",             kind: FamilyKind::MultiValue,+            compression: Compression::Lz4,         }],     };     TurboPersistence::<SerialScheduler, 1>::open_with_config(path.to_path_buf(), db_config).unwrap()@@ -964,6 +966,7 @@ fn bench_write_multi_value(c: &mut Criterion) {                             family_configs: [FamilyConfig {                                 name: "test",                                 kind: FamilyKind::MultiValue,+                                compression: Compression::Lz4,                             }],                         };                         let db = TurboPersistence::<SerialScheduler, 1>::open_with_config(@@ -1201,15 +1204,20 @@ fn bench_static_sorted_file_lookup(c: &mut Criterion) {             // Create temp directory and write SST file             let tempdir = tempfile::tempdir().unwrap();             let sst_path = tempdir.path().join("00000001.sst");-            let (meta, _file) =-                write_static_stored_file(&entries, &sst_path, MetaEntryFlags::FRESH).unwrap();+            let (meta, _file) = write_static_stored_file(+                &entries,+                &sst_path,+                MetaEntryFlags::FRESH,+                Compression::Lz4,+            )+            .unwrap();              // Open the SST file             let sst_meta = StaticSortedFileMetaData {                 sequence_number: 1,                 block_count: meta.block_count,             };-            let sst = StaticSortedFile::open(tempdir.path(), sst_meta).unwrap();+            let sst = StaticSortedFile::open(tempdir.path(), sst_meta, Compression::Lz4).unwrap();              // Create block caches             let key_block_cache: BlockCache = BlockCache::with(

@@ -9,6 +9,7 @@ use std::{ use memmap2::Mmap;  use crate::{+    Compression,     compression::decompress_into_arc,     shared_bytes::{SharedBytes, is_subslice_of}, };@@ -144,8 +145,13 @@ impl SharedBytes for ArcBytes {         }     } -    fn from_decompressed(uncompressed_length: u32, block: &[u8]) -> anyhow::Result<Self> {+    fn from_decompressed(+        compression: Compression,+        uncompressed_length: u32,+        block: &[u8],+    ) -> anyhow::Result<Self> {         Ok(ArcBytes::from(decompress_into_arc(+            compression,             uncompressed_length,             block,         )?))

@@ -20,7 +20,7 @@ use fs_err::{self as fs, File}; use lzzzz::lz4::decompress; use memmap2::Mmap; use turbo_persistence::{-    BLOCK_HEADER_SIZE, MAX_INLINE_VALUE_SIZE, checksum_block,+    BLOCK_HEADER_SIZE, Compression, MAX_INLINE_VALUE_SIZE, checksum_block,     meta_file::MetaFile,     mmap_helper::advise_mmap_for_persistence,     read_current_version,@@ -133,6 +133,7 @@ impl SstStats { struct SstInfo {     sequence_number: u32,     block_count: u16,+    compression: Compression, }  /// Accumulates statistics for a single entry of the given type.@@ -266,7 +267,8 @@ fn collect_sst_info(db_path: &Path) -> Result<BTreeMap<u32, Vec<SstInfo>>> {     let mut meta_files: Vec<MetaFile> = meta_seqs         .iter()         .map(|&seq| {-            MetaFile::open(db_path, seq).with_context(|| format!("Failed to open {seq:08}.meta"))+            MetaFile::open(db_path, seq, None)+                .with_context(|| format!("Failed to open {seq:08}.meta"))         })         .collect::<Result<_>>()?; @@ -283,6 +285,7 @@ fn collect_sst_info(db_path: &Path) -> Result<BTreeMap<u32, Vec<SstInfo>>> {             family_sst_info.entry(family).or_default().push(SstInfo {                 sequence_number: entry.sequence_number(),                 block_count: entry.block_count(),+                compression: meta.compression(),             });         }     }@@ -304,6 +307,7 @@ fn read_block(     block_offsets_start: usize,     block_index: u16,     sequence_number: u32,+    compression: Compression, ) -> Result<RawBlock> {     let offset = block_offsets_start + block_index as usize * size_of::<u32>(); @@ -343,7 +347,13 @@ fn read_block(      let data = if was_compressed {         let mut buffer = vec![0u8; uncompressed_length as usize];-        let bytes_written = decompress(compressed_data, &mut buffer)?;+        let bytes_written = match compression {+            Compression::Lz4 => {+                decompress(compressed_data, &mut buffer).context("LZ4 decompression failed")?+            }+            Compression::Zstd3 => zstd::bulk::decompress_to_buffer(compressed_data, &mut buffer)+                .context("zstd decompression failed")?,+        };         assert_eq!(             bytes_written, uncompressed_length as usize,             "Decompressed length does not match expected"@@ -471,6 +481,7 @@ fn iter_key_block_entry_types(  /// Analyze an SST file and return entry type statistics fn analyze_sst_file(db_path: &Path, info: &SstInfo) -> Result<SstStats> {+    let compression = info.compression;     let filename = format!("{:08}.sst", info.sequence_number);     let path = db_path.join(&filename); @@ -496,6 +507,7 @@ fn analyze_sst_file(db_path: &Path, info: &SstInfo) -> Result<SstStats> {         block_offsets_start,         index_block_index,         info.sequence_number,+        compression,     )?;     let key_block_indices = parse_key_block_indices(&index_raw.data); @@ -512,6 +524,7 @@ fn analyze_sst_file(db_path: &Path, info: &SstInfo) -> Result<SstStats> {             block_offsets_start,             block_index,             info.sequence_number,+            compression,         ) {             Ok(raw) => raw,             Err(e) => {@@ -930,7 +943,7 @@ fn main() -> Result<()> {         db_path.display()     ); -    // Analyze and report by family+    // Analyze and report by family.     for (family, sst_list) in &family_sst_info {         let mut family_stats = SstStats::default();         let mut sst_stats_list: Vec<(u32, SstStats)> = Vec::new();

@@ -1,25 +1,59 @@-use std::{mem::MaybeUninit, rc::Rc, sync::Arc};+use std::{cell::RefCell, mem::MaybeUninit, rc::Rc, sync::Arc}; -use anyhow::{Context, Result};+use anyhow::{Context, Result, ensure}; use lzzzz::lz4::{self, decompress}; +/// Compression algorithm used for a family's SST blocks and blob values.+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]+#[repr(u8)]+pub enum Compression {+    /// Fast LZ4 compression using the default acceleration level.+    #[default]+    Lz4 = 0,+    /// Zstandard compression at level 3.+    Zstd3 = 1,+}++thread_local! {+    /// Zstd decompression contexts are reusable and relatively expensive to create. Keep one per+    /// worker thread to avoid allocation on every block read without a global lock.+    static ZSTD_DECOMPRESSOR: RefCell<zstd::bulk::Decompressor<'static>> = RefCell::new(+        zstd::bulk::Decompressor::new().expect("zstd decompressor initialization should succeed")+    );+}+ /// Decompresses `block` into `dest`, verifying the output length matches `expected_len`.-fn decompress_block(block: &[u8], dest: &mut [u8], expected_len: u32) -> Result<()> {+fn decompress_block(+    compression: Compression,+    block: &[u8],+    dest: &mut [u8],+    expected_len: u32,+) -> Result<()> {     debug_assert!(         expected_len > 0,         "decompress_block called with uncompressed_length=0; uncompressed blocks should use \          zero-copy mmap path"     );-    let bytes_written = decompress(block, dest).with_context(|| {+    let bytes_written = match compression {+        Compression::Lz4 => decompress(block, dest).map_err(anyhow::Error::from),+        Compression::Zstd3 => ZSTD_DECOMPRESSOR.with_borrow_mut(|decompressor| {+            decompressor+                .decompress_to_buffer(block, dest)+                .map_err(anyhow::Error::from)+        }),+    }+    .with_context(|| {         format!(-            "Failed to decompress block ({} bytes compressed, {} bytes uncompressed)",+            "Failed to decompress {compression:?} block ({} bytes compressed, {} bytes \+             uncompressed)",             block.len(),             expected_len         )     })?;-    assert_eq!(-        bytes_written, expected_len as usize,-        "Decompressed length does not match expected length"+    ensure!(+        bytes_written == expected_len as usize,+        "Decompressed length does not match expected length: decompressed {bytes_written} bytes, \+         expected {expected_len}"     );     Ok(()) }@@ -28,27 +62,35 @@ fn decompress_block(block: &[u8], dest: &mut [u8], expected_len: u32) -> Result< /// /// The caller must ensure `uncompressed_length > 0` (i.e., the block is actually compressed). /// Uncompressed blocks should be handled via zero-copy mmap slices before calling this.-pub fn decompress_into_arc(uncompressed_length: u32, block: &[u8]) -> Result<Arc<[u8]>> {+pub(crate) fn decompress_into_arc(+    compression: Compression,+    uncompressed_length: u32,+    block: &[u8],+) -> Result<Arc<[u8]>> {     // Allocate directly into an Arc to avoid a copy. The buffer is uninitialized;     // decompression will overwrite it completely (verified by decompress_block).     let buffer: Arc<[MaybeUninit<u8>]> = Arc::new_uninit_slice(uncompressed_length as usize);-    // Safety: decompression will fully initialize the buffer (verified by the assert in+    // Safety: decompression will fully initialize the buffer (verified by the length check in     // decompress_block).     let mut buffer = unsafe { buffer.assume_init() };     // We just created this Arc so refcount is 1; get_mut always succeeds.     let dest = Arc::get_mut(&mut buffer).expect("Arc refcount should be 1");-    decompress_block(block, dest, uncompressed_length)?;+    decompress_block(compression, block, dest, uncompressed_length)?;     Ok(buffer) }  /// Like [`decompress_into_arc`] but returns an `Rc<[u8]>` for thread-local use.-pub fn decompress_into_rc(uncompressed_length: u32, block: &[u8]) -> Result<Rc<[u8]>> {+pub(crate) fn decompress_into_rc(+    compression: Compression,+    uncompressed_length: u32,+    block: &[u8],+) -> Result<Rc<[u8]>> {     let buffer: Rc<[MaybeUninit<u8>]> = Rc::new_uninit_slice(uncompressed_length as usize);-    // Safety: decompression will fully initialize the buffer (verified by the assert in+    // Safety: decompression will fully initialize the buffer (verified by the length check in     // decompress_block).     let mut buffer = unsafe { buffer.assume_init() };     let dest = Rc::get_mut(&mut buffer).expect("Rc refcount should be 1");-    decompress_block(block, dest, uncompressed_length)?;+    decompress_block(compression, block, dest, uncompressed_length)?;     Ok(buffer) } @@ -57,8 +99,62 @@ pub fn checksum_block(data: &[u8]) -> u32 {     crc32fast::hash(data) } -#[tracing::instrument(level = "trace", skip_all)]-pub fn compress_into_buffer(block: &[u8], buffer: &mut Vec<u8>) -> Result<()> {-    lz4::compress_to_vec(block, buffer, lz4::ACC_LEVEL_DEFAULT).context("Compression failed")?;-    Ok(())+/// Reusable compressor for a stream of blocks using the same family configuration.+pub(crate) struct Compressor {+    compression: Compression,+    zstd: Option<zstd::bulk::Compressor<'static>>,+}++impl Compressor {+    pub(crate) fn new(compression: Compression) -> Result<Self> {+        let zstd = match compression {+            Compression::Zstd3 => {+                Some(zstd::bulk::Compressor::new(3).context("Failed to create zstd compressor")?)+            }+            Compression::Lz4 => None,+        };+        Ok(Self { compression, zstd })+    }++    #[tracing::instrument(level = "trace", skip_all)]+    pub(crate) fn compress_into_buffer(+        &mut self,+        block: &[u8],+        buffer: &mut Vec<u8>,+    ) -> Result<()> {+        match self.compression {+            Compression::Lz4 => {+                lz4::compress_to_vec(block, buffer, lz4::ACC_LEVEL_DEFAULT)+                    .context("LZ4 compression failed")?;+            }+            Compression::Zstd3 => {+                buffer.reserve(zstd::zstd_safe::compress_bound(block.len()));+                self.zstd+                    .as_mut()+                    .expect("zstd compressor not initialized")+                    .compress_to_buffer(block, buffer)+                    .context("zstd compression failed")?;+            }+        }+        Ok(())+    }+}++#[cfg(test)]+mod tests {+    use super::*;++    #[test]+    fn compression_round_trips() {+        let input = b"turbo persistence compression ".repeat(1024);+        for compression in [Compression::Lz4, Compression::Zstd3] {+            let mut compressor = Compressor::new(compression).unwrap();+            let mut compressed = Vec::new();+            compressor+                .compress_into_buffer(&input, &mut compressed)+                .unwrap();+            let output = decompress_into_arc(compression, input.len() as u32, &compressed).unwrap();+            assert_eq!(&*output, input);+        }+    } }

@@ -29,7 +29,7 @@ use tracing::span::EnteredSpan;  pub use crate::compaction::selector::CompactConfig; use crate::{-    DbConfig, FamilyKind, QueryKey,+    Compression, DbConfig, FamilyKind, QueryKey,     arc_bytes::ArcBytes,     compaction::selector::{Compactable, get_merge_segments},     compression::{checksum_block, decompress_into_arc},@@ -324,7 +324,7 @@ pub struct TurboPersistence<S: ParallelScheduler, const FAMILIES: usize> {     /// A cache for decompressed value blocks. Allocated lazily on first read via     /// [`Self::value_block_cache`]; see [`Self::key_block_cache`].     value_block_cache: OnceLock<BlockCache>,-    /// Per-family configuration for file limits.+    /// Per-family storage configuration.     config: DbConfig<FAMILIES>,     /// Statistics for the database.     #[cfg(feature = "stats")]@@ -634,7 +634,7 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>         let mut meta_files = self             .parallel_scheduler             .parallel_map_collect::<_, _, Result<Vec<MetaFile>>>(&meta_files, |&seq| {-                let meta_file = MetaFile::open(&self.path, seq)?;+                let meta_file = MetaFile::open(&self.path, seq, Some(&self.config.family_configs))?;                 Ok(meta_file)             })?; @@ -653,7 +653,7 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>      /// Reads and decompresses a blob file. This is not backed by any cache.     #[tracing::instrument(level = "info", name = "reading database blob", skip_all)]-    fn read_blob(&self, seq: u32) -> Result<ArcBytes> {+    fn read_blob(&self, seq: u32, compression: Compression) -> Result<ArcBytes> {         let path = self.path.join(format!("{seq:08}.blob"));         let file = File::open(&path)?;         let mmap = unsafe { Mmap::map(file.file()) }.with_context(|| {@@ -686,7 +686,7 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>             );         } -        let buffer = decompress_into_arc(uncompressed_length, reader)?;+        let buffer = decompress_into_arc(compression, uncompressed_length, reader)?;         Ok(ArcBytes::from(buffer))     } @@ -926,7 +926,8 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>             .parallel_map_collect_owned::<_, _, Result<Vec<_>>>(sync_items, |item| match item {                 SyncItem::Meta(seq, file) => {                     file.sync_data()?;-                    let meta_file = MetaFile::open(&self.path, seq)?;+                    let meta_file =+                        MetaFile::open(&self.path, seq, Some(&self.config.family_configs))?;                     Ok(SyncResult::Meta(meta_file))                 }                 SyncItem::Sst(file) => {@@ -1593,7 +1594,11 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>                                     let meta_index = ssts_with_ranges[index].meta_index;                                     let index_in_meta = ssts_with_ranges[index].index_in_meta;                                     let entry = meta_files[meta_index].entry(index_in_meta);-                                    StaticSortedFileIter::open(path, entry.sst_metadata())+                                    StaticSortedFileIter::open(+                                        path,+                                        entry.sst_metadata(),+                                        self.config.family_configs[family as usize].compression,+                                    )                                 })                                 .collect::<Result<Vec<_>>>()?; @@ -1610,17 +1615,19 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>                                 /// used set).                                 writer: Option<(u32, StreamingSstWriter<LookupEntry>)>,                                 flags: MetaEntryFlags,+                                compression: Compression,                                 new_sst_files:                                     Vec<(u32, File, StaticSortedFileBuilderMeta<'static>)>,                                 /// Hash of the last key added. Used to ensure we only split                                 /// SST files at key boundaries (not mid-key-group for MultiValue).                                 last_hash: Option<u64>,                             }                             impl Collector {-                                fn new(flags: MetaEntryFlags) -> Self {+                                fn new(flags: MetaEntryFlags, compression: Compression) -> Self {                                     Self {                                         writer: None,                                         flags,+                                        compression,                                         new_sst_files: Vec::new(),                                         last_hash: None,                                     }@@ -1641,6 +1648,7 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>                                             &sst_path,                                             self.flags,                                             MAX_ENTRIES_PER_COMPACTED_FILE as u64,+                                            self.compression,                                         )?;                                         self.writer = Some((seq, writer));                                     }@@ -1700,8 +1708,12 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>                                     }                                 }                             }-                            let mut used_collector = Collector::new(MetaEntryFlags::WARM);-                            let mut unused_collector = Collector::new(MetaEntryFlags::COLD);+                            let compression =+                                self.config.family_configs[family as usize].compression;+                            let mut used_collector =+                                Collector::new(MetaEntryFlags::WARM, compression);+                            let mut unused_collector =+                                Collector::new(MetaEntryFlags::COLD, compression);                             let mut current_key: Option<RcBytes> = None;                             let mut keys_written = 0; @@ -1834,7 +1846,10 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>                     let mut blob_seq_numbers_to_delete = Vec::with_capacity(blob_delete_len);                      let meta_seq = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;-                    let mut meta_file_builder = MetaFileBuilder::new(family);+                    let mut meta_file_builder = MetaFileBuilder::new(+                        family,+                        self.config.family_configs[family as usize].compression,+                    );                      let mut keys_written = 0;                     self.parallel_scheduler.block_in_place(|| {@@ -2091,7 +2106,10 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>                                 LookupValue::Blob { sequence_number } => {                                     #[cfg(feature = "stats")]                                     self.stats.hits_blob.fetch_add(1, Ordering::Relaxed);-                                    let blob = self.read_blob(sequence_number)?;+                                    let blob = self.read_blob(+                                        sequence_number,+                                        self.config.family_configs[family].compression,+                                    )?;                                     if deleted_values.iter().any(|d| **d == *blob) {                                         continue;                                     }@@ -2230,7 +2248,10 @@ impl<S: ParallelScheduler, const FAMILIES: usize> TurboPersistence<S, FAMILIES>                     LookupValue::Blob { sequence_number } => {                         #[cfg(feature = "stats")]                         self.stats.hits_blob.fetch_add(1, Ordering::Relaxed);-                        let blob = self.read_blob(sequence_number)?;+                        let blob = self.read_blob(+                            sequence_number,+                            self.config.family_configs[family].compression,+                        )?;                         result_size += blob.len();                         Some(blob)                     }

@@ -29,7 +29,7 @@ mod write_batch; mod tests;  pub use arc_bytes::ArcBytes;-pub use compression::checksum_block;+pub use compression::{Compression, checksum_block}; pub use db::{     CommitStats, CompactConfig, CurrentDbVersion, MetaFileEntryInfo, MetaFileInfo,     TurboPersistence, read_current_version,@@ -54,12 +54,13 @@ pub enum FamilyKind { pub struct FamilyConfig {     pub name: &'static str,     pub kind: FamilyKind,+    pub compression: Compression, } -/// Database-wide configuration with per-family settings.+/// Database-wide configuration with per-family storage settings. ///-/// Each family (keyspace) can have different file size limits to optimize-/// for its specific access patterns and data characteristics.+/// Each family (keyspace) can select storage behavior suited to its access patterns and data+/// characteristics. #[derive(Clone, Debug)] pub struct DbConfig<const FAMILIES: usize> {     pub family_configs: [FamilyConfig; FAMILIES],@@ -71,6 +72,7 @@ impl<const FAMILIES: usize> Default for DbConfig<FAMILIES> {             family_configs: [FamilyConfig {                 name: "unknown",                 kind: FamilyKind::SingleValue,+                compression: Compression::Lz4,             }; FAMILIES],         }     }

@@ -5,7 +5,7 @@ use std::{     sync::OnceLock, }; -use anyhow::{Context, Result, bail};+use anyhow::{Context, Result, bail, ensure}; use bitfield::bitfield; use byteorder::{BE, ReadBytesExt}; use fs_err::File;@@ -14,7 +14,7 @@ use smallvec::SmallVec; use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout, Ref, big_endian as be};  use crate::{-    QueryKey,+    Compression, FamilyConfig, QueryKey,     lookup_entry::LookupValue,     mmap_helper::advise_mmap_for_persistence,     static_sorted_file::{BlockCache, SstLookupResult, StaticSortedFile, StaticSortedFileMetaData},@@ -117,6 +117,8 @@ pub struct MetaEntry {     ///     /// The `'static` lifetime is transmuted — the actual borrow is from `MetaFile::mmap`.     amqf: qfilter::FilterRef<'static>,+    /// Compression recorded in this entry's meta file.+    compression: Compression,     /// The static sorted file that is lazily loaded     sst: OnceLock<StaticSortedFile>, }@@ -153,12 +155,14 @@ impl MetaEntry {      fn sst(&self, meta: &MetaFile) -> Result<&StaticSortedFile> {         self.sst.get_or_try_init(|| {-            StaticSortedFile::open(&meta.db_path, self.sst_data).with_context(|| {-                format!(-                    "Unable to open static sorted file referenced from {:08}.meta",-                    meta.sequence_number()-                )-            })+            StaticSortedFile::open(&meta.db_path, self.sst_data, self.compression).with_context(+                || {+                    format!(+                        "Unable to open static sorted file referenced from {:08}.meta",+                        meta.sequence_number()+                    )+                },+            )         })     } @@ -244,6 +248,8 @@ pub struct MetaFile {     sequence_number: u32,     /// The key family of the SST files in this meta file.     family: u32,+    /// Compression recorded for this family.+    compression: Compression,     /// The entries of the file. Dropped before `mmap` (field declaration order).     entries: Vec<MetaEntry>,     /// The entries that have been marked as obsolete.@@ -266,14 +272,28 @@ pub struct MetaFile { impl MetaFile {     /// Opens a meta file at the given path. Memory maps the entire file and eagerly deserializes     /// all AMQF filters as zero-copy [`qfilter::FilterRef`]s that borrow from the mmap.-    pub fn open(db_path: &Path, sequence_number: u32) -> Result<Self> {+    pub fn open(+        db_path: &Path,+        sequence_number: u32,+        family_configs: Option<&[FamilyConfig]>,+    ) -> Result<Self> {         let filename = format!("{sequence_number:08}.meta");         let path = db_path.join(&filename);-        Self::open_internal(db_path.to_path_buf(), sequence_number, &path)-            .with_context(|| format!("Unable to open meta file {filename}"))+        Self::open_internal(+            db_path.to_path_buf(),+            sequence_number,+            &path,+            family_configs,+        )+        .with_context(|| format!("Unable to open meta file {filename}"))     } -    fn open_internal(db_path: PathBuf, sequence_number: u32, path: &Path) -> Result<Self> {+    fn open_internal(+        db_path: PathBuf,+        sequence_number: u32,+        path: &Path,+        family_configs: Option<&[FamilyConfig]>,+    ) -> Result<Self> {         let file = File::open(path)?;         let mmap = unsafe { MmapOptions::new().map(file.file()) }.context("Failed to mmap")?;         #[cfg(unix)]@@ -287,6 +307,22 @@ impl MetaFile {             bail!("Invalid magic number");         }         let family = reader.read_u32::<BE>()?;+        let compression = match reader.read_u8()? {+            value if value == Compression::Lz4 as u8 => Compression::Lz4,+            value if value == Compression::Zstd3 as u8 => Compression::Zstd3,+            value => bail!("Invalid compression algorithm {value}"),+        };+        if let Some(configs) = family_configs {+            let configured = configs+                .get(family as usize)+                .with_context(|| format!("No configuration for family {family}"))?+                .compression;+            ensure!(+                compression == configured,+                "Compression configuration mismatch for family {family}: meta file uses \+                 {compression:?}, runtime config uses {configured:?}"+            );+        }         let obsolete_count = reader.read_u32::<BE>()?;         let mut obsolete_sst_files = Vec::with_capacity(obsolete_count as usize);         for _ in 0..obsolete_count {@@ -344,6 +380,7 @@ impl MetaFile {                 flags,                 amqf_data_offset: start_of_amqf_data_offset..end_of_amqf_data_offset,                 amqf,+                compression,                 sst: OnceLock::new(),             });             start_of_amqf_data_offset = end_of_amqf_data_offset;@@ -356,6 +393,7 @@ impl MetaFile {             db_path,             sequence_number,             family,+            compression,             entries,             obsolete_entries: Vec::new(),             obsolete_sst_files,@@ -386,6 +424,10 @@ impl MetaFile {         self.family     } +    pub fn compression(&self) -> Compression {+        self.compression+    }+     /// The on-disk size of this meta file in bytes (the length of its memory map).     pub fn byte_size(&self) -> u64 {         self.mmap.len() as u64

@@ -10,12 +10,14 @@ use qfilter::Filter; use zerocopy::IntoBytes;  use crate::{+    Compression,     meta_file::{EntryHeader, META_FILE_MAGIC},     static_sorted_file_builder::StaticSortedFileBuilderMeta, };  pub struct MetaFileBuilder<'a> {     family: u32,+    compression: Compression,     /// Entries in the meta file, tuples of (sequence_number, StaticSortedFileBuilderMetaResult)     entries: Vec<(u32, StaticSortedFileBuilderMeta<'a>)>,     /// Obsolete SST files, represented by their sequence numbers@@ -25,9 +27,10 @@ pub struct MetaFileBuilder<'a> { }  impl<'a> MetaFileBuilder<'a> {-    pub fn new(family: u32) -> Self {+    pub fn new(family: u32, compression: Compression) -> Self {         Self {             family,+            compression,             entries: Vec::new(),             obsolete_sst_files: Vec::new(),             used_key_hashes_amqf: None,@@ -59,6 +62,7 @@ impl<'a> MetaFileBuilder<'a> {         let mut file = CountingWriter::new(BufWriter::new(File::create(file)?));         file.write_u32::<BE>(META_FILE_MAGIC)?; // Magic number         file.write_u32::<BE>(self.family)?;+        file.write_u8(self.compression as u8)?;          self.obsolete_sst_files.sort();         file.write_u32::<BE>(self.obsolete_sst_files.len() as u32)?;

@@ -9,6 +9,7 @@ use std::{ use memmap2::Mmap;  use crate::{+    Compression,     compression::decompress_into_rc,     shared_bytes::{SharedBytes, is_subslice_of}, };@@ -124,8 +125,13 @@ impl SharedBytes for RcBytes {         }     } -    fn from_decompressed(uncompressed_length: u32, block: &[u8]) -> anyhow::Result<Self> {+    fn from_decompressed(+        compression: Compression,+        uncompressed_length: u32,+        block: &[u8],+    ) -> anyhow::Result<Self> {         Ok(RcBytes::from(decompress_into_rc(+            compression,             uncompressed_length,             block,         )?))

@@ -2,6 +2,8 @@ use std::ops::{Deref, Range};  use memmap2::Mmap; +use crate::Compression;+ /// Trait abstracting over `ArcBytes` and `RcBytes`. /// /// Both types are owned byte slices backed by either a ref-counted heap@@ -36,7 +38,11 @@ pub trait SharedBytes: Clone + Deref<Target = [u8]> + Sized {     unsafe fn from_mmap(mmap: &Self::MmapHandle, subslice: &[u8]) -> Self;      /// Creates an instance from a decompressed block.-    fn from_decompressed(uncompressed_length: u32, block: &[u8]) -> anyhow::Result<Self>;+    fn from_decompressed(+        compression: Compression,+        uncompressed_length: u32,+        block: &[u8],+    ) -> anyhow::Result<Self>; }  /// Returns `true` if `subslice` lies entirely within `backing`.

@@ -17,7 +17,7 @@ use rustc_hash::FxHasher; use smallvec::SmallVec;  use crate::{-    QueryKey,+    Compression, QueryKey,     arc_bytes::ArcBytes,     be,     compression::checksum_block,@@ -155,6 +155,7 @@ trait ValueBlockCache<B: SharedBytes> {         mmap: &B::MmapHandle,         meta: &StaticSortedFileMetaData,         block_index: u16,+        compression: Compression,     ) -> Result<B>; } @@ -174,26 +175,39 @@ impl ValueBlockCache<ArcBytes> for ArcBlockCacheReader<'_> {         mmap: &Arc<Mmap>,         meta: &StaticSortedFileMetaData,         block_index: u16,+        compression: Compression,     ) -> Result<ArcBytes> {-        get_or_cache_block(mmap, meta, block_index, self.cache, self.verified_blocks)+        get_or_cache_block(+            mmap,+            meta,+            block_index,+            self.cache,+            self.verified_blocks,+            compression,+        )     } }  /// Iteration-path: lightweight single-entry cache for sequential reads.-impl ValueBlockCache<RcBytes> for &mut Option<(u16, RcBytes)> {+struct RcBlockCacheReader<'a> {+    cache: &'a mut Option<(u16, RcBytes)>,+}++impl ValueBlockCache<RcBytes> for RcBlockCacheReader<'_> {     fn get_or_read(         self,         mmap: &Rc<Mmap>,         meta: &StaticSortedFileMetaData,         block_index: u16,+        compression: Compression,     ) -> Result<RcBytes> {-        if let Some((idx, block)) = self.as_ref()+        if let Some((idx, block)) = self.cache.as_ref()             && *idx == block_index         {             return Ok(block.clone());         }-        let block: RcBytes = read_block_generic(mmap, meta, block_index)?;-        *self = Some((block_index, block.clone()));+        let block: RcBytes = read_block_generic(mmap, meta, block_index, compression)?;+        *self.cache = Some((block_index, block.clone()));         Ok(block)     } }@@ -226,12 +240,17 @@ pub struct StaticSortedFile {     /// bitmap the CRC would be re-computed on every access. `Relaxed` ordering     /// suffices: racing first-time verifications are idempotent.     verified_blocks: Box<[AtomicU64]>,+    compression: Compression, }  impl StaticSortedFile {-    /// Opens an SST file at the given path. This memory maps the file, but does not read it yet.-    /// It's lazy read on demand.-    pub fn open(db_path: &Path, meta: StaticSortedFileMetaData) -> Result<Self> {+    /// Opens an SST file at the given path with the compression algorithm specified by its meta+    /// file. This memory maps the file, but does not read it yet.+    pub fn open(+        db_path: &Path,+        meta: StaticSortedFileMetaData,+        compression: Compression,+    ) -> Result<Self> {         let filename = format!("{:08}.sst", meta.sequence_number);         let path = db_path.join(&filename);         let file = File::open(&path)?;@@ -257,6 +276,7 @@ impl StaticSortedFile {             meta,             mmap: Arc::new(mmap),             verified_blocks,+            compression,         })     } @@ -281,6 +301,7 @@ impl StaticSortedFile {             index_block_index,             key_block_cache,             &self.verified_blocks,+            self.compression,         )?;         let key_block_index = self.lookup_index_block(&index_block, key_hash)?; @@ -290,6 +311,7 @@ impl StaticSortedFile {             key_block_index,             key_block_cache,             &self.verified_blocks,+            self.compression,         )?;         let reader = ArcBlockCacheReader {             cache: value_block_cache,@@ -491,7 +513,15 @@ impl StaticSortedFile {         key_block_arc: &ArcBytes,         reader: ArcBlockCacheReader<'_>,     ) -> Result<LookupValue> {-        handle_key_match_generic(&self.mmap, &self.meta, ty, val, key_block_arc, reader)+        handle_key_match_generic(+            &self.mmap,+            &self.meta,+            ty,+            val,+            key_block_arc,+            self.compression,+            reader,+        )     } } @@ -509,6 +539,7 @@ fn get_or_cache_block(     block_index: u16,     cache: &BlockCache,     verified_blocks: &[AtomicU64],+    compression: Compression, ) -> Result<ArcBytes> {     let (uncompressed_length, checksum, block_data) = get_raw_block_slice(mmap, meta, block_index)         .with_context(|| {@@ -533,13 +564,15 @@ fn get_or_cache_block(                 // A cached block may have been evicted, so re-reading still                 // benefits from the bitmap to skip redundant CRC verification.                 verify_checksum_once(meta, block_data, checksum, block_index, verified_blocks)?;-                let block = ArcBytes::from_decompressed(uncompressed_length, block_data)-                    .with_context(|| {-                        format!(-                            "Failed to decompress block {} from {:08}.sst ({} bytes uncompressed)",-                            block_index, meta.sequence_number, uncompressed_length-                        )-                    })?;+                let block =+                    ArcBytes::from_decompressed(compression, uncompressed_length, block_data)+                        .with_context(|| {+                            format!(+                                "Failed to decompress block {} from {:08}.sst ({} bytes \+                                 uncompressed)",+                                block_index, meta.sequence_number, uncompressed_length+                            )+                        })?;                 let _ = guard.insert(block.clone());                 block             }@@ -673,6 +706,7 @@ fn read_block_generic<B: SharedBytes>(     mmap: &B::MmapHandle,     meta: &StaticSortedFileMetaData,     block_index: u16,+    compression: Compression, ) -> Result<B> {     let (uncompressed_length, expected_checksum, block) =         get_raw_block_slice(mmap, meta, block_index).with_context(|| {@@ -689,12 +723,13 @@ fn read_block_generic<B: SharedBytes>(         return Ok(unsafe { B::from_mmap(mmap, block) });     } -    let buffer = B::from_decompressed(uncompressed_length, block).with_context(|| {-        format!(-            "Failed to decompress block {} from {:08}.sst ({} bytes uncompressed)",-            block_index, meta.sequence_number, uncompressed_length-        )-    })?;+    let buffer =+        B::from_decompressed(compression, uncompressed_length, block).with_context(|| {+            format!(+                "Failed to decompress block {} from {:08}.sst ({} bytes uncompressed)",+                block_index, meta.sequence_number, uncompressed_length+            )+        })?;     Ok(buffer) } @@ -705,6 +740,7 @@ fn handle_key_match_generic<B: SharedBytes>(     ty: u8,     val: &[u8],     key_block: &B,+    compression: Compression,     reader: impl ValueBlockCache<B>, ) -> Result<LookupValue<B>> {     Ok(match ty {@@ -713,13 +749,13 @@ fn handle_key_match_generic<B: SharedBytes>(             let size = be::read_u16(&val[2..]) as usize;             let position = be::read_u32(&val[4..]) as usize;             let value = reader-                .get_or_read(mmap, meta, block)?+                .get_or_read(mmap, meta, block, compression)?                 .slice(position..position + size);             LookupValue::Slice { value }         }         KEY_BLOCK_ENTRY_TYPE_MEDIUM => {             let block = be::read_u16(val);-            let value = read_block_generic(mmap, meta, block)?;+            let value = read_block_generic(mmap, meta, block, compression)?;             LookupValue::Slice { value }         }         KEY_BLOCK_ENTRY_TYPE_BLOB => {@@ -763,6 +799,7 @@ pub struct StaticSortedFileIter {     /// value blocks sequentially and don't revisit earlier blocks, so caching     /// just the current one avoids redundant decompression.     value_block_cache: Option<(u16, RcBytes)>,+    compression: Compression, }  enum CurrentKeyBlockKind {@@ -796,10 +833,14 @@ impl Iterator for StaticSortedFileIter { }  impl StaticSortedFileIter {-    /// Opens an SST file for sequential iteration. Uses `MADV_SEQUENTIAL` for-    /// read-ahead and wraps the mmap in `Rc<Mmap>` directly (no `Arc`),-    /// eliminating all atomic refcounting during iteration.-    pub fn open(db_path: &Path, meta: StaticSortedFileMetaData) -> Result<Self> {+    /// Opens an SST file for sequential iteration with the compression algorithm specified by its+    /// meta file. Uses `MADV_SEQUENTIAL` for read-ahead and wraps the mmap in `Rc<Mmap>` directly+    /// (no `Arc`), eliminating all atomic refcounting during iteration.+    pub fn open(+        db_path: &Path,+        meta: StaticSortedFileMetaData,+        compression: Compression,+    ) -> Result<Self> {         let filename = format!("{:08}.sst", meta.sequence_number);         let path = db_path.join(&filename);         let file = File::open(&path)?;@@ -813,13 +854,17 @@ impl StaticSortedFileIter {         #[cfg(unix)]         mmap.advise(memmap2::Advice::Sequential)?;         advise_mmap_for_persistence(&mmap)?;-        Self::new(Rc::new(mmap), meta)+        Self::new(Rc::new(mmap), meta, compression)             .with_context(|| format!("Unable to open static sorted file {filename}"))     } -    fn new(mmap: Rc<Mmap>, meta: StaticSortedFileMetaData) -> Result<Self> {+    fn new(+        mmap: Rc<Mmap>,+        meta: StaticSortedFileMetaData,+        compression: Compression,+    ) -> Result<Self> {         let root_block_index = meta.block_count - 1;-        let block: RcBytes = read_block_generic(&mmap, &meta, root_block_index)?;+        let block: RcBytes = read_block_generic(&mmap, &meta, root_block_index, compression)?;         let block_type = block[0];          // The builder always writes an index block as the root block.@@ -838,7 +883,7 @@ impl StaticSortedFileIter {             - size_of::<u16>())             / INDEX_BLOCK_ENTRY_SIZE; -        let current_key_block = Self::parse_key_block(&mmap, &meta, first_child)?;+        let current_key_block = Self::parse_key_block(&mmap, &meta, first_child, compression)?;         Ok(StaticSortedFileIter {             mmap,             meta,@@ -847,6 +892,7 @@ impl StaticSortedFileIter {             index_pos: 1,             current_key_block,             value_block_cache: None,+            compression,         })     } @@ -855,8 +901,9 @@ impl StaticSortedFileIter {         mmap: &Rc<Mmap>,         meta: &StaticSortedFileMetaData,         block_index: u16,+        compression: Compression,     ) -> Result<CurrentKeyBlock> {-        let block: RcBytes = read_block_generic(mmap, meta, block_index)?;+        let block: RcBytes = read_block_generic(mmap, meta, block_index, compression)?;         let data = &*block;         ensure!(data.len() >= 4, "key block too short");         let block_type = data[0];@@ -959,7 +1006,10 @@ impl StaticSortedFileIter {                         ty,                         val,                         &kb.entries,-                        &mut self.value_block_cache,+                        self.compression,+                        RcBlockCacheReader {+                            cache: &mut self.value_block_cache,+                        },                     )?                     .into()                 };@@ -976,7 +1026,7 @@ impl StaticSortedFileIter {                 let block_index = be::read_u16(&self.index_entries[base..]);                 self.index_pos += 1;                 self.current_key_block =-                    Self::parse_key_block(&self.mmap, &self.meta, block_index)?;+                    Self::parse_key_block(&self.mmap, &self.meta, block_index, self.compression)?;             } else {                 return Ok(None);             }

@@ -10,7 +10,8 @@ use byteorder::{BE, ByteOrder, WriteBytesExt}; use fs_err::File;  use crate::{-    compression::{checksum_block, compress_into_buffer},+    Compression,+    compression::{Compressor, checksum_block},     constants::{MAX_INLINE_VALUE_SIZE, MAX_SMALL_VALUE_SIZE, MIN_SMALL_VALUE_BLOCK_SIZE},     meta_file::MetaEntryFlags,     static_sorted_file::{@@ -309,9 +310,10 @@ pub fn write_static_stored_file<E: Entry>(     entries: &[E],     file: &Path,     flags: MetaEntryFlags,+    compression: Compression, ) -> Result<(StaticSortedFileBuilderMeta<'static>, File)> {     debug_assert!(entries.iter().map(|e| e.key_hash()).is_sorted());-    let mut writer = StreamingSstWriter::new(file, flags, entries.len() as u64)?;+    let mut writer = StreamingSstWriter::new(file, flags, entries.len() as u64, compression)?;     for entry in entries {         writer.add(entry)?;     }@@ -363,9 +365,10 @@ fn write_block_to_file(     block_offsets: &mut Vec<u32>,     block: &[u8],     try_compress: bool,+    compressor: &mut Compressor, ) -> Result<u16> {     let (uncompressed_size, data_to_write): (u32, &[u8]) = if try_compress {-        compress_into_buffer(block, compress_buffer)?;+        compressor.compress_into_buffer(block, compress_buffer)?;         // Same threshold as LevelDB/RocksDB: require at least 12.5% savings.         if compress_buffer.len() < block.len() - (block.len() / 8) {             (block.len().try_into().unwrap(), compress_buffer.as_slice())@@ -505,6 +508,7 @@ pub struct StreamingSstWriter<E: Entry> {     file: Option<BufWriter<File>>,     compress_buffer: Vec<u8>,     block_offsets: Vec<u32>,+    compressor: Compressor,      /// Pending key entries waiting to be flushed as key blocks.     ///@@ -578,8 +582,14 @@ impl<E: Entry> StreamingSstWriter<E> {     /// Creates a new streaming SST writer.     ///     /// `max_entry_count` is used to pre-allocate buffers and estimate block counts.-    pub fn new(file: &Path, flags: MetaEntryFlags, max_entry_count: u64) -> Result<Self> {+    pub fn new(+        file: &Path,+        flags: MetaEntryFlags,+        max_entry_count: u64,+        compression: Compression,+    ) -> Result<Self> {         let file = BufWriter::new(File::create(file)?);+        let compressor = Compressor::new(compression)?;          // Estimate number of key blocks based on max entry count.         // Each key block holds up to MAX_KEY_BLOCK_ENTRIES entries.@@ -598,6 +608,7 @@ impl<E: Entry> StreamingSstWriter<E> {             file: Some(file),             compress_buffer: Vec::with_capacity(MIN_SMALL_VALUE_BLOCK_SIZE + MAX_SMALL_VALUE_SIZE),             block_offsets: Vec::with_capacity(estimated_total_blocks),+            compressor,             pending_keys: VecDeque::with_capacity(entries_per_value_block),             first_pending_small_index: 0,             #[cfg(debug_assertions)]@@ -681,6 +692,7 @@ impl<E: Entry> StreamingSstWriter<E> {                     &mut self.block_offsets,                     value,                     true,+                    &mut self.compressor,                 )                 .context("Failed to write value block")?;                 ValueRef::Medium { block_index }@@ -838,6 +850,7 @@ impl<E: Entry> StreamingSstWriter<E> {             &mut self.block_offsets,             &self.pending_small_value_block,             true,+            &mut self.compressor,         )         .context("Failed to write small value block")?; @@ -930,6 +943,7 @@ impl<E: Entry> StreamingSstWriter<E> {             &mut self.block_offsets,             &self.key_buffer,             try_compress,+            &mut self.compressor,         )         .context("Failed to write key block")?;         self.key_block_boundaries.push((first_hash, block_index));@@ -1432,6 +1446,7 @@ mod tests {                 sequence_number: seq,                 block_count: meta.block_count,             },+            Compression::Lz4,         )     } @@ -1443,7 +1458,8 @@ mod tests {         flags: MetaEntryFlags,     ) -> Result<StaticSortedFileBuilderMeta<'static>> {         let sst_path = dir.join(format!("{seq:08}.sst"));-        let mut writer = StreamingSstWriter::new(&sst_path, flags, entries.len() as u64)?;+        let mut writer =+            StreamingSstWriter::new(&sst_path, flags, entries.len() as u64, Compression::Lz4)?;         for entry in entries {             writer.add(entry)?;         }@@ -1679,7 +1695,8 @@ mod tests {         let dir = tempfile::tempdir().unwrap();         let sst_path = dir.path().join("test.sst");         let mut writer =-            StreamingSstWriter::new(&sst_path, MetaEntryFlags::default(), 100).unwrap();+            StreamingSstWriter::new(&sst_path, MetaEntryFlags::default(), 100, Compression::Lz4)+                .unwrap();          let max_entries = 50;         for i in 0..max_entries {@@ -1705,7 +1722,8 @@ mod tests {         let dir = tempfile::tempdir().unwrap();         let sst_path = dir.path().join("test.sst");         let mut writer =-            StreamingSstWriter::new(&sst_path, MetaEntryFlags::default(), 100).unwrap();+            StreamingSstWriter::new(&sst_path, MetaEntryFlags::default(), 100, Compression::Lz4)+                .unwrap();          let value = vec![0u8; 1000];         for i in 0..10 {@@ -1741,15 +1759,20 @@ mod tests {          // Write via convenience function         let batch_path = dir.path().join("00000001.sst");-        let (meta1, _) =-            write_static_stored_file(&entries, &batch_path, MetaEntryFlags::default())?;+        let (meta1, _) = write_static_stored_file(+            &entries,+            &batch_path,+            MetaEntryFlags::default(),+            Compression::Lz4,+        )?;          // Write via streaming API         let streaming_path = dir.path().join("00000002.sst");         let mut writer = StreamingSstWriter::new(             &streaming_path,             MetaEntryFlags::default(),             entries.len() as u64,+            Compression::Lz4,         )?;         for entry in &entries {             writer.add(entry)?;@@ -1769,13 +1792,15 @@ mod tests {                 sequence_number: 1,                 block_count: meta1.block_count,             },+            Compression::Lz4,         )?;         let sst2 = StaticSortedFile::open(             dir.path(),             StaticSortedFileMetaData {                 sequence_number: 2,                 block_count: meta2.block_count,             },+            Compression::Lz4,         )?;         let kc = make_cache();         let vc = make_cache();@@ -1830,8 +1855,13 @@ mod tests {     fn close_empty_writer_panics() {         let dir = tempfile::tempdir().unwrap();         let sst_path = dir.path().join("empty.sst");-        let writer =-            StreamingSstWriter::<TestEntry>::new(&sst_path, MetaEntryFlags::default(), 0).unwrap();+        let writer = StreamingSstWriter::<TestEntry>::new(+            &sst_path,+            MetaEntryFlags::default(),+            0,+            Compression::Lz4,+        )+        .unwrap();         writer.close().unwrap();     } 

@@ -4,7 +4,7 @@ use anyhow::Result; use rayon::iter::{IntoParallelIterator, ParallelIterator};  use crate::{-    DbConfig, FamilyConfig, FamilyKind,+    Compression, DbConfig, FamilyConfig, FamilyKind,     constants::{MAX_INLINE_VALUE_SIZE, MAX_MEDIUM_VALUE_SIZE, MAX_SMALL_VALUE_SIZE},     db::{CompactConfig, TurboPersistence, read_current_version},     lookup_entry::IterValue,@@ -1055,8 +1055,11 @@ fn batch_get_different_sizes() -> Result<()> {     let tempdir = tempfile::tempdir()?;     let path = tempdir.path(); -    let db = TurboPersistence::<_, 16>::open_with_parallel_scheduler(+    let mut config = DbConfig::default();+    config.family_configs[0].compression = Compression::Zstd3;+    let db = TurboPersistence::<_, 16>::open_with_config_and_parallel_scheduler(         path.to_path_buf(),+        config,         RayonParallelScheduler,     )?; @@ -1109,16 +1112,22 @@ fn batch_get_across_families() -> Result<()> {     let tempdir = tempfile::tempdir()?;     let path = tempdir.path(); -    let db = TurboPersistence::<_, 16>::open_with_parallel_scheduler(+    let mut config = DbConfig::default();+    // set zstd on an arbitrary family, lz4 is used by default+    config.family_configs[2].compression = Compression::Zstd3;+    let db = TurboPersistence::<_, 16>::open_with_config_and_parallel_scheduler(         path.to_path_buf(),+        config.clone(),         RayonParallelScheduler,     )?; -    // Write to multiple families+    // Write compressible values to multiple families so every configured codec is exercised.     let batch = db.write_batch()?;     for family in 0..4u32 {         for i in 0..20u8 {-            batch.put(family, vec![i], vec![family as u8, i].into())?;+            let mut value = vec![family as u8; 1024];+            value[0] = i;+            batch.put(family, vec![i], value.into())?;         }     }     db.commit_write_batch(batch)?;@@ -1132,7 +1141,13 @@ fn batch_get_across_families() -> Result<()> {         for (i, result) in results.iter().enumerate() {             assert_eq!(                 result.as_deref(),-                Some(&vec![family as u8, i as u8][..]),+                Some(+                    &{+                        let mut value = vec![family as u8; 1024];+                        value[0] = i as u8;+                        value+                    }[..]+                ),                 "Failed at family {family}, index {i}"             );         }@@ -1147,6 +1162,29 @@ fn batch_get_across_families() -> Result<()> {     assert_ne!(results_f0[0].as_deref(), results_f1[0].as_deref());      db.shutdown()?;+    drop(db);++    // Reopen with the same family configuration recorded in the meta files.+    let db = TurboPersistence::<_, 16>::open_with_config_and_parallel_scheduler(+        path.to_path_buf(),+        config,+        RayonParallelScheduler,+    )?;+    let value = db.get(2, &vec![7u8])?.expect("zstd family value exists");+    assert_eq!(value[0], 7);+    assert!(value[1..].iter().all(|byte| *byte == 2));+    db.shutdown()?;+    drop(db);++    // Reopening with the wrong codec must fail while validating the meta files.+    assert!(+        TurboPersistence::<RayonParallelScheduler, 16>::open_with_config_and_parallel_scheduler(+            path.to_path_buf(),+            DbConfig::default(),+            RayonParallelScheduler,+        )+        .is_err()+    );     Ok(()) } @@ -1155,8 +1193,11 @@ fn batch_get_after_compaction() -> Result<()> {     let tempdir = tempfile::tempdir()?;     let path = tempdir.path(); -    let db = TurboPersistence::<_, 16>::open_with_parallel_scheduler(+    let mut config = DbConfig::default();+    config.family_configs[0].compression = Compression::Zstd3;+    let db = TurboPersistence::<_, 16>::open_with_config_and_parallel_scheduler(         path.to_path_buf(),+        config,         RayonParallelScheduler,     )?; @@ -1174,7 +1215,7 @@ fn batch_get_after_compaction() -> Result<()> {     let keys_to_fetch: Vec<Vec<u8>> = (0..100u8).map(|i| vec![i]).collect();     let results_before = db.batch_get(0, &keys_to_fetch)?; -    // Compact database+    // Compact database using zstd to cover recompression with a non-default codec.     db.full_compact()?;      // Fetch after compaction@@ -1525,6 +1566,7 @@ fn multi_value_config() -> DbConfig<1> {     config.family_configs[0] = FamilyConfig {         name: "test",         kind: FamilyKind::MultiValue,+        compression: Compression::Lz4,     };     config }@@ -2104,6 +2146,7 @@ fn compaction_deletes_blob_multi_value_tombstone() -> Result<()> {         family_configs: [FamilyConfig {             name: "test",             kind: FamilyKind::MultiValue,+            compression: Compression::Lz4,         }],     }; @@ -2468,7 +2511,7 @@ fn count_tombstones(                 sequence_number: entry.sequence_number,                 block_count: entry.block_count,             };-            for item in StaticSortedFileIter::open(path, sst)? {+            for item in StaticSortedFileIter::open(path, sst, Compression::Lz4)? {                 if matches!(                     item?.value,                     IterValue::KeyDeleted | IterValue::KeyValueDeleted { .. }

@@ -18,7 +18,7 @@ use crate::{     FamilyConfig, FamilyKind, ValueBuffer,     collector::Collector,     collector_entry::CollectorEntry,-    compression::{checksum_block, compress_into_buffer},+    compression::{Compressor, checksum_block},     constants::{MAX_INLINE_VALUE_SIZE, MAX_MEDIUM_VALUE_SIZE, THREAD_LOCAL_SIZE_SHIFT},     db::WriteOperationGuard,     key::StoreKey,@@ -79,7 +79,7 @@ pub struct WriteBatch<'db, K: StoreKey + Send, S: ParallelScheduler, const FAMIL     parallel_scheduler: S,     /// The database path     db_path: PathBuf,-    /// Per-family configuration (kind: SingleValue/MultiValue).+    /// Per-family storage configuration.     #[cfg_attr(not(feature = "verify_sst_content"), allow(dead_code))]     family_configs: [FamilyConfig; FAMILIES],     /// The current sequence number counter. Increased for every new SST file or blob file.@@ -238,7 +238,7 @@ impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize         if value.len() <= MAX_MEDIUM_VALUE_SIZE {             collector.put(key, value);         } else {-            let blob = self.create_blob(&value)?;+            let blob = self.create_blob(family, &value)?;             collector.put_blob(key, blob.seq);             state.new_blob_files.push(blob);         }@@ -318,14 +318,13 @@ impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize             })?;          // Now we flush the global collector(s).-        let mut collector_state = self.collectors[usize_from_u32(family)].lock();+        let family_usize = usize_from_u32(family);+        let mut collector_state = self.collectors[family_usize].lock();+        let family_config = self.family_configs[family_usize];         match &mut *collector_state {             GlobalCollectorState::Unsharded(collector) => {                 if !collector.is_empty() {-                    let sst = self.create_sst_file(-                        family,-                        collector.sorted(self.family_configs[usize_from_u32(family)].kind),-                    )?;+                    let sst = self.create_sst_file(family, collector.sorted(family_config.kind))?;                     collector.clear();                     self.new_sst_files.lock().push(sst);                 }@@ -340,10 +339,8 @@ impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize                 self.parallel_scheduler                     .try_parallel_for_each_mut(&mut shards, |collector| {                         if !collector.is_empty() {-                            let sst = self.create_sst_file(-                                family,-                                collector.sorted(self.family_configs[usize_from_u32(family)].kind),-                            )?;+                            let sst =+                                self.create_sst_file(family, collector.sorted(family_config.kind))?;                             collector.clear();                             self.new_sst_files.lock().push(sst);                             collector.drop_contents();@@ -457,7 +454,10 @@ impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize                 |(family, sst_files)| {                     let family = family as u32;                     let mut entries = 0;-                    let mut builder = MetaFileBuilder::new(family);+                    let mut builder = MetaFileBuilder::new(+                        family,+                        self.family_configs[usize_from_u32(family)].compression,+                    );                     for (seq, sst) in sst_files {                         entries += sst.entries;                         builder.add(seq, sst);@@ -484,10 +484,12 @@ impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize      /// Creates a new blob file with the given value.     #[tracing::instrument(level = "trace", skip(self, value), fields(value_len = value.len()))]-    fn create_blob(&self, value: &[u8]) -> Result<NewFile> {+    fn create_blob(&self, family: u32, value: &[u8]) -> Result<NewFile> {         let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;         let mut compressed = Vec::new();-        compress_into_buffer(value, &mut compressed)+        let compression = self.family_configs[usize_from_u32(family)].compression;+        Compressor::new(compression)?+            .compress_into_buffer(value, &mut compressed)             .context("Compression of value for blob file failed")?;          let mut buffer = Vec::with_capacity(8 + compressed.len());@@ -516,7 +518,14 @@ impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize         let path = self.db_path.join(format!("{seq:08}.sst"));         let (meta, file) = self             .parallel_scheduler-            .block_in_place(|| write_static_stored_file(entries, &path, MetaEntryFlags::FRESH))+            .block_in_place(|| {+                write_static_stored_file(+                    entries,+                    &path,+                    MetaEntryFlags::FRESH,+                    self.family_configs[usize_from_u32(family)].compression,+                )+            })             .with_context(|| format!("Unable to write SST file {seq:08}.sst"))?;          #[cfg(feature = "verify_sst_content")]@@ -540,6 +549,7 @@ impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize                     sequence_number: seq,                     block_count: meta.block_count,                 },+                self.family_configs[usize_from_u32(family)].compression,             )?;             let cache2 = BlockCache::with(                 10,

@@ -1,4 +1,4 @@-use turbo_persistence::{FamilyConfig, FamilyKind};+use turbo_persistence::{Compression, FamilyConfig, FamilyKind};  #[derive(Debug, Clone, Copy)] pub enum KeySpace {@@ -35,14 +35,21 @@ impl KeySpace {     /// Returns the persistence configuration for this keyspace.     pub const fn family_config(&self) -> FamilyConfig {         match self {-            KeySpace::Infra | KeySpace::TaskMeta | KeySpace::TaskData => FamilyConfig {+            KeySpace::Infra | KeySpace::TaskMeta => FamilyConfig {                 name: self.name(),                 kind: FamilyKind::SingleValue,+                compression: Compression::Lz4,+            },+            KeySpace::TaskData => FamilyConfig {+                name: self.name(),+                kind: FamilyKind::SingleValue,+                compression: Compression::Zstd3,             },             KeySpace::TaskCache => FamilyConfig {                 name: self.name(),                 // TaskCache uses hash-based lookups with potential collisions.                 kind: FamilyKind::MultiValue,+                compression: Compression::Lz4,             },         }     }