From d3526b574857866c67a95d0c49c31512390bf6f1 Mon Sep 17 00:00:00 2001 From: Matt Katz Date: Thu, 17 Sep 2026 10:45:40 -0400 Subject: [PATCH] Carry the writer's permitted serialized IDs into compression schemes Signed-off-by: Matt Katz --- vortex-btrblocks/src/builder.rs | 62 +++++++-- vortex-btrblocks/tests/scheme_modes.rs | 131 ++++++++++++++++++ vortex-compressor/src/compressor/cascade.rs | 2 +- .../src/compressor/edition_tests.rs | 97 +++++++++++++ vortex-compressor/src/compressor/mod.rs | 42 ++++++ vortex-compressor/src/scheme/allowed.rs | 51 +++++++ vortex-compressor/src/scheme/ctx.rs | 29 ++++ vortex-compressor/src/scheme/mod.rs | 10 +- vortex-file/src/tests.rs | 35 +++++ 9 files changed, 446 insertions(+), 13 deletions(-) create mode 100644 vortex-btrblocks/tests/scheme_modes.rs create mode 100644 vortex-compressor/src/compressor/edition_tests.rs create mode 100644 vortex-compressor/src/scheme/allowed.rs diff --git a/vortex-btrblocks/src/builder.rs b/vortex-btrblocks/src/builder.rs index 341569f408c..7d22b01f05d 100644 --- a/vortex-btrblocks/src/builder.rs +++ b/vortex-btrblocks/src/builder.rs @@ -4,6 +4,7 @@ //! Builder for configuring `BtrBlocksCompressor` instances. use vortex_array::ArrayId; +use vortex_compressor::scheme::AllowedSerializedIds; use vortex_utils::aliases::hash_set::HashSet; use crate::BtrBlocksCompressor; @@ -90,12 +91,15 @@ pub const ALL_SCHEMES: &[&dyn Scheme] = &[ #[derive(Debug, Clone)] pub struct BtrBlocksCompressorBuilder { schemes: Vec<&'static dyn Scheme>, + /// The serialized IDs the compressor may write under. + allowed_serialized_ids: AllowedSerializedIds, } impl Default for BtrBlocksCompressorBuilder { fn default() -> Self { Self { schemes: ALL_SCHEMES.to_vec(), + allowed_serialized_ids: AllowedSerializedIds::All, } } } @@ -107,6 +111,7 @@ impl BtrBlocksCompressorBuilder { pub fn empty() -> Self { Self { schemes: Vec::new(), + allowed_serialized_ids: AllowedSerializedIds::All, } } @@ -202,23 +207,29 @@ impl BtrBlocksCompressorBuilder { /// Retains only schemes whose produced serialized IDs all belong to `allowed`. /// - /// `allowed` holds serialized IDs. The file writer passes the array IDs its enabled editions - /// permit. + /// The set is also handed to the compressor, intersected with any earlier call, so a scheme + /// with several wire formats writes a newer one only when permitted. The file writer passes + /// the array IDs its enabled editions permit. pub fn retain_allowed_encodings(mut self, allowed: &HashSet) -> Self { self.schemes .retain(|s| s.produced_encodings().iter().all(|id| allowed.contains(id))); + self.allowed_serialized_ids.restrict(allowed); self } /// Builds the configured [`BtrBlocksCompressor`]. pub fn build(self) -> BtrBlocksCompressor { - BtrBlocksCompressor(CascadingCompressor::new(self.schemes)) + BtrBlocksCompressor( + CascadingCompressor::new(self.schemes) + .with_allowed_serialized_ids(&self.allowed_serialized_ids), + ) } } #[cfg(test)] mod tests { use vortex_array::VTable; + use vortex_fastlanes::BitPacked; use vortex_fastlanes::FoR; use super::*; @@ -238,12 +249,20 @@ mod tests { #[test] fn retain_allowed_encodings_filters_schemes() { let allowed: HashSet = [FoR.id()].into_iter().collect(); - let builder = BtrBlocksCompressorBuilder::default().retain_allowed_encodings(&allowed); - assert_eq!(builder.schemes.len(), 1); - assert_eq!(builder.schemes[0].id(), integer::FoRScheme.id()); + let compressor = BtrBlocksCompressorBuilder::default() + .retain_allowed_encodings(&allowed) + .build(); + assert_eq!(compressor.schemes().len(), 1); + assert_eq!(compressor.schemes()[0].id(), integer::FoRScheme.id()); + assert_eq!( + compressor.allowed_serialized_ids(), + &AllowedSerializedIds::Only(allowed) + ); - let none = BtrBlocksCompressorBuilder::default().retain_allowed_encodings(&HashSet::new()); - assert!(none.schemes.is_empty()); + let none = BtrBlocksCompressorBuilder::default() + .retain_allowed_encodings(&HashSet::new()) + .build(); + assert!(none.schemes().is_empty()); } #[test] @@ -252,8 +271,31 @@ mod tests { .iter() .flat_map(|scheme| scheme.produced_encodings()) .collect(); - let builder = BtrBlocksCompressorBuilder::default().retain_allowed_encodings(&allowed); - assert_eq!(builder.schemes.len(), ALL_SCHEMES.len()); + let compressor = BtrBlocksCompressorBuilder::default() + .retain_allowed_encodings(&allowed) + .build(); + assert_eq!(compressor.schemes().len(), ALL_SCHEMES.len()); + } + + #[test] + fn unrestricted_builds_permit_everything() { + let compressor = BtrBlocksCompressorBuilder::default().build(); + assert_eq!( + compressor.allowed_serialized_ids(), + &AllowedSerializedIds::All + ); + } + + #[test] + fn repeated_restrictions_intersect() { + let first: HashSet = [FoR.id(), BitPacked.id()].into_iter().collect(); + let second: HashSet = [BitPacked.id()].into_iter().collect(); + let compressor = BtrBlocksCompressorBuilder::default() + .retain_allowed_encodings(&first) + .retain_allowed_encodings(&second) + .build(); + assert!(!compressor.has_scheme(integer::FoRScheme.id())); + assert!(compressor.has_scheme(integer::BitPackingScheme.id())); } #[test] diff --git a/vortex-btrblocks/tests/scheme_modes.rs b/vortex-btrblocks/tests/scheme_modes.rs new file mode 100644 index 00000000000..8ee02ff9913 --- /dev/null +++ b/vortex-btrblocks/tests/scheme_modes.rs @@ -0,0 +1,131 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +#[cfg(test)] +mod tests { + use rstest::rstest; + use vortex_array::ArrayId; + use vortex_array::ArrayRef; + use vortex_array::Canonical; + use vortex_array::ExecutionCtx; + use vortex_array::IntoArray; + use vortex_array::VTable; + use vortex_array::VortexSessionExecute; + use vortex_array::array_session; + use vortex_array::arrays::PrimitiveArray; + use vortex_array::assert_arrays_eq; + use vortex_btrblocks::ArrayAndStats; + use vortex_btrblocks::BtrBlocksCompressorBuilder; + use vortex_btrblocks::CascadingCompressor; + use vortex_btrblocks::CompressorContext; + use vortex_btrblocks::Scheme; + use vortex_btrblocks::SchemeExt; + use vortex_btrblocks::schemes::integer::BitPackingScheme; + use vortex_compressor::scheme::CompressionEstimate; + use vortex_compressor::scheme::EstimateVerdict; + use vortex_error::VortexResult; + use vortex_fastlanes::BitPacked; + use vortex_fastlanes::Delta; + use vortex_session::registry::CachedId; + + static NEWER_ID: CachedId = CachedId::new("test.delta_newer"); + + /// A scheme with two wire formats. It always writes `Delta`, and when the writer permits the + /// newer format it takes a different branch. The newer branch stands in for a format this test + /// cannot serialize, so it returns the input unchanged and the compressor falls back to + /// canonical output. + #[derive(Debug)] + struct TwoFormatDelta; + + impl Scheme for TwoFormatDelta { + fn scheme_name(&self) -> &'static str { + "test.two_format_delta" + } + + fn matches(&self, canonical: &Canonical) -> bool { + canonical.dtype().is_int() + } + + fn produced_encodings(&self) -> Vec { + vec![Delta.id()] + } + + /// Children: bases=0, deltas=1. + fn num_children(&self) -> usize { + 2 + } + + fn expected_compression_ratio( + &self, + _data: &ArrayAndStats, + _compress_ctx: CompressorContext, + _exec_ctx: &mut ExecutionCtx, + ) -> CompressionEstimate { + CompressionEstimate::Verdict(EstimateVerdict::AlwaysUse) + } + + fn compress( + &self, + compressor: &CascadingCompressor, + data: &ArrayAndStats, + compress_ctx: CompressorContext, + exec_ctx: &mut ExecutionCtx, + ) -> VortexResult { + if compress_ctx.allows_serialized_id(&NEWER_ID) { + return Ok(data.array().clone()); + } + let primitive = data.array().clone().execute::(exec_ctx)?; + let (bases, deltas) = vortex_fastlanes::delta_compress(&primitive, exec_ctx)?; + let bases = compressor.compress_child( + &bases.into_array(), + &compress_ctx, + self.id(), + 0, + exec_ctx, + )?; + let deltas = compressor.compress_child( + &deltas.into_array(), + &compress_ctx, + self.id(), + 1, + exec_ctx, + )?; + Delta::try_new(bases, deltas, 0, primitive.len()).map(IntoArray::into_array) + } + } + + fn compressor(allowed: &[ArrayId]) -> vortex_btrblocks::BtrBlocksCompressor { + BtrBlocksCompressorBuilder::empty() + .with_new_scheme(&TwoFormatDelta) + .with_new_scheme(&BitPackingScheme) + .retain_allowed_encodings(&allowed.iter().copied().collect()) + .build() + } + + /// The scheme reads the writer's permitted IDs from its context and picks its format. + #[rstest] + #[case::frozen_only(vec![Delta.id(), BitPacked.id()], true)] + #[case::newer_permitted(vec![Delta.id(), BitPacked.id(), *NEWER_ID], false)] + fn a_scheme_picks_its_format_from_the_permitted_ids( + #[case] allowed: Vec, + #[case] expect_delta: bool, + ) -> VortexResult<()> { + let session = array_session(); + vortex_fastlanes::initialize(&session); + let compressor = compressor(&allowed); + let array = PrimitiveArray::from_iter((0..65_536u32).map(|i| i / 3)).into_array(); + let mut ctx = session.create_execution_ctx(); + let compressed = compressor.compress(&array, &mut ctx)?; + assert_eq!(compressed.encoding_id() == Delta.id(), expect_delta); + assert_arrays_eq!(compressed, array, &mut ctx); + Ok(()) + } + + /// The declared format is required. Permitting only the newer one leaves the scheme out. + #[test] + fn the_declared_format_must_be_permitted() { + let compressor = compressor(&[*NEWER_ID, BitPacked.id()]); + assert!(!compressor.has_scheme(TwoFormatDelta.id())); + assert!(compressor.has_scheme(BitPackingScheme.id())); + } +} diff --git a/vortex-compressor/src/compressor/cascade.rs b/vortex-compressor/src/compressor/cascade.rs index 86d45d2c0d9..dd98f4ea3c6 100644 --- a/vortex-compressor/src/compressor/cascade.rs +++ b/vortex-compressor/src/compressor/cascade.rs @@ -59,7 +59,7 @@ impl CascadingCompressor { let canonical = array.clone().execute::(exec_ctx)?.0; let compact = canonical.compact(exec_ctx)?; - let compressed = self.compress_canonical(compact, CompressorContext::new(), exec_ctx)?; + let compressed = self.compress_canonical(compact, self.root_context(), exec_ctx)?; trace::record_compress_outcome(&span, before_nbytes, compressed.nbytes()); diff --git a/vortex-compressor/src/compressor/edition_tests.rs b/vortex-compressor/src/compressor/edition_tests.rs new file mode 100644 index 00000000000..3fa5153fc12 --- /dev/null +++ b/vortex-compressor/src/compressor/edition_tests.rs @@ -0,0 +1,97 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +use vortex_array::ArrayId; +use vortex_array::ArrayRef; +use vortex_array::Canonical; +use vortex_array::ExecutionCtx; +use vortex_error::VortexResult; +use vortex_session::registry::CachedId; + +use super::*; +use crate::scheme::CompressionEstimate; +use crate::scheme::EstimateVerdict; +use crate::stats::ArrayAndStats; + +static V1_ID: CachedId = CachedId::new("test.format_v1"); +static V2_ID: CachedId = CachedId::new("test.format_v2"); + +/// A scheme that always writes `test.format_v1` and, when permitted, also `test.format_v2`. +#[derive(Debug)] +struct ModeScheme; + +impl Scheme for ModeScheme { + fn scheme_name(&self) -> &'static str { + "test.mode" + } + + fn matches(&self, canonical: &Canonical) -> bool { + canonical.dtype().is_int() + } + + fn produced_encodings(&self) -> Vec { + vec![*V1_ID] + } + + fn expected_compression_ratio( + &self, + _data: &ArrayAndStats, + _compress_ctx: CompressorContext, + _exec_ctx: &mut ExecutionCtx, + ) -> CompressionEstimate { + CompressionEstimate::Verdict(EstimateVerdict::Skip) + } + + fn compress( + &self, + _compressor: &CascadingCompressor, + data: &ArrayAndStats, + _compress_ctx: CompressorContext, + _exec_ctx: &mut ExecutionCtx, + ) -> VortexResult { + Ok(data.array().clone()) + } +} + +fn only(ids: &[&CachedId]) -> AllowedSerializedIds { + AllowedSerializedIds::Only(ids.iter().map(|id| ***id).collect()) +} + +#[test] +fn unrestricted_contexts_permit_everything() { + let compressor = CascadingCompressor::new(vec![&ModeScheme]); + assert_eq!( + compressor.allowed_serialized_ids(), + &AllowedSerializedIds::All + ); + let ctx = compressor.root_context(); + assert!(ctx.allows_serialized_id(&V1_ID)); + assert!(ctx.allows_serialized_id(&V2_ID)); +} + +#[test] +fn the_permitted_set_reaches_descendant_contexts() { + let compressor = + CascadingCompressor::new(vec![&ModeScheme]).with_allowed_serialized_ids(&only(&[&V1_ID])); + let root = compressor.root_context(); + assert!(root.allows_serialized_id(&V1_ID)); + assert!(!root.allows_serialized_id(&V2_ID)); + + let child = root.descend_with_scheme(ModeScheme.id(), 0); + assert!(child.allows_serialized_id(&V1_ID)); + assert!(!child.allows_serialized_id(&V2_ID)); + assert_eq!(child.allowed_serialized_ids(), &only(&[&V1_ID])); +} + +#[test] +fn repeated_restrictions_intersect() { + let compressor = CascadingCompressor::new(vec![&ModeScheme]) + .with_allowed_serialized_ids(&only(&[&V1_ID, &V2_ID])) + .with_allowed_serialized_ids(&only(&[&V1_ID])); + assert_eq!(compressor.allowed_serialized_ids(), &only(&[&V1_ID])); + assert!(!compressor.root_context().allows_serialized_id(&V2_ID)); + + // Intersecting with `All` changes nothing. + let compressor = compressor.with_allowed_serialized_ids(&AllowedSerializedIds::All); + assert_eq!(compressor.allowed_serialized_ids(), &only(&[&V1_ID])); +} diff --git a/vortex-compressor/src/compressor/mod.rs b/vortex-compressor/src/compressor/mod.rs index 219b67e2519..05bda024417 100644 --- a/vortex-compressor/src/compressor/mod.rs +++ b/vortex-compressor/src/compressor/mod.rs @@ -9,8 +9,12 @@ mod sample; mod select; mod structural; +use std::sync::Arc; + use crate::builtins::IntDictScheme; +use crate::scheme::AllowedSerializedIds; use crate::scheme::ChildSelection; +use crate::scheme::CompressorContext; use crate::scheme::DescendantExclusion; use crate::scheme::Scheme; use crate::scheme::SchemeExt; @@ -46,6 +50,9 @@ pub struct CascadingCompressor { /// Descendant exclusion rules for the compressor's own cascading (e.g. excluding Dict from /// list offsets). root_exclusions: Vec, + + /// The serialized IDs the writer may emit, handed to every [`CompressorContext`]. + allowed_serialized_ids: Arc, } impl CascadingCompressor { @@ -63,9 +70,41 @@ impl CascadingCompressor { Self { schemes, root_exclusions, + allowed_serialized_ids: Arc::new(AllowedSerializedIds::All), } } + /// Hands the compressor the serialized IDs the writer may emit, intersecting with any + /// earlier call. + /// + /// The set reaches every scheme through [`CompressorContext::allows_serialized_id`], so a + /// scheme with several wire formats writes a newer one only when permitted. Callers filter + /// the scheme list themselves: every scheme given to [`new`](Self::new) should have all of its + /// [`produced_encodings`](Scheme::produced_encodings) permitted, as + /// `BtrBlocksCompressorBuilder::retain_allowed_encodings` ensures. + pub fn with_allowed_serialized_ids(mut self, allowed: &AllowedSerializedIds) -> Self { + let mut merged = (*self.allowed_serialized_ids).clone(); + merged.intersect(allowed); + self.allowed_serialized_ids = Arc::new(merged); + self + } + + /// The serialized IDs the writer may emit. + pub fn allowed_serialized_ids(&self) -> &AllowedSerializedIds { + &self.allowed_serialized_ids + } + + /// The compression schemes, in registration order. + pub fn schemes(&self) -> &[&'static dyn Scheme] { + &self.schemes + } + + /// The context a compress call starts from, carrying the permitted serialized IDs. + pub(crate) fn root_context(&self) -> CompressorContext { + CompressorContext::new() + .with_allowed_serialized_ids(Arc::clone(&self.allowed_serialized_ids)) + } + /// Returns whether the compressor was configured with `scheme`. pub fn has_scheme(&self, scheme: SchemeId) -> bool { self.schemes @@ -78,3 +117,6 @@ impl CascadingCompressor { #[cfg(test)] mod tests; + +#[cfg(test)] +mod edition_tests; diff --git a/vortex-compressor/src/scheme/allowed.rs b/vortex-compressor/src/scheme/allowed.rs new file mode 100644 index 00000000000..a71501b6061 --- /dev/null +++ b/vortex-compressor/src/scheme/allowed.rs @@ -0,0 +1,51 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! The serialized IDs a writer permits a compressor to emit. + +use vortex_array::ArrayId; +use vortex_utils::aliases::hash_set::HashSet; + +/// The serialized IDs a compressor may write under. +/// +/// The file writer derives this from its enabled editions. The compressor drops schemes whose +/// [`produced_encodings`](crate::scheme::Scheme::produced_encodings) it does not permit and hands +/// it to the rest through [`CompressorContext`](crate::scheme::CompressorContext). +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub enum AllowedSerializedIds { + /// Unrestricted. Every scheme resolves to its newest version. + #[default] + All, + /// Only these IDs. + Only(HashSet), +} + +impl AllowedSerializedIds { + /// Whether the writer may emit `id`. + pub fn permits(&self, id: &ArrayId) -> bool { + match self { + Self::All => true, + Self::Only(ids) => ids.contains(id), + } + } + + /// Whether the writer may emit every one of `ids`. + pub fn permits_all(&self, ids: &[ArrayId]) -> bool { + ids.iter().all(|id| self.permits(id)) + } + + /// Narrows the permitted IDs to those also permitted by `other`. + pub fn intersect(&mut self, other: &Self) { + if let Self::Only(ids) = other { + self.restrict(ids); + } + } + + /// Narrows the permitted IDs to those also in `allowed`. + pub fn restrict(&mut self, allowed: &HashSet) { + *self = match self { + Self::All => Self::Only(allowed.clone()), + Self::Only(existing) => Self::Only(existing.intersection(allowed).copied().collect()), + }; + } +} diff --git a/vortex-compressor/src/scheme/ctx.rs b/vortex-compressor/src/scheme/ctx.rs index 4eed7538daa..abd43300b97 100644 --- a/vortex-compressor/src/scheme/ctx.rs +++ b/vortex-compressor/src/scheme/ctx.rs @@ -4,10 +4,13 @@ //! Compression context for recursive compression. use std::fmt; +use std::sync::Arc; +use vortex_array::ArrayId; use vortex_error::VortexExpect; use crate::compressor::ROOT_SCHEME_ID; +use crate::scheme::AllowedSerializedIds; use crate::scheme::SchemeId; use crate::stats::GenerateStatsOptions; @@ -38,6 +41,9 @@ pub struct CompressorContext { /// [`descendant_exclusions`]: crate::scheme::Scheme::descendant_exclusions /// [`ancestor_exclusions`]: crate::scheme::Scheme::ancestor_exclusions cascade_history: Vec<(SchemeId, usize)>, + + /// The serialized IDs the writer may emit, shared by every context in a cascade. + allowed_serialized_ids: Arc, } impl CompressorContext { @@ -50,8 +56,18 @@ impl CompressorContext { allowed_cascading: MAX_CASCADE, merged_stats_options: GenerateStatsOptions::default(), cascade_history: Vec::new(), + allowed_serialized_ids: Arc::new(AllowedSerializedIds::All), } } + + /// Returns a context that carries the writer's permitted serialized IDs. + pub(crate) fn with_allowed_serialized_ids( + mut self, + allowed: Arc, + ) -> Self { + self.allowed_serialized_ids = allowed; + self + } } #[cfg(test)] @@ -62,6 +78,19 @@ impl Default for CompressorContext { } impl CompressorContext { + /// Whether the writer may emit `id`. + /// + /// A scheme with several wire formats calls this before writing any format it does not + /// declare in [`produced_encodings`](crate::scheme::Scheme::produced_encodings). + pub fn allows_serialized_id(&self, id: &ArrayId) -> bool { + self.allowed_serialized_ids.permits(id) + } + + /// The serialized IDs the writer may emit. + pub fn allowed_serialized_ids(&self) -> &AllowedSerializedIds { + &self.allowed_serialized_ids + } + /// Whether this context is for sample compression (ratio estimation). pub fn is_sample(&self) -> bool { self.is_sample diff --git a/vortex-compressor/src/scheme/mod.rs b/vortex-compressor/src/scheme/mod.rs index 0ba1c90202a..13b02fc4c13 100644 --- a/vortex-compressor/src/scheme/mod.rs +++ b/vortex-compressor/src/scheme/mod.rs @@ -4,6 +4,8 @@ //! Everything a scheme author implements or receives: the [`Scheme`] trait, exclusion rules, //! compression estimates, and the compression context. +mod allowed; +pub use allowed::AllowedSerializedIds; mod ctx; pub use ctx::CompressorContext; pub use ctx::MAX_CASCADE; @@ -124,7 +126,7 @@ pub trait Scheme: Debug + Send + Sync { /// Whether this scheme can compress the given canonical array. fn matches(&self, canonical: &Canonical) -> bool; - /// The serialized IDs this scheme itself may write into its compressed output. + /// The serialized IDs this scheme always writes into its compressed output. /// /// Every declared ID must be permitted for the scheme to be used. Cascaded children are /// compressed by other schemes, which declare their own IDs, so only arrays constructed @@ -132,7 +134,11 @@ pub trait Scheme: Debug + Send + Sync { /// merely rearranges do not need to be declared. /// /// For most encodings this is the in-memory encoding ID. An encoding with several wire - /// formats declares the wire IDs the scheme writes, which may differ from its in-memory ID. + /// formats declares only the format it always writes, which may differ from its in-memory + /// ID. It writes a newer format only after [`CompressorContext::allows_serialized_id`] + /// permits it, in both [`expected_compression_ratio`](Scheme::expected_compression_ratio) + /// and [`compress`](Scheme::compress), so the same writer configuration always yields the + /// same output. fn produced_encodings(&self) -> Vec; /// Returns the stats generation options this scheme requires. The compressor merges all diff --git a/vortex-file/src/tests.rs b/vortex-file/src/tests.rs index 640de874d2b..f740aacb318 100644 --- a/vortex-file/src/tests.rs +++ b/vortex-file/src/tests.rs @@ -1729,6 +1729,41 @@ async fn test_encoding_registered_after_write_options() -> VortexResult<()> { Ok(()) } +#[rstest] +#[case::sparse(PrimitiveArray::from_iter( + (0..4096i32).map(|i| if i % 100 == 0 { i + 1 } else { 0 }), +).into_array())] +#[case::fsst(VarBinViewArray::from_iter( + (0..4096).map(|i| Some(format!("this_is_a_common_prefix_with_some_variation_{i}_and_a_common_suffix_pattern"))), + DType::Utf8(Nullability::NonNullable), +).into_array())] +#[tokio::test] +async fn test_writer_excludes_schemes_with_unavailable_outputs( + #[case] array: ArrayRef, +) -> VortexResult<()> { + let session = array_session() + .with::() + .with::() + .with::(); + // Permit Constant and VarBin, but not the subsequently registered Sparse and FSST. + crate::enable_all_registered_array_encodings(&session); + crate::register_default_encodings(&session); + let mut buf = ByteBufferMut::empty(); + session + .write_options() + .write(&mut buf, array.clone().to_array_stream()) + .await?; + let read = session + .open_options() + .open_buffer(buf)? + .scan()? + .into_array_stream()? + .read_all() + .await?; + assert_arrays_eq!(read, array, &mut session.create_execution_ctx()); + Ok(()) +} + #[tokio::test] async fn test_writer_empty_chunks() -> VortexResult<()> { let mut ctx = SESSION.create_execution_ctx();