From 2c86191beffaf76e884fb3276aa36cf991091d9d Mon Sep 17 00:00:00 2001 From: Matt Katz Date: Wed, 29 Jul 2026 11:42:21 +0100 Subject: [PATCH 1/2] Record list boundaries for parallel reads Signed-off-by: Matt Katz --- .github/workflows/sql-benchmarks.yml | 1 + vortex-layout/src/layouts/list/mod.rs | 161 +++++++++++++++++++++- vortex-layout/src/layouts/list/reader.rs | 167 +++++++++++++++++++---- vortex-layout/src/layouts/list/writer.rs | 69 +++++++++- 4 files changed, 365 insertions(+), 33 deletions(-) diff --git a/.github/workflows/sql-benchmarks.yml b/.github/workflows/sql-benchmarks.yml index 918762cea96..62cc634aa14 100644 --- a/.github/workflows/sql-benchmarks.yml +++ b/.github/workflows/sql-benchmarks.yml @@ -49,6 +49,7 @@ jobs: env: VORTEX_EXPERIMENTAL_PATCHED_ARRAY: "1" FLAT_LAYOUT_INLINE_ARRAY_NODE: "1" + VORTEX_EXPERIMENTAL_LIST_LAYOUT: "1" # Makes python output nicer COLUMNS: 120 strategy: diff --git a/vortex-layout/src/layouts/list/mod.rs b/vortex-layout/src/layouts/list/mod.rs index 0ba5be17dbc..ca19f6b8a9c 100644 --- a/vortex-layout/src/layouts/list/mod.rs +++ b/vortex-layout/src/layouts/list/mod.rs @@ -55,6 +55,31 @@ pub use List as ListLayoutEncoding; #[derive(Clone, Debug)] pub struct ListData { offsets_ptype: PType, + block_boundaries: Arc<[ListBlockBoundary]>, +} + +/// Cumulative row boundaries for one input block of a structural list layout. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) struct ListBlockBoundary { + outer_row_end: u64, + element_row_end: u64, +} + +impl ListBlockBoundary { + pub(super) fn new(outer_row_end: u64, element_row_end: u64) -> Self { + Self { + outer_row_end, + element_row_end, + } + } + + pub(super) fn outer_row_end(self) -> u64 { + self.outer_row_end + } + + pub(super) fn element_row_end(self) -> u64 { + self.element_row_end + } } /// A list layout shredded into elements, offsets, and optional validity children. @@ -70,7 +95,10 @@ impl VTable for List { } fn metadata(layout: &Layout) -> Self::Metadata { - ProstMetadata(ListLayoutMetadata::new(layout.offsets_ptype)) + ProstMetadata(ListLayoutMetadata::new_with_boundaries( + layout.offsets_ptype, + &layout.block_boundaries, + )) } fn deserialize( @@ -83,7 +111,7 @@ impl VTable for List { .dtype .as_list_element_opt() .ok_or_else(|| vortex_err!("ListLayout requires a List dtype, got {}", args.dtype))?; - args.children.child(ELEMENTS_CHILD_INDEX, elements_dtype)?; + let elements = args.children.child(ELEMENTS_CHILD_INDEX, elements_dtype)?; let offsets_dtype = DType::Primitive(metadata.offsets_ptype(), Nullability::NonNullable); let offsets = args.children.child(OFFSETS_CHILD_INDEX, &offsets_dtype)?; vortex_error::vortex_ensure!( @@ -99,8 +127,18 @@ impl VTable for List { "List validity row count does not match parent" ); } + let block_boundaries = metadata + .block_boundaries + .iter() + .map(|boundary| { + ListBlockBoundary::new(boundary.outer_row_end, boundary.element_row_end) + }) + .collect::>(); + validate_block_boundaries(args.row_count, elements.row_count(), &block_boundaries)?; + Ok(ListData { offsets_ptype: metadata.offsets_ptype(), + block_boundaries: block_boundaries.into(), }) } @@ -170,9 +208,21 @@ impl Layout { elements: LayoutRef, offsets: LayoutRef, validity: Option, + ) -> Self { + Self::new_with_boundaries(dtype, elements, offsets, validity, Vec::new()) + } + + pub(super) fn new_with_boundaries( + dtype: DType, + elements: LayoutRef, + offsets: LayoutRef, + validity: Option, + block_boundaries: Vec, ) -> Self { let row_count = offsets.row_count().saturating_sub(1); let offsets_ptype = offsets.dtype().as_ptype(); + validate_block_boundaries(row_count, elements.row_count(), &block_boundaries) + .vortex_expect("invalid list block boundaries"); let mut children = vec![elements, offsets]; children.extend(validity); Self::validate_children(&dtype, children.len()).vortex_expect("invalid list children"); @@ -182,7 +232,10 @@ impl Layout { row_count, Vec::new(), OwnedLayoutChildren::layout_children(children), - ListData { offsets_ptype }, + ListData { + offsets_ptype, + block_boundaries: block_boundaries.into(), + }, ) .into_typed() } @@ -209,6 +262,10 @@ impl Layout { self.offsets_ptype } + pub(super) fn block_boundaries(&self) -> &[ListBlockBoundary] { + &self.block_boundaries + } + /// Returns the list element dtype. pub fn elements_dtype(&self) -> &DType { self.dtype() @@ -227,12 +284,110 @@ impl Layout { pub struct ListLayoutMetadata { #[prost(enumeration = "PType", tag = "1")] offsets_ptype: i32, + #[prost(message, repeated, tag = "2")] + block_boundaries: Vec, +} + +#[derive(Clone, PartialEq, Eq, prost::Message)] +struct ListBlockBoundaryMetadata { + #[prost(uint64, tag = "1")] + outer_row_end: u64, + #[prost(uint64, tag = "2")] + element_row_end: u64, } impl ListLayoutMetadata { pub fn new(offsets_ptype: PType) -> Self { + Self::new_with_boundaries(offsets_ptype, &[]) + } + + fn new_with_boundaries(offsets_ptype: PType, block_boundaries: &[ListBlockBoundary]) -> Self { let mut metadata = Self::default(); metadata.set_offsets_ptype(offsets_ptype); + metadata.block_boundaries = block_boundaries + .iter() + .map(|boundary| ListBlockBoundaryMetadata { + outer_row_end: boundary.outer_row_end(), + element_row_end: boundary.element_row_end(), + }) + .collect(); metadata } } + +fn validate_block_boundaries( + outer_row_count: u64, + element_row_count: u64, + block_boundaries: &[ListBlockBoundary], +) -> VortexResult<()> { + let Some(last) = block_boundaries.last() else { + return Ok(()); + }; + + vortex_error::vortex_ensure!( + block_boundaries[0].outer_row_end != 0, + "List block outer-row boundaries must not contain zero" + ); + vortex_error::vortex_ensure!( + block_boundaries + .windows(2) + .all(|pair| pair[0].outer_row_end < pair[1].outer_row_end), + "List block outer-row boundaries must be strictly increasing" + ); + vortex_error::vortex_ensure!( + block_boundaries + .windows(2) + .all(|pair| pair[0].element_row_end <= pair[1].element_row_end), + "List block element-row boundaries must be non-decreasing" + ); + vortex_error::vortex_ensure_eq!(last.outer_row_end, outer_row_count); + vortex_error::vortex_ensure_eq!(last.element_row_end, element_row_count); + Ok(()) +} + +#[cfg(test)] +mod tests { + use vortex_array::DeserializeMetadata; + use vortex_array::SerializeMetadata; + + use super::*; + + #[test] + fn block_boundaries_round_trip_through_metadata() -> VortexResult<()> { + let boundaries = [ + ListBlockBoundary::new(8, 21), + ListBlockBoundary::new(16, 50), + ]; + let encoded = ProstMetadata(ListLayoutMetadata::new_with_boundaries( + PType::U64, + &boundaries, + )) + .serialize(); + let decoded = + as DeserializeMetadata>::deserialize(&encoded)?; + + assert_eq!(decoded.offsets_ptype(), PType::U64); + assert_eq!( + decoded + .block_boundaries + .iter() + .map(|boundary| { + ListBlockBoundary::new(boundary.outer_row_end, boundary.element_row_end) + }) + .collect::>(), + boundaries + ); + Ok(()) + } + + #[test] + fn legacy_metadata_has_no_block_boundaries() -> VortexResult<()> { + let encoded = ProstMetadata(ListLayoutMetadata::new(PType::U32)).serialize(); + let decoded = + as DeserializeMetadata>::deserialize(&encoded)?; + + assert_eq!(decoded.offsets_ptype(), PType::U32); + assert!(decoded.block_boundaries.is_empty()); + Ok(()) + } +} diff --git a/vortex-layout/src/layouts/list/reader.rs b/vortex-layout/src/layouts/list/reader.rs index 1e21714fdd0..b787da3d634 100644 --- a/vortex-layout/src/layouts/list/reader.rs +++ b/vortex-layout/src/layouts/list/reader.rs @@ -25,6 +25,7 @@ use vortex_array::scalar_fn::fns::operators::Operator; use vortex_array::validity::Validity; use vortex_error::VortexExpect; use vortex_error::VortexResult; +use vortex_error::vortex_bail; use vortex_mask::Mask; use vortex_session::VortexSession; @@ -220,18 +221,36 @@ impl ListReader { let expr = expr.clone(); let reader = self.clone(); let offsets_fut = self.fetch_raw_offsets(&selected_row_range)?; + let validity_fut = fetch_validity( + self.validity.as_ref(), + &selected_row_range, + MaskFuture::new_true(selected_mask.len()), + )?; + let element_prefetch_range = self.element_prefetch_range(&selected_row_range); + let children_fut = if let Some(prefetch_range) = element_prefetch_range.clone() { + let elements_fut = self.fetch_raw_elements(&prefetch_range)?; + let session = self.session.clone(); + async move { + let (offsets, elements, validity) = + try_join!(offsets_fut, elements_fut, validity_fut)?; + let exact_range = elements_range_from_offsets(&offsets, &session)?; + let elements = slice_prefetched_elements(elements, &prefetch_range, &exact_range)?; + Ok::<_, vortex_error::VortexError>((offsets, elements, validity, exact_range)) + } + .boxed() + } else { + let reader = reader.clone(); + async move { + let (offsets, validity) = try_join!(offsets_fut, validity_fut)?; + let exact_range = elements_range_from_offsets(&offsets, &reader.session)?; + let elements = reader.fetch_raw_elements(&exact_range)?.await?; + Ok::<_, vortex_error::VortexError>((offsets, elements, validity, exact_range)) + } + .boxed() + }; Ok(async move { - let offsets = offsets_fut.await?; - - let elements_range = elements_range_from_offsets(&offsets, &reader.session)?; - let elements_fut = reader.fetch_raw_elements(&elements_range)?; - let validity_fut = fetch_validity( - reader.validity.as_ref(), - &selected_row_range, - MaskFuture::new_true(selected_mask.len()), - )?; - let (elements, validity) = try_join!(elements_fut, validity_fut)?; + let (offsets, elements, validity, elements_range) = children_fut.await?; let offsets = rebase_offsets(offsets, elements_range.start)?; // SAFETY: the selected offsets remain monotonically increasing, rebasing them against @@ -312,12 +331,69 @@ impl ListReader { self.elements .projection_evaluation(row_range, &root(), MaskFuture::new_true(row_count)) } + + /// Return an element-row envelope known from write-time list block boundaries. + /// + /// The envelope is used only when the selected outer rows cover at least half of its outer-row + /// block span. This keeps the prefetch bounded for narrow selections while allowing broad and + /// nested reads to issue their element I/O without waiting for offsets first. + fn element_prefetch_range(&self, row_range: &Range) -> Option> { + let boundaries = self.layout.block_boundaries(); + if boundaries.is_empty() || row_range.is_empty() { + return None; + } + + let start_idx = + boundaries.partition_point(|boundary| boundary.outer_row_end() <= row_range.start); + let end_idx = + boundaries.partition_point(|boundary| boundary.outer_row_end() < row_range.end); + let end_boundary = *boundaries.get(end_idx)?; + + let outer_start = start_idx + .checked_sub(1) + .map(|idx| boundaries[idx].outer_row_end()) + .unwrap_or(0); + let element_start = start_idx + .checked_sub(1) + .map(|idx| boundaries[idx].element_row_end()) + .unwrap_or(0); + let outer_end = end_boundary.outer_row_end(); + let selected_len = row_range.end - row_range.start; + let envelope_len = outer_end - outer_start; + if u128::from(selected_len) * 2 < u128::from(envelope_len) { + return None; + } + + Some(element_start..end_boundary.element_row_end()) + } } fn selected_row_range(mask: &Mask) -> Option> { Some(mask.first()?..mask.last()? + 1) } +fn slice_prefetched_elements( + elements: ArrayRef, + prefetched_range: &Range, + exact_range: &Range, +) -> VortexResult { + if exact_range.start < prefetched_range.start || exact_range.end > prefetched_range.end { + vortex_bail!( + "Exact list element range {:?} falls outside prefetched range {:?}", + exact_range, + prefetched_range + ); + } + + let start = usize::try_from(exact_range.start - prefetched_range.start)?; + let end = usize::try_from(exact_range.end - prefetched_range.start)?; + if start == 0 && end == elements.len() { + Ok(elements) + } else { + elements.slice(start..end) + } +} + fn create_validity(validity_array: Option, nullability: Nullability) -> Validity { match validity_array { Some(arr) => Validity::Array(arr), @@ -353,9 +429,32 @@ impl LayoutReader for ListReader { ) -> VortexResult<()> { split_range.check_bounds(self.layout.row_count())?; + let row_range = split_range.row_range(); + let block_boundaries = self.layout.block_boundaries(); + if !block_boundaries.is_empty() { + for boundary in block_boundaries { + let split = boundary.outer_row_end(); + if split <= row_range.start { + continue; + } + if split >= row_range.end { + break; + } + splits.push( + split_range + .row_offset() + .checked_add(split) + .vortex_expect("List layout split offset overflow"), + ); + } + splits.push(split_range.root_row_range().end); + return Ok(()); + } + // Splits are difficult to calculate because all children live in different row coordinate spaces. // List elements typically comprise the majority of the data in a list, and validity/offsets can be treated - // as metadata. We therefore want to parallelize the scan based on element work. + // as metadata. Legacy list layouts do not carry exact block boundaries, so retain the + // element-weighted heuristic for those files. // // Scan splits must be expressed in the list layout's outer-row space, but the elements child // reports its natural boundaries in element-row space. So we translate the element splits using a @@ -370,7 +469,6 @@ impl LayoutReader for ListReader { &mut element_splits, )?; - let row_range = split_range.row_range(); let mut last_split = None; for element_split in element_splits.into_sorted_deduped() { let Some(split) = map_element_split_to_outer_grid( @@ -605,6 +703,7 @@ mod tests { use rstest::rstest; use vortex_array::ArrayContext; use vortex_array::arrays::BoolArray; + use vortex_array::arrays::ChunkedArray; use vortex_array::arrays::ListArray; use vortex_array::arrays::PrimitiveArray; use vortex_array::assert_arrays_eq; @@ -825,6 +924,16 @@ mod tests { ListLayoutStrategy::default() } + fn chunk_preserving_list_strategy() -> ListLayoutStrategy { + ListLayoutStrategy::default() + .with_elements(Arc::new(ChunkedLayoutStrategy::new( + FlatLayoutStrategy::default(), + ))) + .with_offsets(Arc::new(ChunkedLayoutStrategy::new( + FlatLayoutStrategy::default(), + ))) + } + fn layout_test_session() -> VortexSession { vortex_array::array_session() .with::() @@ -1039,29 +1148,37 @@ mod tests { } #[tokio::test] - async fn nested_list_propagates_element_splits() -> VortexResult<()> { - let inner = ListArray::try_new( - PrimitiveArray::from_iter(0..128_i32).into_array(), - PrimitiveArray::from_iter((0..=8_u32).map(|idx| idx * 16)).into_array(), + async fn list_uses_exact_input_block_splits() -> VortexResult<()> { + let chunk0 = ListArray::try_new( + PrimitiveArray::from_iter(0..32_i32).into_array(), + buffer![0u32, 8, 16, 24, 32].into_array(), Validity::NonNullable, )? .into_array(); - let outer = ListArray::try_new( - inner, - buffer![0u32, 2, 4, 6, 8].into_array(), + let chunk1 = ListArray::try_new( + PrimitiveArray::from_iter(32..96_i32).into_array(), + buffer![0u32, 8, 16, 32, 64].into_array(), Validity::NonNullable, )? .into_array(); + let dtype = chunk0.dtype().clone(); + let list = ChunkedArray::try_new(vec![chunk0, chunk1], dtype)?.into_array(); - let inner_strategy = - ListLayoutStrategy::default().with_elements(chunked_elements_strategy()); - let strategy = ListLayoutStrategy::default().with_elements(Arc::new(inner_strategy)); - let (segments, layout, session) = write_layout(&strategy, outer).await?; + let (segments, layout, session) = + write_layout(&chunk_preserving_list_strategy(), list).await?; let reader = layout.new_reader("".into(), segments, &session, &LayoutReaderContext::new())?; - let splits = SplitBy::Layout.splits(reader.as_ref(), &(0..4), &[FieldMask::All])?; - assert_eq!(splits, vec![0, 1, 2, 3, 4]); + let splits = SplitBy::Layout.splits(reader.as_ref(), &(0..8), &[FieldMask::All])?; + assert_eq!(splits, vec![0, 4, 8]); + + let reader = reader + .as_any() + .downcast_ref::() + .expect("ListReader"); + assert_eq!(reader.element_prefetch_range(&(0..4)), Some(0..32)); + assert_eq!(reader.element_prefetch_range(&(4..8)), Some(32..96)); + assert_eq!(reader.element_prefetch_range(&(1..2)), None); Ok(()) } diff --git a/vortex-layout/src/layouts/list/writer.rs b/vortex-layout/src/layouts/list/writer.rs index f9f0516df23..1b812cfb723 100644 --- a/vortex-layout/src/layouts/list/writer.rs +++ b/vortex-layout/src/layouts/list/writer.rs @@ -27,6 +27,7 @@ use vortex_array::scalar_fn::fns::operators::Operator; use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_bail; +use vortex_error::vortex_err; use vortex_io::kanal_ext::KanalExt; use vortex_io::session::RuntimeSessionExt; use vortex_session::VortexSession; @@ -34,6 +35,7 @@ use vortex_session::VortexSession; use crate::LayoutRef; use crate::LayoutStrategy; use crate::layouts::flat::writer::FlatLayoutStrategy; +use crate::layouts::list::ListBlockBoundary; use crate::layouts::list::ListLayout; use crate::segments::SegmentSinkRef; use crate::sequence::SendableSequentialStream; @@ -196,14 +198,22 @@ impl LayoutStrategy for ListLayoutStrategy { }) .collect(); - let (_, layouts) = try_join(fanout_fut, try_join_all(layout_futures)).await?; + let (block_boundaries, layouts) = + try_join(fanout_fut, try_join_all(layout_futures)).await?; let mut layouts = layouts.into_iter(); let elements_layout = layouts.next().vortex_expect("elements layout present"); let offsets_layout = layouts.next().vortex_expect("offsets layout present"); let validity_layout = is_nullable.then(|| layouts.next().vortex_expect("validity layout present")); - Ok(ListLayout::new(dtype, elements_layout, offsets_layout, validity_layout).into_layout()) + Ok(ListLayout::new_with_boundaries( + dtype, + elements_layout, + offsets_layout, + validity_layout, + block_boundaries, + ) + .into_layout()) } fn buffered_bytes(&self) -> u64 { @@ -226,9 +236,11 @@ async fn transpose_list_column( elements_tx: kanal::AsyncSender, offsets_tx: kanal::AsyncSender, validity_tx: Option>, -) -> VortexResult<()> { +) -> VortexResult> { let mut exec_ctx = session.create_execution_ctx(); + let mut outer_base: u64 = 0; let mut element_base: u64 = 0; + let mut block_boundaries = Vec::new(); let mut first = true; let mut saw_chunk = false; while let Some(chunk) = stream.next().await { @@ -244,7 +256,15 @@ async fn transpose_list_column( let n_elements = elements.len() as u64; let row_count = offsets.len().saturating_sub(1); let offsets = global_offsets(offsets, element_base, first, &mut exec_ctx)?; - element_base += n_elements; + element_base = element_base + .checked_add(n_elements) + .ok_or_else(|| vortex_err!("List element row count overflow"))?; + outer_base = outer_base + .checked_add(u64::try_from(row_count)?) + .ok_or_else(|| vortex_err!("List outer row count overflow"))?; + if row_count != 0 { + block_boundaries.push(ListBlockBoundary::new(outer_base, element_base)); + } first = false; if elements_tx @@ -271,7 +291,7 @@ async fn transpose_list_column( if !saw_chunk { vortex_bail!("ListLayoutStrategy needs at least one chunk"); } - Ok(()) + Ok(block_boundaries) } /// Canonicalize a list-dtype array into [`ListDataParts`]. @@ -359,6 +379,16 @@ mod tests { ListLayoutStrategy::default() } + fn chunk_preserving_list_strategy() -> ListLayoutStrategy { + ListLayoutStrategy::default() + .with_elements(Arc::new(ChunkedLayoutStrategy::new( + FlatLayoutStrategy::default(), + ))) + .with_offsets(Arc::new(ChunkedLayoutStrategy::new( + FlatLayoutStrategy::default(), + ))) + } + async fn write(strategy: &S, array: ArrayRef) -> VortexResult { let session = layout_test_session(); let segments = Arc::new(TestSegments::default()); @@ -541,6 +571,35 @@ mod tests { Ok(()) } + #[tokio::test] + async fn records_input_block_boundaries() -> VortexResult<()> { + let chunk0 = ListArray::try_new( + buffer![1i32, 2, 3].into_array(), + buffer![0u32, 2, 3].into_array(), + Validity::NonNullable, + )? + .into_array(); + let chunk1 = ListArray::try_new( + buffer![4i32, 5, 6, 7].into_array(), + buffer![0u32, 1, 4].into_array(), + Validity::NonNullable, + )? + .into_array(); + let chunked = + ChunkedArray::try_new(vec![chunk0, chunk1], i32_list_dtype(false))?.into_array(); + + let layout = write(&chunk_preserving_list_strategy(), chunked).await?; + let boundaries = layout + .as_::() + .block_boundaries(); + + assert_eq!( + boundaries, + [ListBlockBoundary::new(2, 3), ListBlockBoundary::new(4, 7)] + ); + Ok(()) + } + #[tokio::test] async fn chunked_list_input_with_chunked_strategy_succeeds() -> VortexResult<()> { let chunk0 = ListArray::try_new( From 5a391cf08db4da2edd3c31d726384a272cb5c414 Mon Sep 17 00:00:00 2001 From: Matt Katz Date: Wed, 29 Jul 2026 12:35:51 +0100 Subject: [PATCH 2/2] Decouple list prefetch and scan boundaries Signed-off-by: Matt Katz --- vortex-layout/src/layouts/list/reader.rs | 29 ++++++++++++++++++++---- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/vortex-layout/src/layouts/list/reader.rs b/vortex-layout/src/layouts/list/reader.rs index b787da3d634..9044ccd78dc 100644 --- a/vortex-layout/src/layouts/list/reader.rs +++ b/vortex-layout/src/layouts/list/reader.rs @@ -35,6 +35,7 @@ use crate::LayoutReaderContext; use crate::LayoutReaderRef; use crate::RowSplits; use crate::SplitRange; +use crate::layouts::list::ListBlockBoundary; use crate::layouts::list::ListLayout; use crate::layouts::list::expr::ListChildrenNeeded; use crate::layouts::list::expr::get_necessary_list_children; @@ -372,6 +373,12 @@ fn selected_row_range(mask: &Mask) -> Option> { Some(mask.first()?..mask.last()? + 1) } +fn has_enough_block_splits(block_boundaries: &[ListBlockBoundary]) -> bool { + block_boundaries.len() + >= usize::try_from(MAX_LIST_SPLIT_COUNT) + .vortex_expect("Maximum list split count must fit in usize") +} + fn slice_prefetched_elements( elements: ArrayRef, prefetched_range: &Range, @@ -431,7 +438,7 @@ impl LayoutReader for ListReader { let row_range = split_range.row_range(); let block_boundaries = self.layout.block_boundaries(); - if !block_boundaries.is_empty() { + if has_enough_block_splits(block_boundaries) { for boundary in block_boundaries { let split = boundary.outer_row_end(); if split <= row_range.start { @@ -453,8 +460,8 @@ impl LayoutReader for ListReader { // Splits are difficult to calculate because all children live in different row coordinate spaces. // List elements typically comprise the majority of the data in a list, and validity/offsets can be treated - // as metadata. Legacy list layouts do not carry exact block boundaries, so retain the - // element-weighted heuristic for those files. + // as metadata. Prefer exact block boundaries when they provide sufficient parallelism; + // otherwise retain the element-weighted heuristic to avoid under-filling scan workers. // // Scan splits must be expressed in the list layout's outer-row space, but the elements child // reports its natural boundaries in element-row space. So we translate the element splits using a @@ -1147,8 +1154,20 @@ mod tests { assert_eq!(splits, expected); } + #[test] + fn exact_block_splits_require_enough_parallelism() { + let boundaries = (1..=MAX_LIST_SPLIT_COUNT) + .map(|end| ListBlockBoundary::new(end, end)) + .collect::>(); + + assert!(!has_enough_block_splits( + &boundaries[..boundaries.len() - 1] + )); + assert!(has_enough_block_splits(&boundaries)); + } + #[tokio::test] - async fn list_uses_exact_input_block_splits() -> VortexResult<()> { + async fn list_uses_element_weighted_splits_and_exact_prefetch_boundaries() -> VortexResult<()> { let chunk0 = ListArray::try_new( PrimitiveArray::from_iter(0..32_i32).into_array(), buffer![0u32, 8, 16, 24, 32].into_array(), @@ -1170,7 +1189,7 @@ mod tests { layout.new_reader("".into(), segments, &session, &LayoutReaderContext::new())?; let splits = SplitBy::Layout.splits(reader.as_ref(), &(0..8), &[FieldMask::All])?; - assert_eq!(splits, vec![0, 4, 8]); + assert_eq!(splits, vec![0, 2, 8]); let reader = reader .as_any()