From a3b7c5a0d511ad9c4d39552980a885824da2f6b2 Mon Sep 17 00:00:00 2001 From: Matt Katz Date: Wed, 26 Aug 2026 16:06:05 -0400 Subject: [PATCH 1/2] feat(layout): prototype separate scan splits Signed-off-by: "Matt Katz" --- vortex-layout/Cargo.toml | 4 + .../benches/filter_projection_splits.rs | 232 +++++++++++++ vortex-layout/src/scan/repeated_scan.rs | 324 ++++++++++++++++-- vortex-layout/src/scan/scan_builder.rs | 59 +++- vortex-layout/src/scan/splits.rs | 6 + vortex-layout/src/scan/tasks.rs | 46 ++- 6 files changed, 627 insertions(+), 44 deletions(-) create mode 100644 vortex-layout/benches/filter_projection_splits.rs diff --git a/vortex-layout/Cargo.toml b/vortex-layout/Cargo.toml index 5d4a62d86e7..4cf97b45d38 100644 --- a/vortex-layout/Cargo.toml +++ b/vortex-layout/Cargo.toml @@ -76,6 +76,10 @@ wasm-bindgen = ["dep:uuid"] name = "zone_map_prune" harness = false +[[bench]] +name = "filter_projection_splits" +harness = false + [target.'cfg(all(target_arch = "wasm32", target_os = "unknown"))'.dependencies] uuid = { workspace = true, features = ["js"], optional = true } diff --git a/vortex-layout/benches/filter_projection_splits.rs b/vortex-layout/benches/filter_projection_splits.rs new file mode 100644 index 00000000000..c6b2f436416 --- /dev/null +++ b/vortex-layout/benches/filter_projection_splits.rs @@ -0,0 +1,232 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +#![expect(clippy::unwrap_used)] + +//! Measures scans whose filter and projection fields have opposing physical chunk granularities. + +use std::sync::Arc; +use std::sync::LazyLock; +use std::sync::atomic::AtomicUsize; +use std::sync::atomic::Ordering; + +use async_trait::async_trait; +use divan::Bencher; +use parking_lot::Mutex; +use tokio::runtime::Runtime; +use vortex_array::ArrayContext; +use vortex_array::IntoArray; +use vortex_array::array_session; +use vortex_array::arrays::PrimitiveArray; +use vortex_array::arrays::StructArray; +use vortex_array::buffer::BufferHandle; +use vortex_array::expr::eq; +use vortex_array::expr::get_item; +use vortex_array::expr::lit; +use vortex_array::expr::root; +use vortex_array::expr::select; +use vortex_array::stream::ArrayStreamExt; +use vortex_buffer::ByteBuffer; +use vortex_buffer::ByteBufferMut; +use vortex_error::VortexExpect; +use vortex_error::VortexResult; +use vortex_error::vortex_err; +use vortex_io::session::RuntimeSession; +use vortex_io::session::RuntimeSessionExt; +use vortex_layout::LayoutReaderContext; +use vortex_layout::LayoutReaderRef; +use vortex_layout::LayoutStrategy; +use vortex_layout::layouts::chunked::writer::ChunkedLayoutStrategy; +use vortex_layout::layouts::flat::writer::FlatLayoutStrategy; +use vortex_layout::layouts::repartition::RepartitionStrategy; +use vortex_layout::layouts::repartition::RepartitionWriterOptions; +use vortex_layout::layouts::struct_::StructStrategy; +use vortex_layout::scan::scan_builder::ScanBuilder; +use vortex_layout::segments::SegmentFuture; +use vortex_layout::segments::SegmentId; +use vortex_layout::segments::SegmentSink; +use vortex_layout::segments::SegmentSource; +use vortex_layout::sequence::SequenceId; +use vortex_layout::sequence::SequentialArrayStreamExt; +use vortex_layout::session::LayoutSession; +use vortex_session::VortexSession; + +fn main() { + divan::main(); +} + +const ROW_COUNT: usize = 1_048_576; +const FINE_CHUNK_ROWS: usize = 4_096; +const COARSE_CHUNK_ROWS: usize = 524_288; +const EXPECTED_ROWS: usize = ROW_COUNT.div_ceil(10); + +static RUNTIME: LazyLock = LazyLock::new(|| { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build() + .unwrap() +}); + +static SESSION: LazyLock = LazyLock::new(|| { + let _guard = RUNTIME.enter(); + array_session() + .with::() + .with::() + .with_tokio() +}); + +#[derive(Clone, Default)] +struct CountingSegments { + segments: Arc>>, + requests: Arc, +} + +impl CountingSegments { + fn reset_requests(&self) { + self.requests.store(0, Ordering::Relaxed); + } + + fn request_count(&self) -> usize { + self.requests.load(Ordering::Relaxed) + } +} + +impl SegmentSource for CountingSegments { + fn request(&self, id: SegmentId) -> SegmentFuture { + self.requests.fetch_add(1, Ordering::Relaxed); + let buffer = self.segments.lock().get(*id as usize).cloned(); + Box::pin(async move { + buffer + .map(BufferHandle::new_host) + .ok_or_else(|| vortex_err!("Segment not found")) + }) + } +} + +#[async_trait] +impl SegmentSink for CountingSegments { + async fn write( + &self, + _sequence_id: SequenceId, + buffers: Vec, + ) -> VortexResult { + let mut buffer = ByteBufferMut::empty(); + for part in buffers { + buffer.extend_from_slice(part.as_ref()); + } + + let mut segments = self.segments.lock(); + let id = SegmentId::from(u32::try_from(segments.len()).vortex_expect("Too many segments")); + segments.push(buffer.freeze()); + Ok(id) + } +} + +struct Fixture { + reader: LayoutReaderRef, + segments: Arc, +} + +static FILTER_FINE: LazyLock = + LazyLock::new(|| make_fixture(FINE_CHUNK_ROWS, COARSE_CHUNK_ROWS)); +static PROJECTION_FINE: LazyLock = + LazyLock::new(|| make_fixture(COARSE_CHUNK_ROWS, FINE_CHUNK_ROWS)); + +fn chunked_strategy(rows_per_chunk: usize) -> Arc { + Arc::new(RepartitionStrategy::new( + ChunkedLayoutStrategy::new(FlatLayoutStrategy::default()), + RepartitionWriterOptions { + block_size_minimum: 0, + block_len_multiple: rows_per_chunk, + block_size_target: None, + canonicalize: false, + }, + )) +} + +fn make_fixture(filter_chunk_rows: usize, projection_chunk_rows: usize) -> Fixture { + let filter = PrimitiveArray::from_iter((0..ROW_COUNT).map(|idx| (idx % 10) as i64)); + let projected = PrimitiveArray::from_iter((0..ROW_COUNT).map(|idx| idx as i64)); + let array = StructArray::try_from_iter([ + ("filter", filter.into_array()), + ("projected", projected.into_array()), + ]) + .unwrap() + .into_array(); + + let flat: Arc = Arc::new(FlatLayoutStrategy::default()); + let strategy = StructStrategy::new(Arc::clone(&flat), flat) + .with_field_writer("filter", chunked_strategy(filter_chunk_rows)) + .with_field_writer("projected", chunked_strategy(projection_chunk_rows)); + + let segments = Arc::new(CountingSegments::default()); + let (ptr, eof) = SequenceId::root().split(); + let layout = RUNTIME + .block_on(strategy.write_stream( + ArrayContext::empty().into(), + Arc::::clone(&segments), + array.to_array_stream().sequenced(ptr), + eof, + &SESSION, + )) + .unwrap(); + let reader = layout + .new_reader( + "filter-projection-splits".into(), + Arc::::clone(&segments), + &SESSION, + &LayoutReaderContext::new(), + ) + .unwrap(); + + Fixture { reader, segments } +} + +fn scan(fixture: &Fixture, separate_splits: bool) -> VortexResult<(usize, usize)> { + fixture.segments.reset_requests(); + let dtype = fixture.reader.dtype(); + let filter = eq(get_item("filter", root()), lit(0_i64)) + .optimize_recursive(dtype)? + .bind(dtype)?; + let projection = select(["projected"], root()) + .optimize_recursive(dtype)? + .bind(dtype)?; + let result = RUNTIME.block_on( + ScanBuilder::new(SESSION.clone(), Arc::clone(&fixture.reader)) + .with_filter(filter) + .with_projection(projection) + .with_separate_filter_projection_splits(separate_splits) + .into_array_stream()? + .read_all(), + )?; + Ok((result.len(), fixture.segments.request_count())) +} + +fn run(bencher: Bencher, fixture: &Fixture, separate_splits: bool, label: &str) { + let (rows, requests) = scan(fixture, separate_splits).unwrap(); + assert_eq!(rows, EXPECTED_ROWS); + eprintln!("{label}: rows={rows}, segment_requests={requests}"); + + bencher.bench_local(|| divan::black_box(scan(fixture, separate_splits).unwrap())); +} + +#[divan::bench(sample_count = 20)] +fn coupled_filter_fine_projection_coarse(bencher: Bencher) { + run(bencher, &FILTER_FINE, false, "coupled/filter-fine"); +} + +#[divan::bench(sample_count = 20)] +fn coupled_filter_coarse_projection_fine(bencher: Bencher) { + run(bencher, &PROJECTION_FINE, false, "coupled/projection-fine"); +} + +#[divan::bench(sample_count = 20)] +fn separate_filter_fine_projection_coarse(bencher: Bencher) { + run(bencher, &FILTER_FINE, true, "separate/filter-fine"); +} + +#[divan::bench(sample_count = 20)] +fn separate_filter_coarse_projection_fine(bencher: Bencher) { + run(bencher, &PROJECTION_FINE, true, "separate/projection-fine"); +} diff --git a/vortex-layout/src/scan/repeated_scan.rs b/vortex-layout/src/scan/repeated_scan.rs index 413761b8103..f3807ec5269 100644 --- a/vortex-layout/src/scan/repeated_scan.rs +++ b/vortex-layout/src/scan/repeated_scan.rs @@ -6,8 +6,11 @@ use std::iter; use std::ops::Range; use std::sync::Arc; -use futures::Stream; +use async_stream::try_stream; +use futures::StreamExt; +use futures::TryStreamExt; use futures::future::BoxFuture; +use futures::stream::BoxStream; use itertools::Either; use itertools::Itertools; use vortex_array::ArrayRef; @@ -19,8 +22,11 @@ use vortex_array::stream::ArrayStream; use vortex_array::stream::ArrayStreamAdapter; use vortex_error::VortexExpect; use vortex_error::VortexResult; +use vortex_error::vortex_ensure; use vortex_io::runtime::BlockingRuntime; use vortex_io::session::RuntimeSessionExt; +use vortex_mask::Mask; +use vortex_scan::row_mask::RowMask; use vortex_scan::selection::Selection; use vortex_session::VortexSession; use vortex_utils::parallelism::get_available_parallelism; @@ -29,6 +35,8 @@ use crate::LayoutReaderRef; use crate::scan::filter::FilterExpr; use crate::scan::splits::Splits; use crate::scan::tasks::TaskContext; +use crate::scan::tasks::filter_exec; +use crate::scan::tasks::project_exec; use crate::scan::tasks::split_exec; /// A projected subset (by indices, range, and filter) of rows from a Vortex data source. @@ -123,40 +131,18 @@ impl RepeatedScan { &self, row_range: Option>, ) -> VortexResult>>>> { - let selection_range: Option> = match &self.selection { - Selection::IncludeByIndex(buf) if !buf.is_empty() => { - Some(buf[0]..buf[buf.len() - 1] + 1) - } - Selection::IncludeRoaring(roaring) if !roaring.is_empty() => { - Some(roaring.min().vortex_expect("empty")..roaring.max().vortex_expect("empty") + 1) - } - _ => None, - }; - let row_range = intersect_ranges(self.row_range.as_ref(), row_range); - let row_range = intersect_ranges(row_range.as_ref(), selection_range); + let row_range = self.effective_row_range(row_range); let ranges = match &self.splits { - Splits::Natural(vec) => { - debug_assert!(vec.is_sorted()); - let splits_iter = match row_range { - None => Either::Left(vec.iter().copied()), - Some(range) => { - if range.is_empty() { - return Ok(Vec::new()); - } - let lo = vec.partition_point(|&x| x <= range.start); - let hi = vec.partition_point(|&x| x < range.end); - Either::Right( - iter::once(range.start) - .chain(vec[lo..hi].iter().copied()) - .chain(iter::once(range.end)), - ) - } - }; - - Either::Left(splits_iter.tuple_windows().map(|(start, end)| start..end)) + Splits::Natural(boundaries) => { + Either::Left(natural_ranges(boundaries, row_range.as_ref()).into_iter()) } - Splits::Ranges(ranges) => Either::Right(match row_range { + // `execute_stream` uses the staged path. Keep `execute` correct for callers that ask + // for the task list directly, using projection boundaries as coupled splits. + Splits::FilterProjection { projection, .. } => { + Either::Left(natural_ranges(projection, row_range.as_ref()).into_iter()) + } + Splits::Ranges(ranges) => Either::Right(match row_range.as_ref() { None => Either::Left(ranges.iter().cloned()), Some(range) => { if range.is_empty() { @@ -195,11 +181,18 @@ impl RepeatedScan { Ok(tasks) } + pub(crate) fn has_separate_filter_projection_splits(&self) -> bool { + matches!(self.splits, Splits::FilterProjection { .. }) + } + pub fn execute_stream( &self, row_range: Option>, - ) -> VortexResult> + Send + 'static + use> { - use futures::StreamExt; + ) -> VortexResult>> { + if let Splits::FilterProjection { filter, projection } = &self.splits { + return self.execute_filter_projection_stream(row_range, filter, projection); + } + let num_workers = get_available_parallelism().unwrap_or(1); let concurrency = self.concurrency * num_workers; let handle = self.session.handle(); @@ -213,8 +206,201 @@ impl RepeatedScan { stream.buffer_unordered(concurrency).boxed() }; - Ok(stream.filter_map(|chunk| async move { chunk.transpose() })) + Ok(stream + .filter_map(|chunk| async move { chunk.transpose() }) + .boxed()) + } + + fn execute_filter_projection_stream( + &self, + row_range: Option>, + filter_boundaries: &[u64], + projection_boundaries: &[u64], + ) -> VortexResult>> { + let row_range = self.effective_row_range(row_range); + let filter_ranges = natural_ranges(filter_boundaries, row_range.as_ref()); + let projection_ranges = natural_ranges(projection_boundaries, row_range.as_ref()); + + if filter_ranges.is_empty() || projection_ranges.is_empty() { + return Ok(futures::stream::empty().boxed()); + } + + let ctx = Arc::new(TaskContext { + filter: self.filter.clone().map(|f| Arc::new(FilterExpr::new(f))), + reader: Arc::clone(&self.layout_reader), + projection: self.projection.clone(), + mapper: Arc::clone(&self.map_fn), + }); + + // Build filter evaluations eagerly so the readers can register all filter I/O before the + // tasks begin making progress. Buffering remains ordered because projection masks cannot + // be assembled until all preceding filter ranges are known. + let mut filter_tasks = Vec::with_capacity(filter_ranges.len()); + for range in filter_ranges { + let row_mask = self.selection.row_mask(&range); + filter_tasks.push(filter_exec(Arc::clone(&ctx), row_mask)?); + } + + let num_workers = get_available_parallelism().unwrap_or(1); + let concurrency = self.concurrency * num_workers; + let handle = self.session.handle(); + let filter_handle = handle.clone(); + let filtered_masks = futures::stream::iter(filter_tasks) + .map(move |task: BoxFuture<'static, VortexResult>| filter_handle.spawn(task)) + .buffered(concurrency) + .boxed(); + + let projection_tasks = try_stream! { + let mut repartitioner = ProjectionMaskRepartitioner::new(projection_ranges); + futures::pin_mut!(filtered_masks); + + while let Some(filtered_mask) = filtered_masks.try_next().await? { + for projection_mask in repartitioner.push(filtered_mask)? { + if !projection_mask.mask().all_false() { + yield project_exec(Arc::clone(&ctx), projection_mask)?; + } + } + } + + repartitioner.finish()?; + }; + + let projection_tasks = projection_tasks + .map_ok(move |task: BoxFuture<'static, VortexResult>>| handle.spawn(task)); + let projected = if self.ordered { + projection_tasks.try_buffered(concurrency).boxed() + } else { + projection_tasks.try_buffer_unordered(concurrency).boxed() + }; + + Ok(projected + .filter_map(|chunk| async move { chunk.transpose() }) + .boxed()) } + + fn effective_row_range(&self, row_range: Option>) -> Option> { + let selection_range: Option> = match &self.selection { + Selection::IncludeByIndex(buf) if !buf.is_empty() => { + Some(buf[0]..buf[buf.len() - 1] + 1) + } + Selection::IncludeRoaring(roaring) if !roaring.is_empty() => { + Some(roaring.min().vortex_expect("empty")..roaring.max().vortex_expect("empty") + 1) + } + _ => None, + }; + let row_range = intersect_ranges(self.row_range.as_ref(), row_range); + intersect_ranges(row_range.as_ref(), selection_range) + } +} + +struct ProjectionMaskRepartitioner { + ranges: std::vec::IntoIter>, + current: Option>, + next_row: Option, + fragments: Vec, +} + +impl ProjectionMaskRepartitioner { + fn new(ranges: Vec>) -> Self { + let mut ranges = ranges.into_iter(); + let current = ranges.next(); + let next_row = current.as_ref().map(|range| range.start); + Self { + ranges, + current, + next_row, + fragments: Vec::new(), + } + } + + fn push(&mut self, filtered: RowMask) -> VortexResult> { + let filtered_range = filtered.row_range(); + vortex_ensure!( + self.next_row == Some(filtered_range.start), + "non-contiguous filter mask: expected row {:?}, got {}", + self.next_row, + filtered_range.start + ); + + let mut completed = Vec::new(); + let mut source_start = 0; + while source_start < filtered.mask().len() { + let projection_range = self + .current + .as_ref() + .vortex_expect("filter masks exceed projection ranges"); + let next_row = self.next_row.vortex_expect("current projection range"); + vortex_ensure!( + next_row >= projection_range.start && next_row < projection_range.end, + "invalid projection mask cursor {next_row} for range {projection_range:?}" + ); + + let projection_remaining = usize::try_from(projection_range.end - next_row)?; + let fragment_len = projection_remaining.min(filtered.mask().len() - source_start); + self.fragments.push( + filtered + .mask() + .slice(source_start..source_start + fragment_len), + ); + source_start += fragment_len; + let next_row = next_row + fragment_len as u64; + self.next_row = Some(next_row); + + if next_row == projection_range.end { + let projection_start = projection_range.start; + let mask = if self.fragments.len() == 1 { + self.fragments.pop().vortex_expect("one mask fragment") + } else { + Mask::from_iter(std::mem::take(&mut self.fragments)) + }; + completed.push(RowMask::new(projection_start, mask)); + + self.current = self.ranges.next(); + self.next_row = self.current.as_ref().map(|range| range.start); + if let Some(next_projection) = &self.current { + vortex_ensure!( + next_projection.start == next_row, + "non-contiguous projection ranges: expected {next_row}, got {}", + next_projection.start + ); + } + } + } + + Ok(completed) + } + + fn finish(self) -> VortexResult<()> { + vortex_ensure!( + self.current.is_none() && self.fragments.is_empty(), + "filter masks ended before projection ranges" + ); + Ok(()) + } +} + +fn natural_ranges(boundaries: &[u64], row_range: Option<&Range>) -> Vec> { + debug_assert!(boundaries.is_sorted()); + let splits_iter = match row_range { + None => Either::Left(boundaries.iter().copied()), + Some(range) => { + if range.is_empty() { + return Vec::new(); + } + let lo = boundaries.partition_point(|&x| x <= range.start); + let hi = boundaries.partition_point(|&x| x < range.end); + Either::Right( + iter::once(range.start) + .chain(boundaries[lo..hi].iter().copied()) + .chain(iter::once(range.end)), + ) + } + }; + + splits_iter + .tuple_windows() + .map(|(start, end)| start..end) + .collect() } fn intersect_ranges(left: Option<&Range>, right: Option>) -> Option> { @@ -225,3 +411,71 @@ fn intersect_ranges(left: Option<&Range>, right: Option>) -> Opt (Some(l), Some(r)) => Some(cmp::max(l.start, r.start)..cmp::min(l.end, r.end)), } } + +#[cfg(test)] +mod tests { + use vortex_error::VortexResult; + use vortex_mask::Mask; + use vortex_scan::row_mask::RowMask; + + use super::ProjectionMaskRepartitioner; + use super::natural_ranges; + + #[test] + fn splits_one_filter_mask_across_projection_ranges() -> VortexResult<()> { + let input = Mask::from_iter([ + true, false, true, true, false, false, true, false, false, true, + ]); + let mut repartitioner = ProjectionMaskRepartitioner::new(vec![2..5, 5..9, 9..12]); + + let output = repartitioner.push(RowMask::new(2, input))?; + repartitioner.finish()?; + + assert_eq!(output.len(), 3); + assert_eq!(output[0].row_range(), 2..5); + assert_eq!(output[0].mask(), &Mask::from_iter([true, false, true])); + assert_eq!(output[1].row_range(), 5..9); + assert_eq!( + output[1].mask(), + &Mask::from_iter([true, false, false, true]) + ); + assert_eq!(output[2].row_range(), 9..12); + assert_eq!(output[2].mask(), &Mask::from_iter([false, false, true])); + Ok(()) + } + + #[test] + fn combines_filter_masks_into_projection_ranges() -> VortexResult<()> { + let mut repartitioner = ProjectionMaskRepartitioner::new(vec![2..8, 8..12]); + let mut output = Vec::new(); + + output.extend(repartitioner.push(RowMask::new(2, Mask::from_iter([true, false])))?); + output.extend(repartitioner.push(RowMask::new(4, Mask::from_iter([true, true, false])))?); + output.extend(repartitioner.push(RowMask::new( + 7, + Mask::from_iter([false, true, false, false, true]), + ))?); + repartitioner.finish()?; + + assert_eq!(output.len(), 2); + assert_eq!(output[0].row_range(), 2..8); + assert_eq!( + output[0].mask(), + &Mask::from_iter([true, false, true, true, false, false]) + ); + assert_eq!(output[1].row_range(), 8..12); + assert_eq!( + output[1].mask(), + &Mask::from_iter([true, false, false, true]) + ); + Ok(()) + } + + #[test] + fn natural_ranges_are_clipped_without_gaps() { + assert_eq!( + natural_ranges(&[0, 4, 9, 12], Some(&(2..10))), + [2..4, 4..9, 9..10] + ); + } +} diff --git a/vortex-layout/src/scan/scan_builder.rs b/vortex-layout/src/scan/scan_builder.rs index f7abbb21fb2..3e3619a5489 100644 --- a/vortex-layout/src/scan/scan_builder.rs +++ b/vortex-layout/src/scan/scan_builder.rs @@ -45,6 +45,8 @@ use crate::scan::split_by::SplitBy; use crate::scan::splits::Splits; use crate::scan::splits::attempt_split_ranges; +const SEPARATE_SCAN_SPLITS_ENV: &str = "VORTEX_EXPERIMENTAL_SEPARATE_SCAN_SPLITS"; + /// Builder for scanning a [`LayoutReader`] into arrays, streams, iterators, or mapped outputs. /// /// A scan has three independent row restriction mechanisms: @@ -72,6 +74,8 @@ pub struct ScanBuilder { /// Precomputed full-file natural split boundaries; when set, [`prepare`](Self::prepare) /// uses them instead of walking the layout. natural_splits: Option>, + /// Whether filtered scans use independent physical filter and projection splits. + separate_filter_projection_splits: bool, /// The number of splits to make progress on concurrently **per-thread**. concurrency: usize, /// Function to apply to each [`ArrayRef`] within the spawned split tasks. @@ -100,6 +104,8 @@ impl ScanBuilder { selection: Default::default(), split_by: SplitBy::Layout, natural_splits: None, + separate_filter_projection_splits: std::env::var(SEPARATE_SCAN_SPLITS_ENV) + .is_ok_and(|value| value == "1"), // We default to four tasks per worker thread, which allows for some I/O lookahead // without too much impact on work-stealing. concurrency: 4, @@ -210,6 +216,17 @@ impl ScanBuilder { self } + /// Configure filtered streaming scans to evaluate filters and projections over their own + /// physical splits. + /// + /// Exact index ranges continue to use coupled execution. Caller-supplied natural splits are + /// ignored because a single boundary set cannot represent both physical split domains. + /// This is also enabled by setting `VORTEX_EXPERIMENTAL_SEPARATE_SCAN_SPLITS=1`. + pub fn with_separate_filter_projection_splits(mut self, enabled: bool) -> Self { + self.separate_filter_projection_splits = enabled; + self + } + /// Compute the full-file natural split boundaries for the fields referenced by this scan's /// projection and filter, ignoring any configured row range. /// @@ -288,6 +305,7 @@ impl ScanBuilder { selection: self.selection, split_by: self.split_by, natural_splits: self.natural_splits, + separate_filter_projection_splits: self.separate_filter_projection_splits, concurrency: self.concurrency, metrics_registry: self.metrics_registry, file_stats: self.file_stats, @@ -331,6 +349,25 @@ impl ScanBuilder { let splits = if let Some(ranges) = attempt_split_ranges(&self.selection, self.row_range.as_ref()) { Splits::Ranges(ranges) + } else if self.separate_filter_projection_splits + && let Some(filter) = bound_filter.as_ref() + { + let split_range = self + .row_range + .clone() + .unwrap_or_else(|| 0..layout_reader.row_count()); + let filter_fields = referenced_field_masks(filter, None)?; + let projection_fields = referenced_field_masks(&bound_projection, None)?; + Splits::FilterProjection { + filter: self + .split_by + .splits(layout_reader.as_ref(), &split_range, &filter_fields)? + .into(), + projection: self + .split_by + .splits(layout_reader.as_ref(), &split_range, &projection_fields)? + .into(), + } } else if let Some(boundaries) = self.natural_splits { // Caller-supplied full-file boundaries; execution clamps them to the row range. Splits::Natural(boundaries) @@ -397,13 +434,16 @@ enum LazyScanState { Error(Option), } -type PreparedScanTasks = Vec>>>; +enum PreparedScan { + Tasks(Vec>>>), + Stream(BoxStream<'static, VortexResult>), +} struct PreparingScan { ordered: bool, concurrency: usize, handle: Handle, - task: Task>>, + task: Task>>, } struct LazyScanStream { @@ -432,8 +472,14 @@ impl Stream for LazyScanStream { let num_workers = get_available_parallelism().unwrap_or(1); let concurrency = builder.concurrency * num_workers; let handle = builder.session.handle(); - let task = handle - .spawn_cpu(move || builder.prepare().and_then(|scan| scan.execute(None))); + let task = handle.spawn_cpu(move || { + let scan = builder.prepare()?; + if scan.has_separate_filter_projection_splits() { + Ok(PreparedScan::Stream(scan.execute_stream(None)?)) + } else { + Ok(PreparedScan::Tasks(scan.execute(None)?)) + } + }); self.state = LazyScanState::Preparing(PreparingScan { ordered, concurrency, @@ -443,7 +489,7 @@ impl Stream for LazyScanStream { } LazyScanState::Preparing(preparing) => { match ready!(Pin::new(&mut preparing.task).poll(cx)) { - Ok(tasks) => { + Ok(PreparedScan::Tasks(tasks)) => { let ordered = preparing.ordered; let concurrency = preparing.concurrency; let handle = preparing.handle.clone(); @@ -459,6 +505,9 @@ impl Stream for LazyScanStream { .boxed(); self.state = LazyScanState::Stream(stream); } + Ok(PreparedScan::Stream(stream)) => { + self.state = LazyScanState::Stream(stream) + } Err(err) => self.state = LazyScanState::Error(Some(err)), } } diff --git a/vortex-layout/src/scan/splits.rs b/vortex-layout/src/scan/splits.rs index 2d48ee9e3fd..4cfe29a7344 100644 --- a/vortex-layout/src/scan/splits.rs +++ b/vortex-layout/src/scan/splits.rs @@ -22,6 +22,12 @@ pub enum Splits { /// The boundaries are sorted in ascending order and deduplicated. Natural(Arc<[u64]>), + /// Physical filter and projection boundaries for a staged filtered scan. + FilterProjection { + filter: Arc<[u64]>, + projection: Arc<[u64]>, + }, + /// Exact split ranges. /// /// This is an optimization for when we know the exact rows we need to get from a file (which is diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs index 6fdd3ee0a60..50785a70d5d 100644 --- a/vortex-layout/src/scan/tasks.rs +++ b/vortex-layout/src/scan/tasks.rs @@ -40,10 +40,44 @@ pub fn split_exec( read_mask: RowMask, limit: Option<&mut u64>, ) -> VortexResult>> { + let row_range = read_mask.row_range(); + let filter_mask = filter_mask(Arc::clone(&ctx), &read_mask, limit)?; + project_exec_with_mask(ctx, row_range, filter_mask) +} + +/// Evaluate pruning and the exact filter for one physical filter range. +pub fn filter_exec( + ctx: Arc>, + read_mask: RowMask, +) -> VortexResult> { + let row_offset = read_mask.row_range().start; + let filter_mask = filter_mask(ctx, &read_mask, None)?; + Ok(async move { Ok(RowMask::new(row_offset, filter_mask.await?)) }.boxed()) +} + +/// Project one physical projection range after its exact filter mask is known. +pub fn project_exec( + ctx: Arc>, + read_mask: RowMask, +) -> VortexResult>> { + if read_mask.mask().all_false() { + return Ok(async { Ok(None) }.boxed()); + } + + let row_range = read_mask.row_range(); + let mask = MaskFuture::ready(read_mask.mask().clone()); + project_exec_with_mask(ctx, row_range, mask) +} + +fn filter_mask( + ctx: Arc>, + read_mask: &RowMask, + limit: Option<&mut u64>, +) -> VortexResult { let row_range = read_mask.row_range(); let row_mask = read_mask.mask().clone(); - let filter_mask = match ctx.filter.as_ref() { + Ok(match ctx.filter.as_ref() { // No filter == immediate mask None => { let row_mask = match limit { @@ -68,8 +102,6 @@ pub fn split_exec( // we want to start prefetching the IO for this split. let reader = Arc::clone(&ctx.reader); let filter = Arc::clone(filter); - let row_range = row_range.clone(); - MaskFuture::new(row_mask.len(), async move { let mut mask = row_mask; let mut dynamic_versions = vec![None; filter.conjuncts().len()]; @@ -133,8 +165,14 @@ pub fn split_exec( Ok(mask) }) } - }; + }) +} +fn project_exec_with_mask( + ctx: Arc>, + row_range: std::ops::Range, + filter_mask: MaskFuture, +) -> VortexResult>> { // Step 4: execute the projection, only at the mask for rows which match the filter let projection_future = ctx.reader From 4182f55567e63f59135eaab84a365e9d8f759aba Mon Sep 17 00:00:00 2001 From: Matt Katz Date: Thu, 27 Aug 2026 09:58:52 -0400 Subject: [PATCH 2/2] separate pruning, filter, and projection splits Signed-off-by: Matt Katz --- vortex-layout/src/scan/repeated_scan.rs | 281 +++++++++++++++++------- vortex-layout/src/scan/tasks.rs | 142 +++++++++--- 2 files changed, 312 insertions(+), 111 deletions(-) diff --git a/vortex-layout/src/scan/repeated_scan.rs b/vortex-layout/src/scan/repeated_scan.rs index f3807ec5269..0ad638c601f 100644 --- a/vortex-layout/src/scan/repeated_scan.rs +++ b/vortex-layout/src/scan/repeated_scan.rs @@ -10,10 +10,12 @@ use async_stream::try_stream; use futures::StreamExt; use futures::TryStreamExt; use futures::future::BoxFuture; +use futures::future::try_join_all; use futures::stream::BoxStream; use itertools::Either; use itertools::Itertools; use vortex_array::ArrayRef; +use vortex_array::MaskFuture; use vortex_array::dtype::DType; use vortex_array::expr::BoundExpression; use vortex_array::iter::ArrayIterator; @@ -24,9 +26,9 @@ use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_error::vortex_ensure; use vortex_io::runtime::BlockingRuntime; +use vortex_io::runtime::Handle; use vortex_io::session::RuntimeSessionExt; use vortex_mask::Mask; -use vortex_scan::row_mask::RowMask; use vortex_scan::selection::Selection; use vortex_session::VortexSession; use vortex_utils::parallelism::get_available_parallelism; @@ -35,10 +37,14 @@ use crate::LayoutReaderRef; use crate::scan::filter::FilterExpr; use crate::scan::splits::Splits; use crate::scan::tasks::TaskContext; -use crate::scan::tasks::filter_exec; -use crate::scan::tasks::project_exec; +use crate::scan::tasks::filter_after_pruning; +use crate::scan::tasks::project_exec_with_mask; +use crate::scan::tasks::prune_exec; use crate::scan::tasks::split_exec; +/// Number of pruning-surviving projections to register before exact filters begin polling. +const PROJECTION_REGISTRATION_WINDOW: usize = 16; + /// A projected subset (by indices, range, and filter) of rows from a Vortex data source. /// /// The method of this struct enable, possibly concurrent, scanning of multiple row ranges of this @@ -232,37 +238,66 @@ impl RepeatedScan { mapper: Arc::clone(&self.map_fn), }); - // Build filter evaluations eagerly so the readers can register all filter I/O before the - // tasks begin making progress. Buffering remains ordered because projection masks cannot - // be assembled until all preceding filter ranges are known. - let mut filter_tasks = Vec::with_capacity(filter_ranges.len()); + // Resolve metadata pruning first. Projection I/O is registered only for ranges which + // pruning cannot eliminate; their exact filters remain unresolved so filtering can make + // progress concurrently with those projection reads. + let mut pruning_tasks = Vec::with_capacity(filter_ranges.len()); for range in filter_ranges { let row_mask = self.selection.row_mask(&range); - filter_tasks.push(filter_exec(Arc::clone(&ctx), row_mask)?); + pruning_tasks.push(prune_exec(Arc::clone(&ctx), row_mask)?); } let num_workers = get_available_parallelism().unwrap_or(1); let concurrency = self.concurrency * num_workers; let handle = self.session.handle(); - let filter_handle = handle.clone(); - let filtered_masks = futures::stream::iter(filter_tasks) - .map(move |task: BoxFuture<'static, VortexResult>| filter_handle.spawn(task)) + let pruning_handle = handle.clone(); + let pruned_masks = futures::stream::iter(pruning_tasks) + .map(move |task| pruning_handle.spawn(task)) .buffered(concurrency) .boxed(); + let registration_batch_size = cmp::min(PROJECTION_REGISTRATION_WINDOW, concurrency).max(1); + let projection_handle = handle.clone(); let projection_tasks = try_stream! { - let mut repartitioner = ProjectionMaskRepartitioner::new(projection_ranges); - futures::pin_mut!(filtered_masks); + let mut repartitioner = ProjectionMaskRepartitioner::new( + projection_ranges, + projection_handle, + ); + let mut pending = Vec::with_capacity(registration_batch_size); + futures::pin_mut!(pruned_masks); + + while let Some(pruned) = pruned_masks.try_next().await? { + let row_range = pruned.row_range(); + let pruning_mask = pruned.mask().clone(); + let filter_mask = filter_after_pruning(Arc::clone(&ctx), pruned)?; + let filter_mask = PrunedFilterMask { + row_range, + pruning_mask, + filter_mask, + }; + + for projection_mask in repartitioner.push(filter_mask)? { + if projection_mask.pruning_mask.all_false() { + continue; + } - while let Some(filtered_mask) = filtered_masks.try_next().await? { - for projection_mask in repartitioner.push(filtered_mask)? { - if !projection_mask.mask().all_false() { - yield project_exec(Arc::clone(&ctx), projection_mask)?; + pending.push(project_exec_with_mask( + Arc::clone(&ctx), + projection_mask.row_range, + projection_mask.filter_mask, + )?); + if pending.len() == registration_batch_size { + for task in pending.drain(..) { + yield task; + } } } } repartitioner.finish()?; + for task in pending { + yield task; + } }; let projection_tasks = projection_tasks @@ -297,11 +332,25 @@ struct ProjectionMaskRepartitioner { ranges: std::vec::IntoIter>, current: Option>, next_row: Option, - fragments: Vec, + pruning_fragments: Vec, + filter_fragments: Vec, + handle: Handle, +} + +struct PrunedFilterMask { + row_range: Range, + pruning_mask: Mask, + filter_mask: MaskFuture, +} + +struct ProjectionMask { + row_range: Range, + pruning_mask: Mask, + filter_mask: MaskFuture, } impl ProjectionMaskRepartitioner { - fn new(ranges: Vec>) -> Self { + fn new(ranges: Vec>, handle: Handle) -> Self { let mut ranges = ranges.into_iter(); let current = ranges.next(); let next_row = current.as_ref().map(|range| range.start); @@ -309,12 +358,14 @@ impl ProjectionMaskRepartitioner { ranges, current, next_row, - fragments: Vec::new(), + pruning_fragments: Vec::new(), + filter_fragments: Vec::new(), + handle, } } - fn push(&mut self, filtered: RowMask) -> VortexResult> { - let filtered_range = filtered.row_range(); + fn push(&mut self, filtered: PrunedFilterMask) -> VortexResult> { + let filtered_range = filtered.row_range; vortex_ensure!( self.next_row == Some(filtered_range.start), "non-contiguous filter mask: expected row {:?}, got {}", @@ -324,7 +375,7 @@ impl ProjectionMaskRepartitioner { let mut completed = Vec::new(); let mut source_start = 0; - while source_start < filtered.mask().len() { + while source_start < filtered.pruning_mask.len() { let projection_range = self .current .as_ref() @@ -336,24 +387,45 @@ impl ProjectionMaskRepartitioner { ); let projection_remaining = usize::try_from(projection_range.end - next_row)?; - let fragment_len = projection_remaining.min(filtered.mask().len() - source_start); - self.fragments.push( - filtered - .mask() - .slice(source_start..source_start + fragment_len), - ); + let fragment_len = projection_remaining.min(filtered.pruning_mask.len() - source_start); + let fragment_range = source_start..source_start + fragment_len; + self.pruning_fragments + .push(filtered.pruning_mask.slice(fragment_range.clone())); + self.filter_fragments + .push(filtered.filter_mask.slice(fragment_range)); source_start += fragment_len; let next_row = next_row + fragment_len as u64; self.next_row = Some(next_row); if next_row == projection_range.end { - let projection_start = projection_range.start; - let mask = if self.fragments.len() == 1 { - self.fragments.pop().vortex_expect("one mask fragment") + let row_range = projection_range.clone(); + let pruning_mask = if self.pruning_fragments.len() == 1 { + self.pruning_fragments + .pop() + .vortex_expect("one pruning mask fragment") + } else { + Mask::from_iter(std::mem::take(&mut self.pruning_fragments)) + }; + let filter_mask = if self.filter_fragments.len() == 1 { + self.filter_fragments + .pop() + .vortex_expect("one filter mask fragment") } else { - Mask::from_iter(std::mem::take(&mut self.fragments)) + let fragments = std::mem::take(&mut self.filter_fragments); + let handle = self.handle.clone(); + MaskFuture::new(pruning_mask.len(), async move { + let masks = try_join_all( + fragments.into_iter().map(|fragment| handle.spawn(fragment)), + ) + .await?; + Ok(Mask::from_iter(masks)) + }) }; - completed.push(RowMask::new(projection_start, mask)); + completed.push(ProjectionMask { + row_range, + pruning_mask, + filter_mask, + }); self.current = self.ranges.next(); self.next_row = self.current.as_ref().map(|range| range.start); @@ -372,7 +444,9 @@ impl ProjectionMaskRepartitioner { fn finish(self) -> VortexResult<()> { vortex_ensure!( - self.current.is_none() && self.fragments.is_empty(), + self.current.is_none() + && self.pruning_fragments.is_empty() + && self.filter_fragments.is_empty(), "filter masks ended before projection ranges" ); Ok(()) @@ -414,61 +488,108 @@ fn intersect_ranges(left: Option<&Range>, right: Option>) -> Opt #[cfg(test)] mod tests { + use std::sync::Arc; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + + use futures::future::try_join_all; + use vortex_array::MaskFuture; use vortex_error::VortexResult; + use vortex_io::runtime::single::block_on; use vortex_mask::Mask; - use vortex_scan::row_mask::RowMask; use super::ProjectionMaskRepartitioner; + use super::PrunedFilterMask; use super::natural_ranges; #[test] - fn splits_one_filter_mask_across_projection_ranges() -> VortexResult<()> { - let input = Mask::from_iter([ - true, false, true, true, false, false, true, false, false, true, - ]); - let mut repartitioner = ProjectionMaskRepartitioner::new(vec![2..5, 5..9, 9..12]); - - let output = repartitioner.push(RowMask::new(2, input))?; - repartitioner.finish()?; - - assert_eq!(output.len(), 3); - assert_eq!(output[0].row_range(), 2..5); - assert_eq!(output[0].mask(), &Mask::from_iter([true, false, true])); - assert_eq!(output[1].row_range(), 5..9); - assert_eq!( - output[1].mask(), - &Mask::from_iter([true, false, false, true]) - ); - assert_eq!(output[2].row_range(), 9..12); - assert_eq!(output[2].mask(), &Mask::from_iter([false, false, true])); - Ok(()) + fn splits_shared_filter_future_across_projection_ranges() -> VortexResult<()> { + block_on(|handle| async move { + let evaluations = Arc::new(AtomicUsize::new(0)); + let evaluation_count = Arc::clone(&evaluations); + let filter_mask = MaskFuture::new(10, async move { + evaluation_count.fetch_add(1, Ordering::Relaxed); + Ok(Mask::from_iter([ + true, false, true, true, false, false, true, false, false, true, + ])) + }); + let input = PrunedFilterMask { + row_range: 2..12, + pruning_mask: Mask::from_iter([ + true, true, true, true, false, false, true, true, true, true, + ]), + filter_mask, + }; + let mut repartitioner = + ProjectionMaskRepartitioner::new(vec![2..5, 5..9, 9..12], handle); + + let output = repartitioner.push(input)?; + repartitioner.finish()?; + + assert_eq!(evaluations.load(Ordering::Relaxed), 0); + assert_eq!(output.len(), 3); + assert_eq!(output[0].row_range, 2..5); + assert_eq!(output[0].pruning_mask, Mask::from_iter([true, true, true])); + assert_eq!(output[1].row_range, 5..9); + assert_eq!( + output[1].pruning_mask, + Mask::from_iter([true, false, false, true]) + ); + assert_eq!(output[2].row_range, 9..12); + assert_eq!(output[2].pruning_mask, Mask::from_iter([true, true, true])); + + let masks = try_join_all(output.into_iter().map(|mask| mask.filter_mask)).await?; + assert_eq!(evaluations.load(Ordering::Relaxed), 1); + assert_eq!(masks[0], Mask::from_iter([true, false, true])); + assert_eq!(masks[1], Mask::from_iter([true, false, false, true])); + assert_eq!(masks[2], Mask::from_iter([false, false, true])); + Ok(()) + }) } #[test] - fn combines_filter_masks_into_projection_ranges() -> VortexResult<()> { - let mut repartitioner = ProjectionMaskRepartitioner::new(vec![2..8, 8..12]); - let mut output = Vec::new(); - - output.extend(repartitioner.push(RowMask::new(2, Mask::from_iter([true, false])))?); - output.extend(repartitioner.push(RowMask::new(4, Mask::from_iter([true, true, false])))?); - output.extend(repartitioner.push(RowMask::new( - 7, - Mask::from_iter([false, true, false, false, true]), - ))?); - repartitioner.finish()?; - - assert_eq!(output.len(), 2); - assert_eq!(output[0].row_range(), 2..8); - assert_eq!( - output[0].mask(), - &Mask::from_iter([true, false, true, true, false, false]) - ); - assert_eq!(output[1].row_range(), 8..12); - assert_eq!( - output[1].mask(), - &Mask::from_iter([true, false, false, true]) - ); - Ok(()) + fn combines_filter_futures_into_projection_ranges() -> VortexResult<()> { + block_on(|handle| async move { + let mut repartitioner = ProjectionMaskRepartitioner::new(vec![2..8, 8..12], handle); + let mut output = Vec::new(); + + output.extend(repartitioner.push(PrunedFilterMask { + row_range: 2..4, + pruning_mask: Mask::from_iter([true, true]), + filter_mask: MaskFuture::ready(Mask::from_iter([true, false])), + })?); + output.extend(repartitioner.push(PrunedFilterMask { + row_range: 4..7, + pruning_mask: Mask::from_iter([true, true, false]), + filter_mask: MaskFuture::ready(Mask::from_iter([true, true, false])), + })?); + output.extend(repartitioner.push(PrunedFilterMask { + row_range: 7..12, + pruning_mask: Mask::from_iter([false, true, true, true, true]), + filter_mask: MaskFuture::ready(Mask::from_iter([false, true, false, false, true])), + })?); + repartitioner.finish()?; + + assert_eq!(output.len(), 2); + assert_eq!(output[0].row_range, 2..8); + assert_eq!( + output[0].pruning_mask, + Mask::from_iter([true, true, true, true, false, false]) + ); + assert_eq!(output[1].row_range, 8..12); + assert_eq!( + output[1].pruning_mask, + Mask::from_iter([true, true, true, true]) + ); + + let masks = try_join_all(output.into_iter().map(|mask| mask.filter_mask)).await?; + assert_eq!( + masks[0], + Mask::from_iter([true, false, true, true, false, false]) + ); + assert_eq!(masks[1], Mask::from_iter([true, false, false, true])); + Ok(()) + }) } #[test] diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs index 50785a70d5d..9cb9a4b0d70 100644 --- a/vortex-layout/src/scan/tasks.rs +++ b/vortex-layout/src/scan/tasks.rs @@ -21,6 +21,22 @@ use crate::scan::filter::FilterExpr; pub type TaskFuture = BoxFuture<'static, VortexResult>; +/// The result of metadata pruning for one filter-native row range. +pub(crate) struct PrunedMask { + row_mask: RowMask, + dynamic_versions: Vec>, +} + +impl PrunedMask { + pub(crate) fn row_range(&self) -> std::ops::Range { + self.row_mask.row_range() + } + + pub(crate) fn mask(&self) -> &Mask { + self.row_mask.mask() + } +} + /// Logic for executing a single split reading task. /// N.B. read_mask should be evaluated against all_false() before calling this /// method to avoid creating an empty TaskFuture. @@ -45,28 +61,104 @@ pub fn split_exec( project_exec_with_mask(ctx, row_range, filter_mask) } -/// Evaluate pruning and the exact filter for one physical filter range. -pub fn filter_exec( +/// Resolve metadata pruning without evaluating the exact filter. +pub(crate) fn prune_exec( ctx: Arc>, read_mask: RowMask, -) -> VortexResult> { - let row_offset = read_mask.row_range().start; - let filter_mask = filter_mask(ctx, &read_mask, None)?; - Ok(async move { Ok(RowMask::new(row_offset, filter_mask.await?)) }.boxed()) +) -> VortexResult> { + let row_range = read_mask.row_range(); + let row_offset = row_range.start; + let row_mask = read_mask.mask().clone(); + + let Some(filter) = ctx.filter.as_ref().cloned() else { + return Ok(async move { + Ok(PrunedMask { + row_mask: RowMask::new(row_offset, row_mask), + dynamic_versions: Vec::new(), + }) + } + .boxed()); + }; + + let reader = Arc::clone(&ctx.reader); + Ok(async move { + let mut mask = row_mask; + let mut dynamic_versions = vec![None; filter.conjuncts().len()]; + + for (idx, conjunct) in filter.conjuncts().iter().enumerate() { + if mask.all_false() { + break; + } + + dynamic_versions[idx] = filter.dynamic_updates(idx).map(|du| du.version()); + let conjunct_mask = reader + .pruning_evaluation(&row_range, conjunct, mask.clone())? + .await?; + mask = mask.bitand(&conjunct_mask); + } + + Ok(PrunedMask { + row_mask: RowMask::new(row_offset, mask), + dynamic_versions, + }) + } + .boxed()) } -/// Project one physical projection range after its exact filter mask is known. -pub fn project_exec( +/// Evaluate the exact filter starting from a resolved pruning mask. +pub(crate) fn filter_after_pruning( ctx: Arc>, - read_mask: RowMask, -) -> VortexResult>> { - if read_mask.mask().all_false() { - return Ok(async { Ok(None) }.boxed()); - } + pruned: PrunedMask, +) -> VortexResult { + let row_range = pruned.row_mask.row_range(); + let row_mask = pruned.row_mask.mask().clone(); - let row_range = read_mask.row_range(); - let mask = MaskFuture::ready(read_mask.mask().clone()); - project_exec_with_mask(ctx, row_range, mask) + let Some(filter) = ctx.filter.as_ref().cloned() else { + return Ok(MaskFuture::ready(row_mask)); + }; + + let reader = Arc::clone(&ctx.reader); + Ok(MaskFuture::new(row_mask.len(), async move { + let mut mask = row_mask; + let mut dynamic_versions = pruned.dynamic_versions; + + // Evaluate conjuncts in their learned order, preserving the dynamic-pruning refresh from + // the coupled scan path. + let mut remaining = BitVec::from_elem(filter.conjuncts().len(), true); + while let Some(idx) = filter.next_conjunct(&remaining) { + remaining.set(idx, false); + if mask.all_false() { + return Ok(mask); + } + + let conjunct = &filter.conjuncts()[idx]; + let current_version = filter.dynamic_updates(idx).map(|du| du.version()); + if let Some(dv) = current_version + && dynamic_versions[idx].is_none_or(|v| v < dv) + { + dynamic_versions[idx] = Some(dv); + let conjunct_mask = reader + .pruning_evaluation(&row_range, conjunct, mask.clone())? + .await?; + mask = mask.bitand(&conjunct_mask); + } + if mask.all_false() { + return Ok(mask); + } + + let input_true_count = mask.true_count(); + let conjunct_mask = reader + .filter_evaluation(&row_range, conjunct, MaskFuture::ready(mask))? + .await?; + filter.report_selectivity( + idx, + conditional_selectivity(input_true_count, conjunct_mask.true_count()), + ); + mask = conjunct_mask; + } + + Ok(mask) + })) } fn filter_mask( @@ -97,32 +189,26 @@ fn filter_mask( MaskFuture::ready(row_mask) } Some(filter) => { - // NOTE: it's very important that the pruning and filter evaluations are built OUTSIDE - // the future. Registering these row ranges eagerly is a hint to the IO system that - // we want to start prefetching the IO for this split. + // Keep the coupled path's original scheduling shape. The pruning-first barrier is + // intentionally limited to the experimental separate-splits stream. let reader = Arc::clone(&ctx.reader); let filter = Arc::clone(filter); MaskFuture::new(row_mask.len(), async move { let mut mask = row_mask; let mut dynamic_versions = vec![None; filter.conjuncts().len()]; - // TODO(ngates): we could use FuturedUnordered to intersect the masks in parallel. for (idx, conjunct) in filter.conjuncts().iter().enumerate() { if mask.all_false() { return Ok(mask); } - // Store the latest version of the dynamic expression prior to pruning. - // We will re-run the pruning later if the version has changed in the meantime. dynamic_versions[idx] = filter.dynamic_updates(idx).map(|du| du.version()); - let conjunct_mask = reader .pruning_evaluation(&row_range, conjunct, mask.clone())? .await?; mask = mask.bitand(&conjunct_mask); } - // Now we loop through the conjuncts in the preferred order and evaluate them. let mut remaining = BitVec::from_elem(filter.conjuncts().len(), true); while let Some(idx) = filter.next_conjunct(&remaining) { remaining.set(idx, false); @@ -131,14 +217,10 @@ fn filter_mask( } let conjunct = &filter.conjuncts()[idx]; - - // If the dynamic expression has changed since pruning, re-run the pruning. - // Store the dynamic update once to avoid TOCTOU race condition let current_version = filter.dynamic_updates(idx).map(|du| du.version()); if let Some(dv) = current_version && dynamic_versions[idx].is_none_or(|v| v < dv) { - // The dynamic expression has been updated, re-run the pruning. dynamic_versions[idx] = Some(dv); let conjunct_mask = reader .pruning_evaluation(&row_range, conjunct, mask.clone())? @@ -157,8 +239,6 @@ fn filter_mask( idx, conditional_selectivity(input_true_count, conjunct_mask.true_count()), ); - - // Filter evaluations return a mask already intersected with the input mask. mask = conjunct_mask; } @@ -168,7 +248,7 @@ fn filter_mask( }) } -fn project_exec_with_mask( +pub(crate) fn project_exec_with_mask( ctx: Arc>, row_range: std::ops::Range, filter_mask: MaskFuture,