From 631b995610caf1689bd1cc167df1d4a9e4cb685d Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 16:30:37 +0000 Subject: [PATCH 1/4] Retain recently read segments across the chunks of V2 scans A V2 scan reads each chunk through a session of its own, and the file's segment source shares a read only while it is in flight. A stored segment that spans a chunk boundary was therefore read again by every chunk that started after its neighbour had finished, and by every partition and every concurrent scan of the file. TPC-DS q28 on SF1 read 188 MB where V1 reads 114 MB, every segment V1 reads up to two and a half times as often. The file's ScanCache now retains up to 64 MiB of recently read host segments in least-recently-used order, and every scan of the file reads through it. ScanCache::with_retained_bytes sets the budget; zero disables it. q28 now reads 125 MB. TPC-DS SF1 on DataFusion, V2 vs V1 geomean: 1.052 -> 1.002. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LYor8ydYMRcjUSh622VTEC Signed-off-by: Claude --- vortex-layout/src/scan/v2/cache.rs | 25 ++++- vortex-layout/src/scan/v2/mod.rs | 7 +- vortex-layout/src/scan/v2/retain.rs | 150 ++++++++++++++++++++++++++++ vortex-layout/src/scan/v2/tests.rs | 53 +++++++++- 4 files changed, 227 insertions(+), 8 deletions(-) create mode 100644 vortex-layout/src/scan/v2/retain.rs diff --git a/vortex-layout/src/scan/v2/cache.rs b/vortex-layout/src/scan/v2/cache.rs index c621bf19289..b83901e8c7d 100644 --- a/vortex-layout/src/scan/v2/cache.rs +++ b/vortex-layout/src/scan/v2/cache.rs @@ -12,9 +12,12 @@ use vortex_array::expr::BoundExpression; use vortex_array::expr::ExactBoundExpr; use super::io::SegmentRanges; +use super::retain::RetainedSegments; +use super::retain::RetainingSegmentSource; use crate::plan::physical::PlanRef; use crate::plan::physical::pipeline::ValueCache; use crate::scan::execution::FilterPlans; +use crate::segments::SegmentSource; /// Reusable state for repeated V2 scans of one file, the V2 counterpart of a cached /// [`LayoutReader`](crate::LayoutReader) tree. @@ -25,7 +28,8 @@ use crate::scan::execution::FilterPlans; /// scans. The lowered plan keeps the children it has initialized, each distinct projection keeps /// its optimized plan, each distinct filter keeps its conjunct and pruning plans along with the /// zones those plans have already excluded, and dictionary values stay keyed by the cached plan -/// that produced them. +/// that produced them. Recently read segments are retained up to a byte budget, so the chunks of +/// the file's scans that share a stored segment read it once. /// /// A cache describes one file's layout, segment locations, and segment source. Use a new cache /// when any of them changes. Cached plans and values live as long as the cache, one set per @@ -44,9 +48,20 @@ struct Inner { /// excluded, so sharing the plans prunes a file's zones once for every scan of that filter. filters: Mutex>>, values: ValueCache, + /// Recently read segments, shared by the chunks of every scan of the file. + retained: Arc, } impl ScanCache { + /// A cache whose scans retain up to `bytes` of recently read segments, so chunks that share + /// a stored segment read it once. The default retains 64 MiB; zero retains nothing. + pub fn with_retained_bytes(bytes: u64) -> Self { + Self(Arc::new(Inner { + retained: Arc::new(RetainedSegments::new(bytes)), + ..Inner::default() + })) + } + /// The lowered root plan an earlier scan kept. pub(super) fn root(&self) -> Option { self.0.root.lock().clone() @@ -112,6 +127,14 @@ impl ScanCache { ) } + /// Serves `segments` through the segments the file's scans retain. + pub(super) fn retaining(&self, segments: Arc) -> Arc { + Arc::new(RetainingSegmentSource { + inner: segments, + retained: Arc::clone(&self.0.retained), + }) + } + pub(super) fn values(&self) -> ValueCache { self.0.values.clone() } diff --git a/vortex-layout/src/scan/v2/mod.rs b/vortex-layout/src/scan/v2/mod.rs index 1f1566ed175..d2ec2156ae4 100644 --- a/vortex-layout/src/scan/v2/mod.rs +++ b/vortex-layout/src/scan/v2/mod.rs @@ -18,11 +18,12 @@ //! serialized bytes per row are high. //! //! A [`ScanFile`] carrying a [`ScanCache`] shares its lowered plan, filter plans and pruned -//! zones, segment index, and dictionary values with every other scan of that file, like a cached -//! layout reader does for V1 scans. +//! zones, segment index, dictionary values, and recently read segments with every other scan of +//! that file, like a cached layout reader does for V1 scans. mod cache; pub(crate) mod io; +mod retain; #[cfg(test)] mod tests; @@ -206,7 +207,7 @@ impl ScanBuilder { session: self.session, rows: self.row_range.unwrap_or(0..row_count), selection: self.selection, - segments: file.segments, + segments: cache.retaining(file.segments), ranges, cache, locations: file.locations, diff --git a/vortex-layout/src/scan/v2/retain.rs b/vortex-layout/src/scan/v2/retain.rs new file mode 100644 index 00000000000..5f50c4b80df --- /dev/null +++ b/vortex-layout/src/scan/v2/retain.rs @@ -0,0 +1,150 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright the Vortex contributors + +//! Keeps recently read segments so the chunks of a scan that share a segment read it once. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use futures::FutureExt; +use futures::TryFutureExt; +use parking_lot::Mutex; +use rustc_hash::FxHashMap; +use vortex_array::buffer::BufferHandle; + +use crate::segments::SegmentFuture; +use crate::segments::SegmentId; +use crate::segments::SegmentSource; + +/// The host bytes of recently read segments, evicting the least recently used past a budget. +/// +/// A V2 scan cuts its rows into chunks that each read through a session of their own, and a +/// stored segment that spans a chunk boundary is read by every chunk it overlaps. The file's segment +/// source shares a read only while it is in flight, so a chunk that starts after its neighbour +/// finished would read the segment again. Neighbouring chunks run close together, so a small +/// budget is enough to read each such segment once. +pub(super) struct RetainedSegments { + budget: u64, + state: Mutex, +} + +/// The bytes a file's scans retain by default. +const DEFAULT_RETAINED_BYTES: u64 = 64 << 20; + +impl Default for RetainedSegments { + fn default() -> Self { + Self::new(DEFAULT_RETAINED_BYTES) + } +} + +#[derive(Default)] +struct State { + bytes: u64, + tick: u64, + /// Each retained segment and the tick of its last use. + segments: FxHashMap, + /// Retained segments by the tick of their last use, oldest first. + by_use: BTreeMap, +} + +impl RetainedSegments { + pub(super) fn new(budget: u64) -> Self { + Self { + budget, + state: Mutex::default(), + } + } + + fn get(&self, id: SegmentId) -> Option { + let mut state = self.state.lock(); + state.tick += 1; + let tick = state.tick; + let (buffer, used) = state.segments.get_mut(&id)?; + let buffer = buffer.clone(); + let previous = std::mem::replace(used, tick); + state.by_use.remove(&previous); + state.by_use.insert(tick, id); + Some(buffer) + } + + fn insert(&self, id: SegmentId, buffer: &BufferHandle) { + let len = buffer.len() as u64; + if !buffer.is_on_host() || len > self.budget / 4 { + return; + } + let mut state = self.state.lock(); + if state.segments.contains_key(&id) { + return; + } + state.tick += 1; + let tick = state.tick; + state.segments.insert(id, (buffer.clone(), tick)); + state.by_use.insert(tick, id); + state.bytes += len; + while state.bytes > self.budget { + let Some((_, oldest)) = state.by_use.pop_first() else { + break; + }; + if let Some((evicted, _)) = state.segments.remove(&oldest) { + state.bytes -= evicted.len() as u64; + } + } + } +} + +/// Serves segments from [`RetainedSegments`] before reading them from `inner`, and retains +/// what it reads. +pub(super) struct RetainingSegmentSource { + pub(super) inner: Arc, + pub(super) retained: Arc, +} + +impl SegmentSource for RetainingSegmentSource { + fn request(&self, id: SegmentId) -> SegmentFuture { + if let Some(buffer) = self.retained.get(id) { + return futures::future::ready(Ok(buffer)).boxed(); + } + let retained = Arc::clone(&self.retained); + self.inner + .request(id) + .inspect_ok(move |buffer| retained.insert(id, buffer)) + .boxed() + } +} + +#[cfg(test)] +mod tests { + use vortex_array::buffer::BufferHandle; + use vortex_buffer::ByteBuffer; + + use super::RetainedSegments; + use crate::segments::SegmentId; + + fn buffer(len: usize) -> BufferHandle { + BufferHandle::new_host(ByteBuffer::from(vec![0u8; len])) + } + + #[test] + fn evicts_least_recently_used_past_budget() { + let retained = RetainedSegments::new(400); + retained.insert(SegmentId::from(0), &buffer(100)); + retained.insert(SegmentId::from(1), &buffer(100)); + retained.insert(SegmentId::from(2), &buffer(100)); + // Using segment 0 makes segment 1 the least recently used. + assert!(retained.get(SegmentId::from(0)).is_some()); + retained.insert(SegmentId::from(3), &buffer(100)); + retained.insert(SegmentId::from(4), &buffer(100)); + + assert!(retained.get(SegmentId::from(1)).is_none()); + for id in [0, 2, 3, 4] { + assert!(retained.get(SegmentId::from(id)).is_some(), "segment {id}"); + } + } + + #[test] + fn skips_segments_over_a_quarter_of_the_budget() { + let retained = RetainedSegments::new(400); + retained.insert(SegmentId::from(0), &buffer(101)); + assert!(retained.get(SegmentId::from(0)).is_none()); + } +} diff --git a/vortex-layout/src/scan/v2/tests.rs b/vortex-layout/src/scan/v2/tests.rs index eff68dfe996..52fca956267 100644 --- a/vortex-layout/src/scan/v2/tests.rs +++ b/vortex-layout/src/scan/v2/tests.rs @@ -550,7 +550,10 @@ async fn dictionary_values_are_reused_between_splits() -> VortexResult<()> { reads: AtomicUsize::new(0), }); let segments = Arc::clone(&counted) as Arc; - let scan = Arc::new(builder(&session, &segments, &layout)?.prepare()?); + let mut file = scan_file(&segments, &layout)?; + // Without retained segments, so a chunk reads every codes chunk it overlaps. + file.cache = Some(ScanCache::with_retained_bytes(0)); + let scan = Arc::new(ScanBuilder::new(session.clone(), file).prepare()?); let chunks = row_chunks(scan.rows().clone(), scan.chunk_rows); assert_eq!( chunks.len(), @@ -583,6 +586,45 @@ async fn dictionary_values_are_reused_between_splits() -> VortexResult<()> { Ok(()) } +/// Work chunks that each overlap half of a stored segment read it once while the file's cache +/// retains segments, and once per chunk when it retains nothing. +#[rstest] +#[case::retained(None, 1)] +#[case::not_retained(Some(0), 2)] +#[tokio::test] +async fn chunks_sharing_a_segment_read_it_once( + #[case] retained_bytes: Option, + #[case] reads_per_segment: usize, +) -> VortexResult<()> { + let session = new_session().with_tokio(); + let (segments, layout) = write(&session, chunked_flat().as_ref()).await?; + let counted = Arc::new(CountingSegments { + inner: segments, + reads: AtomicUsize::new(0), + }); + let segments = Arc::clone(&counted) as Arc; + let mut file = scan_file(&segments, &layout)?; + file.cache = + Some(retained_bytes.map_or_else(ScanCache::default, ScanCache::with_retained_bytes)); + let projection = get_item("ints", root()).bind(layout.dtype())?; + let mut scan = ScanBuilder::new(session, file) + .with_projection(projection) + .prepare()?; + let chunk_rows = CHUNK_ROWS / 2; + scan.chunk_rows = chunk_rows; + let scan = Arc::new(scan); + + // One chunk at a time, so no read is still in flight when the next chunk starts. + for rows in row_chunks(scan.rows().clone(), chunk_rows) { + run_rows(&scan, rows).await?; + } + assert_eq!( + counted.reads.load(Ordering::Relaxed), + CHUNKS * reads_per_segment + ); + Ok(()) +} + /// V2 rejects list layouts: their element ranges are only known once their offsets are read. #[tokio::test] async fn list_layouts_return_an_error() -> VortexResult<()> { @@ -704,7 +746,8 @@ async fn scan_cache_reuses_values_between_scans( }); let segments = Arc::clone(&counted) as Arc; let mut file = scan_file(&segments, &layout)?; - file.cache = cached.then(ScanCache::default); + // Without retained segments, so only the values come from the cache. + file.cache = cached.then(|| ScanCache::with_retained_bytes(0)); let mut reads = Vec::new(); for _ in 0..2 { @@ -762,7 +805,8 @@ async fn repeated_projections_reuse_their_values() -> VortexResult<()> { }); let segments = Arc::clone(&counted) as Arc; let mut file = scan_file(&segments, &layout)?; - file.cache = Some(ScanCache::default()); + // Without retained segments, so only the values come from the cache. + file.cache = Some(ScanCache::with_retained_bytes(0)); let projection = gt(root(), lit(15i32)).bind(layout.dtype())?; let mut reads = Vec::new(); @@ -847,7 +891,8 @@ async fn shared_cache_prunes_zones_once() -> VortexResult<()> { }); let segments = Arc::clone(&counted) as Arc; let mut file = scan_file(&segments, &layout)?; - let cache = ScanCache::default(); + // Without retained segments, so only the proved zones come from the cache. + let cache = ScanCache::with_retained_bytes(0); file.cache = Some(cache.clone()); let filter = gt(get_item("ints", root()), lit(2_999i32)).bind(layout.dtype())?; let projection = get_item("ints", root()).bind(layout.dtype())?; From df8c0d77053494aa66f731f6c2d0bc9a80466d29 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 14:05:58 +0000 Subject: [PATCH 2/4] Default datafusion-bench to the V2 scan for TPC-DS and ClickBench TPC-H already defaults to the planner-driven V2 scan. Extend the default to TPC-DS and ClickBench so the SQL benchmarks measure V2 on them too. `--scan-v2 false` or `VORTEX_SCAN_V2=0` still selects V1. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LYor8ydYMRcjUSh622VTEC Signed-off-by: Claude --- benchmarks/datafusion-bench/src/main.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/benchmarks/datafusion-bench/src/main.rs b/benchmarks/datafusion-bench/src/main.rs index 1a83d374fcb..3d3983f43ee 100644 --- a/benchmarks/datafusion-bench/src/main.rs +++ b/benchmarks/datafusion-bench/src/main.rs @@ -120,7 +120,7 @@ struct Args { #[arg(long, value_delimiter = ',', value_parser = value_parser!(Format))] formats: Vec, - /// Use the planner-driven V2 Vortex scan (defaults to true for TPC-H). + /// Use the planner-driven V2 Vortex scan (defaults to true for TPC-H, TPC-DS and ClickBench). #[arg(long, action = clap::ArgAction::Set)] scan_v2: Option, @@ -137,8 +137,10 @@ async fn main() -> anyhow::Result<()> { let args = Args::parse(); let opts = Opts::from(args.options); let scan_v2 = args.scan_v2.unwrap_or_else(|| { - matches!(args.benchmark, BenchmarkArg::TpcH) - && !std::env::var("VORTEX_SCAN_V2").is_ok_and(|v| v == "0") + matches!( + args.benchmark, + BenchmarkArg::TpcH | BenchmarkArg::TpcDS | BenchmarkArg::ClickBench + ) && !std::env::var("VORTEX_SCAN_V2").is_ok_and(|v| v == "0") }); set_join_set_tracer(get_static_tracer())?; From 8b2cbfa16a8f87f737a175c118d5545f276608c9 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 26 Oct 2026 23:53:14 +0000 Subject: [PATCH 3/4] Share a file's V2 scan plans between separately bound filters The V2 scan cache keyed filter and projection plans by `ExactBoundExpr`, which compares children by pointer. DataFusion opens each partition of a file separately and binds its predicate and projection anew, so no partition ever found the plans another had built: every partition lowered its own filter plans, optimized its own projection and re-evaluated the zone maps. The cost scales with the partition count, so TPC-DS q09 and q28 ran twice as long as V1 on the 32-core benchmark runners. Key the cache by structure instead. Hashing walks the scalar functions and skips dtypes, so a lookup does not walk a wide schema; equality still compares dtypes. Both walk the tree iteratively. Locally (4 cores, 32 partitions), q09/q13/q28/q44/q48 go from a 1.41x geomean of V1 to 1.15x, and q28 from 1.57x to 1.15x. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LYor8ydYMRcjUSh622VTEC Signed-off-by: Claude --- vortex-layout/src/scan/v2/cache.rs | 120 ++++++++++++++++++++++++++--- vortex-layout/src/scan/v2/tests.rs | 14 ++-- 2 files changed, 116 insertions(+), 18 deletions(-) diff --git a/vortex-layout/src/scan/v2/cache.rs b/vortex-layout/src/scan/v2/cache.rs index b83901e8c7d..53b6051686c 100644 --- a/vortex-layout/src/scan/v2/cache.rs +++ b/vortex-layout/src/scan/v2/cache.rs @@ -3,13 +3,14 @@ //! State derived from a [`ScanFile`](super::ScanFile) that repeated V2 scans can share. +use std::hash::Hash; +use std::hash::Hasher; use std::sync::Arc; use std::sync::Weak; use parking_lot::Mutex; use rustc_hash::FxHashMap; use vortex_array::expr::BoundExpression; -use vortex_array::expr::ExactBoundExpr; use super::io::SegmentRanges; use super::retain::RetainedSegments; @@ -42,11 +43,11 @@ struct Inner { root: Mutex>, /// Optimized plans of non-identity projections. Values are cached by the plan that produced /// them, so a projection must reuse its plan for its values to be found again. - projections: Mutex>, + projections: Mutex>, ranges: Mutex>, - /// Plans of filters, keyed by the exact filter expression. A pruning plan remembers the zones it + /// Plans of filters, keyed by the filter expression. A pruning plan remembers the zones it /// excluded, so sharing the plans prunes a file's zones once for every scan of that filter. - filters: Mutex>>, + filters: Mutex>>, values: ValueCache, /// Recently read segments, shared by the chunks of every scan of the file. retained: Arc, @@ -77,7 +78,7 @@ impl ScanCache { self.0 .projections .lock() - .get(&ExactBoundExpr(projection.clone())) + .get(&ExprKey(projection.clone())) .cloned() } @@ -87,7 +88,7 @@ impl ScanCache { self.0 .projections .lock() - .entry(ExactBoundExpr(projection)) + .entry(ExprKey(projection)) .or_insert(plan) .clone() } @@ -104,11 +105,7 @@ impl ScanCache { /// The plans of `filter` an earlier scan kept. pub(super) fn filter_plans(&self, filter: &BoundExpression) -> Option> { - self.0 - .filters - .lock() - .get(&ExactBoundExpr(filter.clone())) - .cloned() + self.0.filters.lock().get(&ExprKey(filter.clone())).cloned() } /// Keeps the plans of `filter`, returning the plans already kept if another scan got there @@ -122,7 +119,7 @@ impl ScanCache { self.0 .filters .lock() - .entry(ExactBoundExpr(filter)) + .entry(ExprKey(filter)) .or_insert(plans), ) } @@ -145,6 +142,47 @@ impl ScanCache { } } +/// A bound expression keyed by structure, so equal expressions bound separately share an entry: +/// an engine that opens a file once per partition binds its filter and projection once per +/// partition. +/// +/// Hashing skips dtypes, which equality still compares, so a lookup does not walk a wide schema. +/// Both walk the tree iteratively, however deep it is. +#[derive(Clone)] +struct ExprKey(BoundExpression); + +impl PartialEq for ExprKey { + fn eq(&self, other: &Self) -> bool { + let mut pending = vec![(&self.0, &other.0)]; + while let Some((lhs, rhs)) = pending.pop() { + if std::ptr::eq(lhs, rhs) { + continue; + } + if lhs.as_scalar() != rhs.as_scalar() + || lhs.children().len() != rhs.children().len() + || lhs.dtype() != rhs.dtype() + { + return false; + } + pending.extend(lhs.children().iter().zip(rhs.children())); + } + true + } +} + +impl Eq for ExprKey {} + +impl Hash for ExprKey { + fn hash(&self, state: &mut H) { + let mut pending = vec![&self.0]; + while let Some(expr) = pending.pop() { + expr.as_scalar().hash(state); + expr.children().len().hash(state); + pending.extend(expr.children()); + } + } +} + /// A [`ScanCache`] handle that does not keep the cache alive. Engines that open a file once per /// partition keep one per file, so concurrent partitions share a cache that is freed with the /// last scan holding it. @@ -157,3 +195,61 @@ impl WeakScanCache { self.0.upgrade().map(ScanCache) } } + +#[cfg(test)] +mod tests { + use std::hash::BuildHasher; + + use rstest::rstest; + use vortex_array::dtype::DType; + use vortex_array::dtype::Nullability; + use vortex_array::dtype::PType; + use vortex_array::expr::Expression; + use vortex_array::expr::get_item; + use vortex_array::expr::gt; + use vortex_array::expr::lit; + use vortex_array::expr::lt; + use vortex_array::expr::root; + use vortex_error::VortexResult; + + use super::ExprKey; + + fn ints_gt(value: i32) -> Expression { + gt(get_item("ints", root()), lit(value)) + } + + fn table(ints: PType) -> DType { + DType::struct_( + [("ints", DType::Primitive(ints, Nullability::NonNullable))], + Nullability::NonNullable, + ) + } + + #[test] + fn separately_bound_equal_expressions_share_a_key() -> VortexResult<()> { + let dtype = table(PType::I32); + let lhs = ExprKey(ints_gt(5).bind(&dtype)?); + let rhs = ExprKey(ints_gt(5).bind(&dtype)?); + assert!(lhs == rhs); + assert_eq!( + rustc_hash::FxBuildHasher.hash_one(&lhs), + rustc_hash::FxBuildHasher.hash_one(&rhs) + ); + Ok(()) + } + + #[rstest] + #[case::literal(ints_gt(6), table(PType::I32))] + #[case::function(lt(get_item("ints", root()), lit(5i32)), table(PType::I32))] + #[case::dtype(get_item("ints", root()), table(PType::I64))] + fn different_expressions_have_different_keys( + #[case] other: Expression, + #[case] dtype: DType, + ) -> VortexResult<()> { + let i32s = table(PType::I32); + for key in [ints_gt(5), get_item("ints", root())] { + assert!(ExprKey(key.bind(&i32s)?) != ExprKey(other.bind(&dtype)?)); + } + Ok(()) + } +} diff --git a/vortex-layout/src/scan/v2/tests.rs b/vortex-layout/src/scan/v2/tests.rs index 52fca956267..30ea1bfa035 100644 --- a/vortex-layout/src/scan/v2/tests.rs +++ b/vortex-layout/src/scan/v2/tests.rs @@ -880,7 +880,8 @@ async fn pruned_rows_get_no_chunk( } /// Scans of one file that share a cache share their filter plans, so the pruning plans run once: -/// the second scan with the same filter re-reads the data it needs but never the zone maps. +/// the second scan with the same filter re-reads the data it needs but never the zone maps. Each +/// scan binds its own filter and projection, as each partition of an engine does. #[tokio::test] async fn shared_cache_prunes_zones_once() -> VortexResult<()> { let session = new_session().with_tokio(); @@ -894,22 +895,23 @@ async fn shared_cache_prunes_zones_once() -> VortexResult<()> { // Without retained segments, so only the proved zones come from the cache. let cache = ScanCache::with_retained_bytes(0); file.cache = Some(cache.clone()); - let filter = gt(get_item("ints", root()), lit(2_999i32)).bind(layout.dtype())?; - let projection = get_item("ints", root()).bind(layout.dtype())?; - assert!(cache.filter_plans(&filter).is_none()); + let filter = || gt(get_item("ints", root()), lit(2_999i32)).bind(layout.dtype()); + let projection = || get_item("ints", root()).bind(layout.dtype()); + assert!(cache.filter_plans(&filter()?).is_none()); let mut reads = Vec::new(); for _ in 0..2 { let before = counted.reads.load(Ordering::Relaxed); + let projection = projection()?; let scan = ScanBuilder::new(session.clone(), file.clone()) .with_projection(projection.clone()) - .with_some_filter(Some(filter.clone())); + .with_some_filter(Some(filter()?)); let actual = collect_scan(projection.dtype().clone(), scan).await?; assert_eq!(actual.len(), 1_000); reads.push(counted.reads.load(Ordering::Relaxed) - before); } let plans = cache - .filter_plans(&filter) + .filter_plans(&filter()?) .expect("the first scan kept its filter plans"); assert_eq!(plans.pruning().count(), 1); // Each scan reads the one chunk pruning keeps; only the first reads the zone map. From 65bf7e98e32c3d0ba1fa6bdb637e1d9869329d01 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 27 Oct 2026 00:22:55 +0000 Subject: [PATCH 4/4] Join short edge chunks of a V2 scan to their neighbours A V2 scan cuts its rows on a fixed grid in the file's row domain, but DataFusion's partition edges fall off that grid. Every partition of a filtered scan then planned a sliver at its edge: TPC-DS q28 at 32 partitions cut each 90,112-row partition into chunks of 81,920 and 8,192 rows. The sliver costs a whole chunk's pipeline compile and conjunct setup, which dominates a selective filter over few rows. The chunk planner now joins an edge chunk under half a chunk to its neighbour; interior boundaries stay on the grid. Locally at 32 partitions, q28's V2 CPU time falls from 1.32x V1 to 1.10x and its wall time matches V1. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LYor8ydYMRcjUSh622VTEC Signed-off-by: Claude --- vortex-layout/src/scan/execution/chunk.rs | 21 ++++++++++++++++++++- vortex-layout/src/scan/execution/mod.rs | 2 ++ vortex-layout/src/scan/v2/tests.rs | 19 +++++++++++++++++++ 3 files changed, 41 insertions(+), 1 deletion(-) diff --git a/vortex-layout/src/scan/execution/chunk.rs b/vortex-layout/src/scan/execution/chunk.rs index 382425d9bb0..a20d855a8e6 100644 --- a/vortex-layout/src/scan/execution/chunk.rs +++ b/vortex-layout/src/scan/execution/chunk.rs @@ -7,6 +7,7 @@ use std::ops::Range; use std::sync::Arc; use std::vec::IntoIter; +use vortex_error::VortexExpect; use vortex_error::VortexResult; use vortex_io::request::IoConsumer; use vortex_io::request::IoRequestId; @@ -58,6 +59,24 @@ pub(crate) fn row_chunks(rows: Range, rows_per_chunk: usize) -> Vec, rows_per_chunk: usize) -> Vec> { + let mut chunks = row_chunks(rows, rows_per_chunk); + let short = |chunk: &Range| chunk.end - chunk.start < rows_per_chunk as u64 / 2; + if chunks.len() > 1 && chunks.last().is_some_and(short) { + let last = chunks.pop().vortex_expect("more than one chunk"); + chunks.last_mut().vortex_expect("more than one chunk").end = last.end; + } + if chunks.len() > 1 && short(&chunks[0]) { + let first = chunks.remove(0); + chunks[0].start = first.start; + } + chunks +} + /// Emits one planner per chunk of a scan's rows, in row order, each spawned as independent so /// the driver may run it beside the others. A chunk none of whose rows is selected and kept is /// never planned. @@ -79,7 +98,7 @@ impl ChunkPlanner { selection: Selection, kept: Option, ) -> Self { - let chunks = row_chunks(rows.clone(), scan.chunk_rows).into_iter(); + let chunks = scan_chunks(rows.clone(), scan.chunk_rows).into_iter(); Self { scan, rows, diff --git a/vortex-layout/src/scan/execution/mod.rs b/vortex-layout/src/scan/execution/mod.rs index 0185f61d6a9..1411b96f01e 100644 --- a/vortex-layout/src/scan/execution/mod.rs +++ b/vortex-layout/src/scan/execution/mod.rs @@ -37,6 +37,8 @@ pub(crate) use self::chunk::ROW_BLOCK_SIZE; #[cfg(test)] pub(crate) use self::chunk::row_chunks; pub(crate) use self::chunk::rows_per_chunk; +#[cfg(test)] +pub(crate) use self::chunk::scan_chunks; use self::filter::FilterPlanner; pub use self::filter::FilterPlans; use self::projection::ProjectionPlanner; diff --git a/vortex-layout/src/scan/v2/tests.rs b/vortex-layout/src/scan/v2/tests.rs index 30ea1bfa035..25d91195b84 100644 --- a/vortex-layout/src/scan/v2/tests.rs +++ b/vortex-layout/src/scan/v2/tests.rs @@ -73,6 +73,7 @@ use crate::layouts::zoned::writer::ZonedStrategy; use crate::scan::arrow::to_record_batch; use crate::scan::execution::SegmentLocation; use crate::scan::execution::row_chunks; +use crate::scan::execution::scan_chunks; use crate::scan::scan_builder::ScanBuilder as V1ScanBuilder; use crate::segments::SegmentFuture; use crate::segments::SegmentId; @@ -479,6 +480,24 @@ fn fixed_row_chunks(#[case] rows: Range, #[case] expected: Vec>) assert_eq!(row_chunks(rows, 65_536), expected); } +/// A scan joins a chunk at either edge of its rows that is under half a chunk to its neighbour, +/// so a partition whose edges fall off the chunk grid plans no sliver of a chunk. +#[rstest] +#[case::partition_tail(0..90_112, 81_920, vec![0..90_112])] +#[case::partition_head(90_112..180_224, 81_920, vec![90_112..180_224])] +#[case::long_head_kept(10..140_000, 65_536, vec![10..65_536, 65_536..140_000])] +#[case::both_edges(60_000..140_000, 65_536, vec![60_000..140_000])] +#[case::interior_kept(60_000..270_000, 65_536, vec![60_000..131_072, 131_072..196_608, 196_608..270_000])] +#[case::single(5..100, 65_536, vec![5..100])] +#[case::empty(5..5, 65_536, vec![])] +fn scans_join_short_edge_chunks( + #[case] rows: Range, + #[case] rows_per_chunk: usize, + #[case] expected: Vec>, +) { + assert_eq!(scan_chunks(rows, rows_per_chunk), expected); +} + #[rstest] #[case::empty(0, 0, 65_536)] #[case::single_chunk(1000, 100_000_000, 65_536)]