Upgrade DecimalScheme to the v2 wire format when the writer permits it

benchmarks/compress-bench/README.md

@@ -29,7 +29,7 @@ cargo run -p compress-bench --profile release_debug
 GPU dataset list in `src/main.rs`. It measures decompression only, for two backends:
 
 - **Vortex** — the file is written with CUDA-compatible BtrBlocks encodings only
-  (`only_cuda_compatible`) and a CUDA flat layout, then decoded on the device all the way to
+  (`cuda_compatible_schemes`) and a CUDA flat layout, then decoded on the device all the way to
   canonical arrays.
 - **Parquet** — the file is rewritten with GPU-friendly writer settings (see below) and read
   back with [cuDF](https://github.com/rapidsai/cudf)'s `read_parquet`, which performs the

benchmarks/compress-bench/src/gpu/vortex.rs

@@ -22,7 +22,7 @@ use vortex::array::IntoArray;
 use vortex::array::VortexSessionExecute;
 use vortex::array::arrays::StructArray;
 use vortex::array::arrays::struct_::StructArrayExt;
-use vortex::compressor::BtrBlocksCompressorBuilder;
+use vortex::compressor::BtrBlocksCompressor;
 use vortex::error::VortexResult;
 use vortex::file::OpenOptionsSessionExt;
 use vortex::file::WriteOptionsSessionExt;
@@ -37,7 +37,6 @@ use vortex_bench::compress::CompressedData;
 use vortex_bench::compress::Compressor;
 use vortex_bench::compress::Uncompressed;
 use vortex_bench::conversions::parquet_to_vortex_chunks_with_batch_size;
-use vortex_bench::retain_edition_encodings;
 use vortex_cuda::CanonicalCudaExt;
 use vortex_cuda::CudaExecutionCtx;
 use vortex_cuda::CudaOpenOptionsExt;
@@ -46,6 +45,7 @@ use vortex_cuda::CudaSession;
 use vortex_cuda::PooledFileReadAtOptions;
 use vortex_cuda::executor::CudaArrayExt;
 use vortex_cuda::layout::CudaFlatLayoutStrategy;
+use vortex_cuda::layout::cuda_compatible_schemes;
 use vortex_cuda::layout::register_cuda_layout;
 
 use crate::gpu::writer::GPU_ROW_GROUP_SIZE;
@@ -100,11 +100,7 @@ impl Compressor for GpuVortexCompressor {
         // partition rather than whatever the default strategy would regroup them into.
         let strategy = Arc::new(ChunkedLayoutStrategy::new(CompressingStrategy::new(
             CudaFlatLayoutStrategy::default(),
-            retain_edition_encodings(
-                &SESSION,
-                BtrBlocksCompressorBuilder::default().only_cuda_compatible(),
-            )
-            .build(),
+            BtrBlocksCompressor::new(cuda_compatible_schemes(&SESSION)),
         )));
         let start = Instant::now();
         SESSION

benchmarks/string-bench/src/serialized.rs

@@ -31,7 +31,8 @@ use vortex::array::IntoArray;
 use vortex::array::VortexSessionExecute;
 use vortex::array::arrays::ChunkedArray;
 use vortex::array::arrays::VarBinViewArray;
-use vortex::compressor::BtrBlocksCompressorBuilder;
+use vortex::compressor::CompressionSessionExt;
+use vortex::compressor::Scheme;
 use vortex::file::OpenOptionsSessionExt;
 use vortex::file::WriteOptionsSessionExt;
 use vortex::file::WriteStrategyBuilder;
@@ -57,7 +58,7 @@ use crate::prepare_column;
 use crate::throughput;
 use crate::verify_canonicalized;
 
-/// The btrblocks string schemes that `BtrBlocksCompressorBuilder::default()` can
+/// The btrblocks string schemes that `BtrBlocksCompressor::from_session` can
 /// choose between. Forcing one encoder excludes every entry except its own
 /// scheme, so this list must track the default scheme set: add a row whenever a
 /// new string encoder becomes selectable by default (e.g. Zstd).
@@ -170,16 +171,23 @@ impl SerializedResult {
 
 /// Build the file writer strategy that forces one selected string scheme while
 /// leaving editioned non-string child compression enabled.
-fn serialized_write_strategy(encoder: StringEncoder) -> Arc<dyn LayoutStrategy> {
+fn serialized_write_strategy(
+    session: &VortexSession,
+    encoder: StringEncoder,
+) -> Arc<dyn LayoutStrategy> {
     let forced = encoder.scheme_id();
-    let compressor = BtrBlocksCompressorBuilder::default().exclude_schemes(
-        default_string_scheme_ids()
-            .into_iter()
-            .filter(|&id| id != forced)
-            .chain([DeltaScheme::default().id()]),
-    );
-    WriteStrategyBuilder::default()
-        .with_btrblocks_builder(compressor)
+    let excluded: Vec<SchemeId> = default_string_scheme_ids()
+        .into_iter()
+        .filter(|&id| id != forced)
+        .chain([DeltaScheme::default().id()])
+        .collect();
+    let schemes: Vec<&'static dyn Scheme> = session
+        .permitted_schemes()
+        .into_iter()
+        .filter(|scheme| !excluded.contains(&scheme.id()))
+        .collect();
+    WriteStrategyBuilder::from_session(session)
+        .with_schemes(schemes)
         .build()
 }
 
@@ -251,7 +259,7 @@ async fn prepare_serialized_file(
     verify: bool,
     ctx: &mut ExecutionCtx,
 ) -> Result<SerializedFile> {
-    let strategy = serialized_write_strategy(encoder);
+    let strategy = serialized_write_strategy(session, encoder);
     let data = write_serialized_file(session, input, &strategy).await?;
     let file_bytes = data.len() as u64;
 
@@ -355,7 +363,7 @@ mod tests {
     use vortex::io::runtime::BlockingRuntime;
     use vortex::io::runtime::current::CurrentThreadRuntime;
     use vortex::io::session::RuntimeSessionExt;
-    use vortex_btrblocks::ALL_SCHEMES;
+    use vortex_btrblocks::DEFAULT_SCHEMES;
     use vortex_btrblocks::SchemeExt;
 
     use super::*;
@@ -365,7 +373,7 @@ mod tests {
         // Every default scheme whose dtype gate accepts canonical Utf8 must be
         // excluded when another root string encoding is forced.
         let canonical = Canonical::VarBinView(VarBinViewArray::from_iter_str(["value"]));
-        let mut actual = ALL_SCHEMES
+        let mut actual = DEFAULT_SCHEMES
             .iter()
             .filter(|scheme| scheme.matches(&canonical))
             .map(|scheme| scheme.id())

encodings/parquet-variant/src/vtable.rs

@@ -466,8 +466,11 @@ mod tests {
     }
 
     #[fixture]
-    fn write_strategy() -> Arc<dyn LayoutStrategy> {
-        vortex_file::WriteStrategyBuilder::default().build()
+    fn write_strategy(
+        parquet_variant_file_session: VortexResult<VortexSession>,
+    ) -> VortexResult<Arc<dyn LayoutStrategy>> {
+        let session = parquet_variant_file_session?;
+        Ok(vortex_file::WriteStrategyBuilder::from_session(&session).build())
     }
 
     #[test]
@@ -544,15 +547,15 @@ mod tests {
     async fn test_file_roundtrip_typed_value_variant_with_zoned_strategy(
         #[from(typed_value_variant_array)] expected: VortexResult<ArrayRef>,
         parquet_variant_file_session: VortexResult<VortexSession>,
-        write_strategy: Arc<dyn LayoutStrategy>,
+        write_strategy: VortexResult<Arc<dyn LayoutStrategy>>,
     ) -> VortexResult<()> {
         let expected = expected?;
         let parquet_variant_file_session = parquet_variant_file_session?;
 
         let mut bytes = ByteBufferMut::empty();
         parquet_variant_file_session
             .write_options()
-            .with_strategy(write_strategy)
+            .with_strategy(write_strategy?)
             .write(&mut bytes, expected.to_array_stream())
             .await?;
 

fuzz/fuzz_targets/file_io.rs

@@ -18,7 +18,8 @@ use vortex_array::dtype::StructFields;
 use vortex_array::expr::lit;
 use vortex_array::expr::root;
 use vortex_array::scalar_fn::fns::operators::Operator;
-use vortex_btrblocks::BtrBlocksCompressorBuilder;
+use vortex_btrblocks::COMPACT_SCHEMES;
+use vortex_btrblocks::CompressionSessionExt;
 use vortex_error::VortexExpect;
 use vortex_error::vortex_panic;
 use vortex_file::OpenOptionsSessionExt;
@@ -65,8 +66,16 @@ fuzz_target!(|fuzz: FuzzFileAction| -> Corpus {
     let write_options = match compressor_strategy {
         CompressorStrategy::Default => SESSION.write_options(),
         CompressorStrategy::Compact => SESSION.write_options().with_strategy(
-            WriteStrategyBuilder::default()
-                .with_btrblocks_builder(BtrBlocksCompressorBuilder::default().with_compact())
+            WriteStrategyBuilder::from_session(&SESSION)
+                .with_schemes(
+                    SESSION.permit(
+                        SESSION
+                            .registered_schemes()
+                            .into_iter()
+                            .chain(COMPACT_SCHEMES.iter().copied())
+                            .collect(),
+                    ),
+                )
                 .build(),
         ),
     };

fuzz/src/array/mod.rs

@@ -66,7 +66,9 @@ use vortex_array::search_sorted::SearchSorted;
 use vortex_array::search_sorted::SearchSortedSide;
 use vortex_btrblocks::BtrBlocksCompressor;
 #[cfg(feature = "zstd")]
-use vortex_btrblocks::BtrBlocksCompressorBuilder;
+use vortex_btrblocks::COMPACT_SCHEMES;
+#[cfg(feature = "zstd")]
+use vortex_btrblocks::DEFAULT_SCHEMES;
 use vortex_error::VortexExpect;
 use vortex_error::vortex_panic;
 use vortex_mask::Mask;
@@ -249,7 +251,7 @@ impl<'a> Arbitrary<'a> for FuzzArrayAction {
                         .into_array()
                     };
 
-                    let compressed = BtrBlocksCompressor::default()
+                    let compressed = BtrBlocksCompressor::from_session(&SESSION)
                         .compress(&indices_array, &mut ctx)
                         .vortex_expect("BtrBlocksCompressor compress should succeed in fuzz test");
                     (
@@ -561,14 +563,18 @@ pub fn compress_array(
     ctx: &mut ExecutionCtx,
 ) -> ArrayRef {
     match strategy {
-        CompressorStrategy::Default => BtrBlocksCompressor::default()
+        CompressorStrategy::Default => BtrBlocksCompressor::from_session(&SESSION)
             .compress(array, ctx)
             .vortex_expect("BtrBlocksCompressor compress should succeed in fuzz test"),
-        CompressorStrategy::Compact => BtrBlocksCompressorBuilder::default()
-            .with_compact()
-            .build()
-            .compress(array, ctx)
-            .vortex_expect("Compact compress should succeed in fuzz test"),
+        CompressorStrategy::Compact => BtrBlocksCompressor::new(
+            DEFAULT_SCHEMES
+                .iter()
+                .copied()
+                .chain(COMPACT_SCHEMES.iter().copied())
+                .collect(),
+        )
+        .compress(array, ctx)
+        .vortex_expect("Compact compress should succeed in fuzz test"),
     }
 }
 
@@ -579,7 +585,7 @@ pub fn compress_array(
     _strategy: CompressorStrategy,
     ctx: &mut ExecutionCtx,
 ) -> ArrayRef {
-    BtrBlocksCompressor::default()
+    BtrBlocksCompressor::from_session(&SESSION)
         .compress(array, ctx)
         .vortex_expect("BtrBlocksCompressor compress should succeed in fuzz test")
 }

vortex-bench/src/conversions.rs

@@ -37,7 +37,7 @@ use vortex::array::arrays::struct_::StructArrayExt;
 use vortex::array::builders::builder_with_capacity_in;
 use vortex::array::stream::ArrayStreamAdapter;
 use vortex::array::stream::ArrayStreamExt;
-use vortex::compressor::BtrBlocksCompressorBuilder;
+use vortex::compressor::BtrBlocksCompressor;
 use vortex::dtype::DType;
 use vortex::dtype::FieldPath;
 use vortex::dtype::StructFields;
@@ -66,7 +66,7 @@ use wkb::writer::write_geometry;
 use crate::CompactionStrategy;
 use crate::Format;
 use crate::SESSION;
-use crate::retain_edition_encodings;
+use crate::compact_schemes;
 use crate::utils::file::idempotent_async;
 
 /// Memory budget per concurrent conversion stream in GB. This is somewhat arbitary.
@@ -246,12 +246,9 @@ fn write_options_for(
         return compaction.apply_options(SESSION.write_options());
     }
 
-    let mut builder = WriteStrategyBuilder::default();
+    let mut builder = WriteStrategyBuilder::from_session(&SESSION);
     if matches!(compaction, CompactionStrategy::Compact) {
-        builder = builder.with_btrblocks_builder(retain_edition_encodings(
-            &SESSION,
-            BtrBlocksCompressorBuilder::default().with_compact(),
-        ));
+        builder = builder.with_schemes(compact_schemes());
     }
     for name in binary_fields {
         builder = builder.with_field_writer(FieldPath::from_name(name), no_dict_layout());
@@ -263,7 +260,7 @@ fn write_options_for(
 fn no_dict_layout() -> Arc<dyn LayoutStrategy> {
     Arc::new(CompressingStrategy::new(
         ChunkedLayoutStrategy::new(FlatLayoutStrategy::default()),
-        retain_edition_encodings(&SESSION, BtrBlocksCompressorBuilder::default()).build(),
+        BtrBlocksCompressor::from_session(&SESSION),
     ))
 }
 

vortex-bench/src/lib.rs

@@ -28,7 +28,9 @@ use tpcds::TpcDsBenchmark;
 use tpch::benchmark::TpcHBenchmark;
 pub use utils::file::*;
 pub use utils::logging::*;
-use vortex::compressor::BtrBlocksCompressorBuilder;
+use vortex::compressor::COMPACT_SCHEMES;
+use vortex::compressor::CompressionSessionExt;
+use vortex::compressor::Scheme;
 use vortex::error::VortexExpect;
 use vortex::error::vortex_err;
 use vortex::file::VortexWriteOptions;
@@ -70,8 +72,6 @@ pub use datasets::BenchmarkDataset;
 pub use output::BenchmarkOutput;
 pub use output::create_output_writer;
 use vortex::VortexSessionDefault;
-use vortex::editions::ComponentKind;
-use vortex::editions::EditionSessionExt;
 pub use vortex::error::vortex_panic;
 use vortex::io::session::RuntimeSessionExt;
 use vortex::session::VortexSession;
@@ -254,31 +254,24 @@ impl CompactionStrategy {
     pub fn apply_options(&self, options: VortexWriteOptions) -> VortexWriteOptions {
         match self {
             CompactionStrategy::Compact => options.with_strategy(
-                WriteStrategyBuilder::default()
-                    .with_btrblocks_builder(retain_edition_encodings(
-                        &SESSION,
-                        BtrBlocksCompressorBuilder::default().with_compact(),
-                    ))
+                WriteStrategyBuilder::from_session(&SESSION)
+                    .with_schemes(compact_schemes())
                     .build(),
             ),
             CompactionStrategy::Default => options,
         }
     }
 }
 
-/// Restrict `builder` to the encodings permitted by the session's enabled editions.
-///
-/// The default writer applies this filter itself. An explicit strategy bypasses it, so a
-/// benchmark that builds its own compressor applies it here to stay within editions.
-pub fn retain_edition_encodings(
-    session: &VortexSession,
-    builder: BtrBlocksCompressorBuilder,
-) -> BtrBlocksCompressorBuilder {
-    let allowed = session
-        .enabled_component_ids(ComponentKind::Array)
-        .into_iter()
-        .collect();
-    builder.retain_allowed_encodings(&allowed)
+/// The schemes [`SESSION`] permits plus the compact ones, for [`CompactionStrategy::Compact`].
+pub fn compact_schemes() -> Vec<&'static dyn Scheme> {
+    SESSION.permit(
+        SESSION
+            .registered_schemes()
+            .into_iter()
+            .chain(COMPACT_SCHEMES.iter().copied())
+            .collect(),
+    )
 }
 
 /// Verify that local data has already been prepared for the requested benchmark formats.

vortex-btrblocks/Cargo.toml

@@ -33,6 +33,7 @@ vortex-onpair = { workspace = true }
 vortex-pco = { workspace = true, optional = true }
 vortex-runend = { workspace = true }
 vortex-sequence = { workspace = true }
+vortex-session = { workspace = true }
 vortex-sparse = { workspace = true }
 vortex-utils = { workspace = true }
 vortex-zigzag = { workspace = true }
@@ -51,7 +52,6 @@ vortex-array = { workspace = true, features = ["_test-harness"] }
 vortex-arrow = { workspace = true }
 vortex-edition = { workspace = true }
 vortex-mask = { workspace = true }
-vortex-session = { workspace = true }
 
 [features]
 pco = ["dep:pco", "dep:vortex-pco"]

vortex-btrblocks/benches/compress.rs

@@ -20,6 +20,7 @@ mod benchmarks {
     use vortex_array::VortexSessionExecute;
     use vortex_array::arrays::PrimitiveArray;
     use vortex_btrblocks::BtrBlocksCompressor;
+    use vortex_btrblocks::DEFAULT_SCHEMES;
     use vortex_buffer::buffer_mut;
     use vortex_session::VortexSession;
     use vortex_utils::aliases::hash_set::HashSet;
@@ -51,7 +52,7 @@ mod benchmarks {
         let array = make_clickbench_window_name()
             .execute::<PrimitiveArray>(&mut ctx)
             .unwrap();
-        let compressor = BtrBlocksCompressor::default();
+        let compressor = BtrBlocksCompressor::new(DEFAULT_SCHEMES.to_vec());
         bencher
             .with_inputs(|| (&array, SESSION.create_execution_ctx()))
             .input_counter(|(array, _)| ItemsCount::new(array.len()))