From d455777b53ba945505c623b83909ac19fa20c317 Mon Sep 17 00:00:00 2001 From: ump45nose <52391318+ump45nose@users.noreply.github.com> Date: Wed, 11 Nov 2026 18:47:43 +0800 Subject: [PATCH 01/27] Return Serde errors for corrupt RLE and RunEnd metadata (#10852) ## Summary - Tracking Issue: #10756 Corrupt or crafted metadata can make a file read panic instead of returning a Serde error. #10760 covers the ALP-RD side of that issue. This PR covers the RLE and RunEnd paths called out there: - RLE `deserialize` called `dtype.as_ptype()`, which panics when the file dtype is not primitive. It now uses `PType::try_from(dtype)?`. - RunEnd `deserialize` used `vortex_expect` on `usize::try_from` for `num_runs` and `offset`, which panics on 32-bit targets when the stored value exceeds `usize::MAX`. Both conversions now return a Serde error. Partially addresses #10756. Signed-off-by: Vincent <1729065730@qq.com> --- encodings/fastlanes/src/rle/vtable/mod.rs | 31 ++++++++++ encodings/fastlanes/src/rle/vtable/plugin.rs | 3 +- vortex-array/src/arrays/runend/array/mod.rs | 60 +++++++++++++++++++ .../src/arrays/runend/array/plugin.rs | 15 ++++- 4 files changed, 106 insertions(+), 3 deletions(-) diff --git a/encodings/fastlanes/src/rle/vtable/mod.rs b/encodings/fastlanes/src/rle/vtable/mod.rs index ae139647c7d..e64206fb4eb 100644 --- a/encodings/fastlanes/src/rle/vtable/mod.rs +++ b/encodings/fastlanes/src/rle/vtable/mod.rs @@ -339,4 +339,35 @@ mod tests { Ok(()) } + + #[test] + fn test_deserialize_rejects_non_primitive_dtype() -> VortexResult<()> { + let metadata = RLEMetadata { + values_len: 0, + indices_len: 0, + indices_ptype: PType::U16 as i32, + values_idx_offsets_len: 0, + values_idx_offsets_ptype: PType::U64 as i32, + offset: 0, + } + .encode_to_vec(); + let children: Vec = Vec::new(); + let result = ArrayPlugin::deserialize( + &RLE, + ArrayDeserialization::new( + VTable::id(&RLE), + &DType::Utf8(Nullability::NonNullable), + 0, + &metadata, + &[], + &children, + ), + &SESSION, + ); + assert!(result.is_err_and(|err| { + err.to_string() + .contains("Cannot convert DType utf8 into PType") + })); + Ok(()) + } } diff --git a/encodings/fastlanes/src/rle/vtable/plugin.rs b/encodings/fastlanes/src/rle/vtable/plugin.rs index 6f78cb6bbb0..149d999fad9 100644 --- a/encodings/fastlanes/src/rle/vtable/plugin.rs +++ b/encodings/fastlanes/src/rle/vtable/plugin.rs @@ -86,9 +86,10 @@ impl ArrayPlugin for RLE { "RLEArray expects 0 buffers, got {}", buffers.len() ); + let values_ptype = PType::try_from(dtype)?; let values = children.get( 0, - &DType::Primitive(dtype.as_ptype(), Nullability::NonNullable), + &DType::Primitive(values_ptype, Nullability::NonNullable), usize::try_from(metadata.values_len)?, )?; diff --git a/vortex-array/src/arrays/runend/array/mod.rs b/vortex-array/src/arrays/runend/array/mod.rs index 000963a85da..ea967ab6295 100644 --- a/vortex-array/src/arrays/runend/array/mod.rs +++ b/vortex-array/src/arrays/runend/array/mod.rs @@ -927,4 +927,64 @@ mod tests { Ok(()) } + + // On 32-bit targets, a crafted num_runs / offset above usize::MAX used to panic via + // vortex_expect. Keep a regression that only builds where the conversion can fail. + #[cfg(target_pointer_width = "32")] + #[test] + fn test_deserialize_rejects_num_runs_overflow() -> VortexResult<()> { + let metadata = RunEndMetadata { + ends_ptype: PType::U32 as i32, + num_runs: u64::from(u32::MAX) + 1, + offset: 0, + } + .encode_to_vec(); + let children: Vec = Vec::new(); + let result = ArrayPlugin::deserialize( + &RunEnd, + ArrayDeserialization::new( + VTable::id(&RunEnd), + &DType::Primitive(PType::I32, Nullability::NonNullable), + 0, + &metadata, + &[], + &children, + ), + &SESSION, + ); + assert!(result.is_err_and(|err| { + err.to_string().contains("RunEndMetadata num_runs") + && err.to_string().contains("exceeds usize range") + })); + Ok(()) + } + + #[cfg(target_pointer_width = "32")] + #[test] + fn test_deserialize_rejects_offset_overflow() -> VortexResult<()> { + let metadata = RunEndMetadata { + ends_ptype: PType::U32 as i32, + num_runs: 0, + offset: u64::from(u32::MAX) + 1, + } + .encode_to_vec(); + let children: Vec = Vec::new(); + let result = ArrayPlugin::deserialize( + &RunEnd, + ArrayDeserialization::new( + VTable::id(&RunEnd), + &DType::Primitive(PType::I32, Nullability::NonNullable), + 0, + &metadata, + &[], + &children, + ), + &SESSION, + ); + assert!(result.is_err_and(|err| { + err.to_string().contains("RunEndMetadata offset") + && err.to_string().contains("exceeds usize range") + })); + Ok(()) + } } diff --git a/vortex-array/src/arrays/runend/array/plugin.rs b/vortex-array/src/arrays/runend/array/plugin.rs index 2de4227fd42..281b1278d68 100644 --- a/vortex-array/src/arrays/runend/array/plugin.rs +++ b/vortex-array/src/arrays/runend/array/plugin.rs @@ -7,6 +7,7 @@ use prost::Message; use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_ensure_eq; +use vortex_error::vortex_err; use vortex_session::VortexSession; use super::RunEnd; @@ -79,11 +80,21 @@ impl ArrayPlugin for RunEnd { let metadata = RunEndMetadata::decode(metadata)?; let ends_ptype = metadata.get_ends_ptype()?; let ends_dtype = DType::Primitive(ends_ptype, Nullability::NonNullable); - let runs = usize::try_from(metadata.num_runs).vortex_expect("Must be a valid usize"); + let runs = usize::try_from(metadata.num_runs).map_err(|_| { + vortex_err!( + "RunEndMetadata num_runs {} exceeds usize range", + metadata.num_runs + ) + })?; let ends = children.get(0, &ends_dtype, runs)?; let values = children.get(1, dtype, runs)?; - let offset = usize::try_from(metadata.offset).vortex_expect("Offset must be a valid usize"); + let offset = usize::try_from(metadata.offset).map_err(|_| { + vortex_err!( + "RunEndMetadata offset {} exceeds usize range", + metadata.offset + ) + })?; RunEndData::validate_parts( &ends, &values, From b2d08cc9203e50bcea93ab2cee37cf09305f60b6 Mon Sep 17 00:00:00 2001 From: ChristopherWirt Date: Wed, 11 Nov 2026 10:52:51 +0000 Subject: [PATCH 02/27] perf(scan): merge narrow splits at boundaries only some fields of a struct have (#11128) ## Summary - Tracking Issue: #11123 **Problem:** A scan splits a file wherever any column's chunk ends, and a split decodes the whole chunk of every column it overlaps. When a struct's fields chunk at unrelated rows, as list elements and dictionary-encoded columns of a wide nested file do, the union of their boundaries leaves spans of a few dozen rows. Successive splits then each decode the same chunks of the other fields. On a 180 MB list-layout file, a full scan held 3.6 GiB of native memory. **Fix:** The struct reader marks boundaries that only some of its fields have as *soft*. `SplitBy::LayoutSubSplitting` merges spans that end at a soft boundary until they reach 8,192 rows. Boundaries that every field has are never merged, and neither are boundaries from layouts above the struct, so files whose columns chunk alike split as before. **Result:** full scan of one file, with Arrow conversion as JNI does it. Live native bytes come from a counting allocator: | | Splits | Peak live | Time | Bytes requested | |---|---|---|---|---| | List layout, 180 MB, before | 139 (smallest 70 rows) | 3.57 GiB | 1.7 s | 6.2 GB | | List layout, 180 MB, after | 60 (smallest 8,192 rows) | **1.10 GiB** | **0.6 s** | 0.99 GB | | Default layout, 111 MB, before | 59 | 224 MiB | 0.2 s | | | Default layout, 111 MB, after | 48 | 224 MiB | 0.2 s | | ## Changes - **`RowSplits` gains soft marks.** Each struct field registers into its own `RowSplits`, and the struct takes their union. Any boundary that not every field has is marked soft. - The shared-boundary intersection is computed once. On the existing 256-column split-collection benchmark, the warm median falls from about 338 ms to 3 ms. - The struct's validity child is classified like any field, so a nullable struct merges the same way at any depth. - Soft marks inherited from nested structs are kept. - A single-field struct without validity registers exactly as before. - `SplitBy::Layout` collects positions only (`BoundaryMode::PositionsOnly`), so it skips the classification. - **`SplitBy::LayoutSubSplitting`** merges spans ending at a soft boundary until they reach `DEFAULT_MIN_SPLIT_ROWS` (8,192, the writer's row block), or `max_rows` if that is smaller. It never merges a span past `max_rows`. The merge runs before the existing sub-division of wide spans. --------- Signed-off-by: Christopher Wirt --- vortex-file/benches/split_collection.rs | 50 ++- vortex-layout/src/layouts/struct_/reader.rs | 384 +++++++++++++++++++- vortex-layout/src/reader.rs | 51 +++ vortex-layout/src/scan/split_by.rs | 201 +++++++++- 4 files changed, 668 insertions(+), 18 deletions(-) diff --git a/vortex-file/benches/split_collection.rs b/vortex-file/benches/split_collection.rs index 3a303be37d4..39627a39b13 100644 --- a/vortex-file/benches/split_collection.rs +++ b/vortex-file/benches/split_collection.rs @@ -1,7 +1,8 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright the Vortex contributors -//! Benchmarks scan split collection (`SplitBy::Layout`) over written files. +//! Benchmarks scan split collection over written files, with `SplitBy::Layout` and, in the +//! `*_sub_splitting` variants, the default `SplitBy::LayoutSubSplitting`. //! //! The default write strategy produces `struct -> zoned -> chunked -> flat` per column, so //! split collection walks every chunk of every column. `cold` builds a fresh reader tree per @@ -55,7 +56,15 @@ fn main() { const ROWS_PER_CHUNK: usize = 1024; /// (columns, chunks) configurations. -const CONFIGS: &[(usize, usize)] = &[(16, 64)]; +const CONFIGS: &[(usize, usize)] = &[ + (16, 8), + (24, 4), + // Keep the larger scaling cases for local runs without lengthening CPU simulation. + #[cfg(not(codspeed))] + (16, 64), + #[cfg(not(codspeed))] + (256, 64), +]; static RUNTIME: LazyLock = LazyLock::new(|| { tokio::runtime::Builder::new_current_thread() @@ -194,8 +203,12 @@ static MISALIGNED_FILES: LazyLock> = LazyLoc }); fn collect_splits(file: &VortexFile) -> Vec { + collect_splits_by(file, SplitBy::Layout) +} + +fn collect_splits_by(file: &VortexFile, split_by: SplitBy) -> Vec { let reader = file.layout_reader().unwrap(); - SplitBy::Layout + split_by .splits(reader.as_ref(), &(0..file.row_count()), &[FieldMask::All]) .unwrap() } @@ -248,3 +261,34 @@ fn cold_misaligned(bencher: Bencher, config: &(usize, usize)) { bencher.bench(|| collect_splits(file)); } + +/// Like `warm`, but with the default `SplitBy::LayoutSubSplitting`, which also classifies soft +/// boundaries. +#[divan::bench(args = CONFIGS)] +fn warm_sub_splitting(bencher: Bencher, config: &(usize, usize)) { + let file = &FILES[config]; + let reader = file.layout_reader().unwrap(); + let row_count = file.row_count(); + + bencher.bench(|| { + SplitBy::default() + .splits(reader.as_ref(), &(0..row_count), &[FieldMask::All]) + .unwrap() + }); +} + +/// Like `cold_misaligned`, but with the default `SplitBy::LayoutSubSplitting`. +#[divan::bench(args = MISALIGNED_CONFIGS)] +fn cold_misaligned_sub_splitting(bencher: Bencher, config: &(usize, usize)) { + let file = &MISALIGNED_FILES[config]; + // Sanity-check once per config that the file has soft boundaries to merge, so this measures + // the classification path rather than the cheap one. + let merged = collect_splits_by(file, SplitBy::default()).len(); + let layout = collect_splits(file).len(); + assert!( + merged < layout, + "expected soft boundaries to merge, got {merged} splits vs {layout} by layout" + ); + + bencher.bench(|| collect_splits_by(file, SplitBy::default())); +} diff --git a/vortex-layout/src/layouts/struct_/reader.rs b/vortex-layout/src/layouts/struct_/reader.rs index 6e80e0fab00..06b3a722bae 100644 --- a/vortex-layout/src/layouts/struct_/reader.rs +++ b/vortex-layout/src/layouts/struct_/reader.rs @@ -44,6 +44,7 @@ use vortex_utils::aliases::dash_map::DashMap; use vortex_utils::aliases::hash_map::HashMap; use crate::ArrayFuture; +use crate::BoundaryMode; use crate::LayoutReader; use crate::LayoutReaderRef; use crate::LazyReaderChildren; @@ -455,19 +456,90 @@ impl LayoutReader for StructReader { split_range: &SplitRange, splits: &mut RowSplits, ) -> VortexResult<()> { - // Register splits for the validity child, if there is one - if let Some(validity_ref) = self.validity()? { - validity_ref.register_splits(field_mask, split_range, splits)?; + if splits.mode() == BoundaryMode::PositionsOnly { + if let Some(validity_ref) = self.validity()? { + validity_ref.register_splits(field_mask, split_range, splits)?; + } + self.layout.matching_fields(field_mask, |mask, idx| { + self.field_reader_by_index(idx)? + .register_splits(&[mask], split_range, splits) + })?; + // Pushed last so it extends the final field's ascending run. + splits.push(split_range.root_row_range().end); + return Ok(()); } + let end = split_range.root_row_range().end; + let validity = self.validity()?; + let mut fields = Vec::new(); self.layout.matching_fields(field_mask, |mask, idx| { - self.field_reader_by_index(idx)? - .register_splits(&[mask], split_range, splits) + fields.push((mask, idx)); + Ok(()) })?; - // In the case of an empty struct, we need to register the end split. Pushed last so it - // extends the final field's ascending run rather than starting a run of its own. - splits.push(split_range.root_row_range().end); + if usize::from(validity.is_some()) + fields.len() <= 1 { + if let Some(validity_ref) = validity { + validity_ref.register_splits(field_mask, split_range, splits)?; + } + for (mask, idx) in fields { + self.field_reader_by_index(idx)? + .register_splits(&[mask], split_range, splits)?; + } + + splits.push(end); + return Ok(()); + } + + // Repeated boundary sets cannot change which positions every child shares. Keep their + // inherited soft marks even when their positions match the previous child. + let mut children = 0; + let mut previous_len = 0; + let mut all = Vec::new(); + let mut soft = Vec::new(); + let mut add = |child: RowSplits| { + let (boundaries, child_soft) = child.into_sorted_deduped_with_soft(); + if boundaries.is_empty() { + return; + } + soft.extend(child_soft); + if children > 0 && all[all.len() - previous_len..] == boundaries[..] { + return; + } + children += 1; + previous_len = boundaries.len(); + all.extend(boundaries); + }; + if let Some(validity_ref) = validity { + let mut child = splits.new_like(16); + validity_ref.register_splits(field_mask, split_range, &mut child)?; + add(child); + } + for (mask, idx) in fields { + let mut child = splits.new_like(16); + self.field_reader_by_index(idx)? + .register_splits(&[mask], split_range, &mut child)?; + add(child); + } + + // Each child contributes a position at most once, so its run length counts the children + // that share it. With one surviving child, `all` is already sorted. + if children > 1 { + all.sort_unstable(); + for run in all.chunk_by(|a, b| a == b) { + if run.len() < children && run[0] != end { + soft.push(run[0]); + } + } + all.dedup(); + } + soft.sort_unstable(); + soft.dedup(); + splits.extend_ascending(all); + splits.mark_soft(soft); + + // Append the struct's end even when its children contributed no boundaries, keeping the + // collected positions in one ascending run. + splits.push(end); Ok(()) } @@ -611,7 +683,9 @@ mod tests { use vortex_array::assert_arrays_eq; use vortex_array::assert_nth_scalar; use vortex_array::dtype::DType; + use vortex_array::dtype::FieldMask; use vortex_array::dtype::FieldName; + use vortex_array::dtype::FieldPath; use vortex_array::dtype::Nullability; use vortex_array::dtype::PType; use vortex_array::dtype::RESERVED_VALIDITY_FIELD_NAME; @@ -640,16 +714,27 @@ mod tests { use vortex_io::runtime::single::block_on; use vortex_io::session::RuntimeSessionExt; use vortex_mask::Mask; + use vortex_session::registry::ReadContext; + use crate::BoundaryMode; use crate::LayoutReaderRef; use crate::LayoutRef; use crate::LayoutStrategy; + use crate::children::OwnedLayoutChildren; + use crate::layouts::chunked::ChunkedLayout; use crate::layouts::chunked::writer::ChunkedLayoutStrategy; + use crate::layouts::flat::FlatLayout; use crate::layouts::flat::writer::FlatLayoutStrategy; + use crate::layouts::repartition::RepartitionStrategy; + use crate::layouts::repartition::RepartitionWriterOptions; + use crate::layouts::struct_::StructLayout; use crate::layouts::table::TableStrategy; use crate::layouts::zoned::writer::ZonedLayoutOptions; use crate::layouts::zoned::writer::ZonedStrategy; + use crate::scan::split_by::SplitBy; + use crate::scan::split_by::register_boundaries; use crate::segments::CountingSegmentSource; + use crate::segments::SegmentId; use crate::segments::SegmentSource; use crate::segments::TestSegments; use crate::sequence::SequenceId; @@ -733,6 +818,289 @@ mod tests { (segments, layout) } + /// A leaf writer that chunks every `rows` rows. + fn chunked_every(rows: usize) -> Arc { + Arc::new(RepartitionStrategy::new( + ChunkedLayoutStrategy::new(FlatLayoutStrategy::default()), + RepartitionWriterOptions { + block_size_minimum: 0, + block_len_multiple: rows, + block_size_target: None, + canonicalize: false, + }, + )) + } + + /// A struct of 12,000 rows whose field `a` chunks every 1,000 rows and `b` every 3,000, + /// written in `pieces` pushes. With `chunked`, the struct is itself a chunk of a chunked layout + /// per push. + fn misaligned_struct_layout( + pieces: usize, + chunked: bool, + nullable: bool, + ) -> VortexResult<(Arc, LayoutRef)> { + let ctx = ArrayContext::empty(); + let segments = Arc::new(TestSegments::default()); + let (ptr, eof) = SequenceId::root().split(); + let table = TableStrategy::new(chunked_every(2000), chunked_every(1000)) + .with_field_writer(FieldPath::from_name("b"), chunked_every(3000)); + let strategy: Arc = if chunked { + Arc::new(ChunkedLayoutStrategy::new(table)) + } else { + Arc::new(table) + }; + let rows = 12_000 / pieces; + let chunks = (0..pieces) + .map(|piece| { + let start = i32::try_from(piece * rows)?; + let end = start + i32::try_from(rows)?; + Ok(StructArray::try_from_iter_with_validity( + [ + ("a", PrimitiveArray::from_iter(start..end).into_array()), + ("b", PrimitiveArray::from_iter(start..end).into_array()), + ], + if nullable { + Validity::Array( + BoolArray::from_iter((start..end).map(|row| row % 2 == 0)).into_array(), + ) + } else { + Validity::NonNullable + }, + )? + .into_array()) + }) + .collect::>>()?; + let dtype = chunks[0].dtype().clone(); + let segments2 = Arc::::clone(&segments); + let layout = block_on(|handle| async move { + let session = new_session().with_handle(handle); + strategy + .write_stream( + ctx.into(), + segments2, + ChunkedArray::try_new(chunks, dtype)? + .into_array() + .to_array_stream() + .sequenced(ptr), + eof, + &session, + ) + .await + })?; + Ok((segments, layout)) + } + + /// Registers `layout`'s boundaries over its rows, as a scan does. + fn boundaries_and_soft( + segments: Arc, + layout: &LayoutRef, + ) -> VortexResult<(Vec, Vec)> { + let reader = layout.new_reader("".into(), segments, &SESSION, &Default::default())?; + Ok(register_boundaries( + reader.as_ref(), + &(0..reader.row_count()), + &[FieldMask::All], + BoundaryMode::WithSoftMarks, + )? + .into_sorted_deduped_with_soft()) + } + + #[test] + fn boundaries_only_some_fields_have_are_soft() -> VortexResult<()> { + let (segments, layout) = misaligned_struct_layout(1, false, false)?; + let (boundaries, soft) = boundaries_and_soft(segments, &layout)?; + // `a` chunks every 1,000 rows and `b` every 3,000, so only `b`'s are shared. + assert_eq!(boundaries, (0..=12).map(|k| k * 1000).collect::>()); + assert_eq!( + soft, + (1..12) + .filter(|k| k % 3 != 0) + .map(|k| k * 1000) + .collect::>() + ); + Ok(()) + } + + #[test] + fn chunk_ends_above_the_struct_stay_hard() -> VortexResult<()> { + // Two top-level chunks of 6,000 rows, each a struct whose `a` chunks every 1,000 rows and + // `b` every 3,000: the top-level chunk end at 6,000 is every column's, so it stays hard. + let (segments, layout) = misaligned_struct_layout(2, true, false)?; + let (boundaries, soft) = boundaries_and_soft(segments, &layout)?; + assert_eq!(boundaries, (0..=12).map(|k| k * 1000).collect::>()); + assert!(!soft.contains(&6000), "the chunk end is soft: {soft:?}"); + assert!(!soft.contains(&3000) && !soft.contains(&9000), "{soft:?}"); + assert!(soft.contains(&1000) && soft.contains(&7000), "{soft:?}"); + Ok(()) + } + + #[rstest] + #[case::non_nullable(false, vec![0, 3000, 6000, 9000, 12000])] + // Validity chunks every 2,000 rows, so only 6,000 is every child's boundary. + #[case::nullable(true, vec![0, 6000, 12000])] + fn scan_merges_soft_boundaries_including_validity_boundaries( + #[case] nullable: bool, + #[case] expected: Vec, + ) -> VortexResult<()> { + let (segments, layout) = misaligned_struct_layout(1, false, nullable)?; + let reader = layout.new_reader("".into(), segments, &SESSION, &Default::default())?; + assert_eq!( + SplitBy::default().splits( + reader.as_ref(), + &(0..reader.row_count()), + &[FieldMask::All], + )?, + expected, + ); + Ok(()) + } + + fn split_test_chunks(lengths: &[u64]) -> LayoutRef { + let dtype = DType::Primitive(PType::I32, Nullability::NonNullable); + let children = lengths + .iter() + .map(|&len| { + FlatLayout::new(len, dtype.clone(), SegmentId::from(0), ReadContext::new([])) + .into_layout() + }) + .collect(); + ChunkedLayout::new( + lengths.iter().sum(), + dtype, + OwnedLayoutChildren::layout_children(children), + ) + .into_layout() + } + + fn split_test_struct(fields: Vec<(&str, LayoutRef)>) -> LayoutRef { + let dtype = DType::Struct( + StructFields::from_iter( + fields + .iter() + .map(|(name, layout)| (*name, layout.dtype().clone())), + ), + Nullability::NonNullable, + ); + StructLayout::new( + 6, + dtype, + fields.into_iter().map(|(_, layout)| layout).collect(), + ) + .into_layout() + } + + const FINE_CHUNKS: &[u64] = &[2, 2, 2]; + const COARSE_CHUNKS: &[u64] = &[4, 2]; + + #[rstest] + #[case::consecutive_duplicates(&[FINE_CHUNKS, FINE_CHUNKS, COARSE_CHUNKS], &[2])] + #[case::separated_duplicates(&[FINE_CHUNKS, COARSE_CHUNKS, FINE_CHUNKS], &[2])] + #[case::all_identical(&[FINE_CHUNKS, FINE_CHUNKS, FINE_CHUNKS], &[])] + fn split_collection_repeated_child_boundaries( + #[case] chunk_lengths: &[&[u64]], + #[case] expected_soft: &[u64], + #[values(BoundaryMode::PositionsOnly, BoundaryMode::WithSoftMarks)] mode: BoundaryMode, + ) -> VortexResult<()> { + let fields = ["a", "b", "c"] + .into_iter() + .zip( + chunk_lengths + .iter() + .map(|lengths| split_test_chunks(lengths)), + ) + .collect(); + let layout = split_test_struct(fields); + let reader = layout.new_reader( + "".into(), + Arc::new(TestSegments::default()), + &SESSION, + &Default::default(), + )?; + let (boundaries, soft) = + register_boundaries(reader.as_ref(), &(0..6), &[FieldMask::All], mode)? + .into_sorted_deduped_with_soft(); + assert_eq!(boundaries, vec![0, 2, 4, 6]); + assert_eq!( + soft, + if mode == BoundaryMode::WithSoftMarks { + expected_soft + } else { + &[] + } + ); + Ok(()) + } + + #[rstest] + #[case::single_child(true, false)] + #[case::nested_first(false, true)] + #[case::nested_last(false, false)] + fn split_collection_preserves_inherited_soft_marks( + #[case] single_child: bool, + #[case] nested_first: bool, + #[values(BoundaryMode::PositionsOnly, BoundaryMode::WithSoftMarks)] mode: BoundaryMode, + #[values(0..6, 0..0)] range: std::ops::Range, + ) -> VortexResult<()> { + let nested = split_test_struct(vec![ + ("fine", split_test_chunks(&[2, 2, 2])), + ("coarse", split_test_chunks(&[6])), + ]); + let mut fields = vec![("nested", nested)]; + if !single_child { + // Identical positions, but only the nested child's interior boundaries are soft. + fields.push(("aligned", split_test_chunks(&[2, 2, 2]))); + if !nested_first { + fields.reverse(); + } + } + let layout = split_test_struct(fields); + let reader = layout.new_reader( + "".into(), + Arc::new(TestSegments::default()), + &SESSION, + &Default::default(), + )?; + let (boundaries, soft) = + register_boundaries(reader.as_ref(), &range, &[FieldMask::All], mode)? + .into_sorted_deduped_with_soft(); + assert_eq!( + boundaries, + if range.is_empty() { + vec![0] + } else { + vec![0, 2, 4, 6] + } + ); + assert_eq!( + soft, + if !range.is_empty() && mode == BoundaryMode::WithSoftMarks { + vec![2, 4] + } else { + vec![] + } + ); + Ok(()) + } + + #[rstest] + fn split_collection_empty_struct( + #[values(BoundaryMode::PositionsOnly, BoundaryMode::WithSoftMarks)] mode: BoundaryMode, + ) -> VortexResult<()> { + let layout = split_test_struct(vec![]); + let reader = layout.new_reader( + "".into(), + Arc::new(TestSegments::default()), + &SESSION, + &Default::default(), + )?; + let (boundaries, soft) = + register_boundaries(reader.as_ref(), &(0..6), &[FieldMask::All], mode)? + .into_sorted_deduped_with_soft(); + assert_eq!(boundaries, vec![0, 6]); + assert!(soft.is_empty()); + Ok(()) + } + #[fixture] /// Create a chunked layout with three chunks of primitive arrays. fn null_struct_layout() -> (Arc, LayoutRef) { diff --git a/vortex-layout/src/reader.rs b/vortex-layout/src/reader.rs index 91dda0e23c5..65bc9325760 100644 --- a/vortex-layout/src/reader.rs +++ b/vortex-layout/src/reader.rs @@ -94,18 +94,35 @@ impl SplitRange { } } +/// What a [`RowSplits`] collector is asked to gather. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub enum BoundaryMode { + /// Boundary positions only; readers skip classifying soft boundaries. + #[default] + PositionsOnly, + /// Boundary positions plus soft marks (see [`RowSplits::mark_soft`]). + WithSoftMarks, +} + /// A collection of root-coordinate row split points. /// /// Boundaries arrive as non-descending runs, one per layout subtree walked; a descending push /// starts a new run. A run that exactly repeats the previous surviving run — common when sibling /// columns share chunk boundaries — is dropped as it completes, and the final sort is skipped /// when only a single run survives (the boundaries are then already ascending). +/// +/// A boundary can also be marked soft: one that only some of a struct's fields have. A split there +/// lets the fields that chunk across it decode the same chunk in two splits, so the scan may merge +/// it away (see [`crate::scan::split_by::SplitBy::LayoutSubSplitting`]). pub struct RowSplits { splits: Vec, /// Start index of the run currently being appended. run_start: usize, /// Start index of the surviving run immediately before `run_start`. prev_run_start: usize, + /// Boundaries marked soft, unsorted and possibly repeated. + soft: Vec, + mode: BoundaryMode, } impl RowSplits { @@ -162,15 +179,49 @@ impl RowSplits { self.splits.reserve(additional); } + /// The collection mode requested by the consumer. + pub fn mode(&self) -> BoundaryMode { + self.mode + } + + /// An empty collector with the given capacity that inherits `self`'s mode. + pub(crate) fn new_like(&self, capacity: usize) -> Self { + Self::new_capacity_with_mode(capacity, self.mode) + } + + /// Create a new RowSplits with preallocated "capacity" and the given mode. + pub(crate) fn new_capacity_with_mode(capacity: usize, mode: BoundaryMode) -> Self { + Self { + mode, + ..Self::new_capacity(capacity) + } + } + + /// Mark boundaries soft: ones that only some fields of a struct have. They must also be + /// pushed. + pub fn mark_soft(&mut self, rows: impl IntoIterator) { + self.soft.extend(rows); + } + /// Create a new RowSplits with preallocated "capacity" pub(crate) fn new_capacity(capacity: usize) -> Self { Self { splits: Vec::with_capacity(capacity), run_start: 0, prev_run_start: 0, + soft: Vec::new(), + mode: BoundaryMode::PositionsOnly, } } + /// The sorted, deduplicated boundaries, and the sorted soft ones among them. + pub(crate) fn into_sorted_deduped_with_soft(mut self) -> (Vec, Vec) { + let mut soft = std::mem::take(&mut self.soft); + soft.sort_unstable(); + soft.dedup(); + (self.into_sorted_deduped(), soft) + } + pub(crate) fn into_sorted_deduped(mut self) -> Vec { let final_run_dropped = self.drop_repeated_run(); // Surviving runs always have a descent between them, so the boundaries are ascending diff --git a/vortex-layout/src/scan/split_by.rs b/vortex-layout/src/scan/split_by.rs index b995e46222f..c5cb98312c9 100644 --- a/vortex-layout/src/scan/split_by.rs +++ b/vortex-layout/src/scan/split_by.rs @@ -9,6 +9,7 @@ use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_ensure; +use crate::BoundaryMode; use crate::LayoutReader; use crate::RowSplits; use crate::SplitRange; @@ -19,6 +20,14 @@ use crate::scan::IDEAL_SPLIT_SIZE; /// be decoded across multiple cores rather than one. pub const DEFAULT_MAX_SPLIT_ROWS: u64 = IDEAL_SPLIT_SIZE; +/// The narrowest span [`SplitBy::LayoutSubSplitting`] leaves ending at a soft boundary, one that +/// only some fields of a struct have; narrower spans merge into a neighbour. It is the writer's +/// default row block, and is lowered to `max_rows` when that is smaller. +/// +/// A split decodes the whole chunk of every column it overlaps, so splits much narrower than a +/// column's chunks decode those chunks once each. +pub const DEFAULT_MIN_SPLIT_ROWS: u64 = 8192; + /// Defines how the Vortex file is split into batches for reading. /// /// Note that each split must fit into the platform's maximum usize. @@ -28,10 +37,12 @@ pub enum SplitBy { /// splits that follow the file's chunking exactly, trading intra-file decode parallelism /// for fewer, larger batches. Layout, - /// Splits like [`SplitBy::Layout`], except that spans between adjacent chunk boundaries - /// wider than `max_rows` are further sub-divided into evenly sized splits of at most - /// `max_rows` rows, so that a file with few, large chunks can still be decoded across - /// multiple cores. `max_rows` must be non-zero. + /// Splits like [`SplitBy::Layout`], except that soft boundaries (ones only some fields of a + /// struct have) are dropped where they leave a span narrower than [`DEFAULT_MIN_SPLIT_ROWS`] + /// (or `max_rows`, if smaller), unless that would merge a span wider than `max_rows`, and spans + /// wider than `max_rows` are further sub-divided into evenly sized + /// splits of at most `max_rows` rows, so that a file with few, large chunks can still be + /// decoded across multiple cores. `max_rows` must be non-zero. /// /// This is the default, with [`DEFAULT_MAX_SPLIT_ROWS`]. LayoutSubSplitting { max_rows: u64 }, @@ -64,8 +75,20 @@ impl SplitBy { max_rows > 0, "SplitBy::LayoutSubSplitting requires a non-zero max_rows" ); + let (boundaries, soft) = register_boundaries( + layout_reader, + row_range, + field_mask, + BoundaryMode::WithSoftMarks, + )? + .into_sorted_deduped_with_soft(); subdivide_large_spans( - layout_boundaries(layout_reader, row_range, field_mask)?, + merge_small_spans( + boundaries, + &soft, + DEFAULT_MIN_SPLIT_ROWS.min(max_rows), + max_rows, + ), max_rows, ) } @@ -85,15 +108,84 @@ fn layout_boundaries( row_range: &Range, field_mask: &[FieldMask], ) -> VortexResult> { + Ok(register_boundaries( + layout_reader, + row_range, + field_mask, + BoundaryMode::PositionsOnly, + )? + .into_sorted_deduped()) +} + +/// The boundaries the layout registers within `row_range`, after the range's start (the layout +/// registers its end). +pub(crate) fn register_boundaries( + layout_reader: &dyn LayoutReader, + row_range: &Range, + field_mask: &[FieldMask], + mode: BoundaryMode, +) -> VortexResult { // We usually have under 100 splits so reserving upfront saves us some allocations - let mut row_splits = RowSplits::new_capacity(128); + let mut row_splits = RowSplits::new_capacity_with_mode(128, mode); row_splits.push(row_range.start); layout_reader.register_splits( field_mask, &SplitRange::root(row_range.clone())?, &mut row_splits, )?; - Ok(row_splits.into_sorted_deduped()) + Ok(row_splits) +} + +/// Remove interior split boundaries in the sorted `soft` until no span ending at one is narrower +/// than `min_span`, without merging any span past `max_span`. +/// +/// Like [`subdivide_large_spans`], this works on the sorted, deduplicated boundary list. It never +/// adds or moves a boundary, and always keeps the first, the last and every boundary not in `soft`, +/// so the ranges still cover the same rows contiguously, each exactly once. Boundaries are dropped +/// from the front: a soft boundary is kept once the span since the last kept one reaches +/// `min_span`. A short span left before the end, or before a hard boundary, keeps its end and drops +/// its start where that is soft. A soft boundary is also kept where dropping it would leave a span +/// wider than `max_span`, which [`subdivide_large_spans`] would re-split at a row no column chunks +/// at. +fn merge_small_spans(boundaries: Vec, soft: &[u64], min_span: u64, max_span: u64) -> Vec { + debug_assert!(boundaries.is_sorted(), "boundaries must be sorted"); + debug_assert!(soft.is_sorted(), "soft boundaries must be sorted"); + let is_hard = |b: u64| soft.binary_search(&b).is_err(); + + // Fast path: nothing to merge (also covers the empty / single-boundary cases). + if soft.is_empty() + || boundaries.len() < 3 + || boundaries.windows(2).all(|w| w[1] - w[0] >= min_span) + { + return boundaries; + } + + let last = boundaries.len() - 1; + let mut out = Vec::with_capacity(boundaries.len()); + out.push(boundaries[0]); + for (i, &boundary) in boundaries.iter().enumerate().skip(1) { + let prev = out[out.len() - 1]; + if i == last || is_hard(boundary) { + // A short span before a kept boundary merges into the one before it, unless its + // start must be kept too or the merged span would be too wide. + if boundary - prev < min_span + && out.len() > 1 + && !is_hard(prev) + && boundary - out[out.len() - 2] <= max_span + { + out.pop(); + } + out.push(boundary); + } else if boundary - prev >= min_span || boundaries[i + 1] - prev > max_span { + out.push(boundary); + } + } + + debug_assert!( + out.windows(2).all(|w| w[0] < w[1]), + "merged boundaries must stay strictly increasing" + ); + out } /// Sub-divide any gap between adjacent split boundaries that is wider than `max_span` into evenly @@ -156,6 +248,10 @@ mod test { use std::sync::Arc; use futures::future::BoxFuture; + use rand::RngExt; + use rand::SeedableRng; + use rand::prelude::StdRng; + use rstest::rstest; use vortex_array::ArrayContext; use vortex_array::ArrayRef; use vortex_array::IntoArray; @@ -363,6 +459,97 @@ mod test { assert_eq!(subdivide_large_spans(vec![7], 100), vec![7]); } + #[rstest] + #[case::empty(vec![], vec![], 5, vec![])] + #[case::single(vec![7], vec![7], 5, vec![7])] + #[case::hard(vec![0, 1, 2, 10], vec![], 5, vec![0, 1, 2, 10])] + #[case::soft(vec![0, 1, 2, 10], vec![1, 2], 5, vec![0, 10])] + #[case::short_tail(vec![0, 5, 6], vec![5], 5, vec![0, 6])] + #[case::hard_collision(vec![0, 3, 5, 6], vec![3, 5], 5, vec![0, 6])] + #[case::hard_start(vec![0, 3, 5, 6], vec![5], 5, vec![0, 3, 6])] + #[case::threshold(vec![0, 5, 10], vec![5], 5, vec![0, 5, 10])] + #[case::nonzero_range(vec![100, 101, 102, 110], vec![101, 102], 5, vec![100, 110])] + fn merge_small_spans_cases( + #[case] boundaries: Vec, + #[case] soft: Vec, + #[case] minimum: u64, + #[case] expected: Vec, + ) { + assert_eq!( + merge_small_spans(boundaries, &soft, minimum, u64::MAX), + expected + ); + } + + #[test] + fn merging_never_widens_a_span_past_max_span() { + // Dropping 2,000 would leave one 101,000-row span to re-split at 50,500, a row no column + // chunks at. + assert_eq!( + merge_small_spans(vec![0, 2000, 101_000], &[2000], 8192, 100_000), + vec![0, 2000, 101_000] + ); + // Nor drop 95,000 to absorb the short tail. + assert_eq!( + merge_small_spans(vec![0, 95_000, 101_000], &[95_000], 8192, 100_000), + vec![0, 95_000, 101_000] + ); + } + + #[test] + fn merging_preserves_hard_boundaries_and_exact_coverage() { + let mut rng = StdRng::seed_from_u64(67); + for _ in 0..1000 { + let start = rng.random_range(0..1000u64); + let mut boundaries = vec![start]; + for _ in 0..rng.random_range(1..50) { + boundaries.push(boundaries.last().unwrap() + rng.random_range(1..1000)); + } + let soft = boundaries + .iter() + .copied() + .filter(|_| rng.random_bool(0.5)) + .collect::>(); + let max_rows = rng.random_range(1..3000); + let minimum = DEFAULT_MIN_SPLIT_ROWS.min(max_rows); + let merged = merge_small_spans(boundaries.clone(), &soft, minimum, max_rows); + assert_eq!(merged.first(), boundaries.first()); + assert_eq!(merged.last(), boundaries.last()); + assert!(merged.windows(2).all(|w| w[0] < w[1])); + assert!( + merged + .iter() + .all(|row| boundaries.binary_search(row).is_ok()) + ); + assert!(boundaries.iter().all(|row| { + soft.binary_search(row).is_ok() || merged.binary_search(row).is_ok() + })); + // A short span may end at a soft boundary only where dropping it would merge past + // `max_rows`. + assert!(merged.windows(2).all(|w| { + Some(&w[1]) == merged.last() + || soft.binary_search(&w[1]).is_err() + || w[1] - w[0] >= minimum + || boundaries[boundaries.binary_search(&w[1]).unwrap() + 1] - w[0] > max_rows + })); + // A merged span wider than `max_rows` would be re-split off any chunk boundary. + assert!(merged.windows(2).all(|w| { + w[1] - w[0] <= max_rows + || boundaries + .binary_search(&w[0]) + .ok() + .map(|i| boundaries[i + 1]) + == Some(w[1]) + })); + let subdivided = subdivide_large_spans(merged, max_rows); + assert!(subdivided.windows(2).all(|w| w[1] - w[0] <= max_rows)); + assert_eq!( + subdivided.windows(2).map(|w| w[1] - w[0]).sum::(), + boundaries.last().unwrap() - start, + ); + } + } + #[test] fn subdivide_near_u64_max_does_not_overflow() { // The increment past the last interior point would overflow without saturating math. From 5dac460f4362e57f4faa1d38a15acd59ccd261c7 Mon Sep 17 00:00:00 2001 From: Will Manning Date: Wed, 11 Nov 2026 06:42:54 -0500 Subject: [PATCH 03/27] Link the vortex.dev FAQ from the docs (#11385) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary vortex.dev has a new FAQ page, https://vortex.dev/faq, from vortex-data/vortex.dev#218. This PR links it from the docs: in the footer before Charter, as the vortex.dev footer has it, in the project page's links, and in the docs' llms.txt beside the homepage. ## Changes `make -C docs html` builds with no warnings, and the built footer, project page, and llms.txt each carry the link. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Signed-off-by: Will Manning --- docs/_ext/vortex_site_files.py | 2 ++ docs/_theme/vortex/layout.html | 1 + docs/project/index.md | 1 + 3 files changed, 4 insertions(+) diff --git a/docs/_ext/vortex_site_files.py b/docs/_ext/vortex_site_files.py index cd52666ef46..654be5a7476 100644 --- a/docs/_ext/vortex_site_files.py +++ b/docs/_ext/vortex_site_files.py @@ -95,6 +95,8 @@ def _llms(app: Sphinx) -> str: "", "- [Benchmark summaries](https://bench.vortex.dev/llms.txt): every benchmark group's rankings and methods", "- [Vortex homepage](https://vortex.dev/llms.txt): what Vortex is, how to install it, and its results", + "- [Questions and answers](https://vortex.dev/faq): how Vortex compares to Parquet and Lance, " + "who uses it, and how it is governed", "", "## Optional", "", diff --git a/docs/_theme/vortex/layout.html b/docs/_theme/vortex/layout.html index eb4e175e3d7..81ee7148135 100644 --- a/docs/_theme/vortex/layout.html +++ b/docs/_theme/vortex/layout.html @@ -132,6 +132,7 @@ The Linux Foundation

Copyright © Vortex a Series of LF Projects, LLC. For web site terms of use, trademark policy and other project policies please see lfprojects.org/policies. Donated by Spiral.