From 0cb8b3f6783fafe8c3008af7965de5ddf19914ea Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 13:28:24 +0000 Subject: [PATCH 1/2] Eagerly build single-conjunct filter evaluations `split_exec` already builds the projection evaluation outside the returned future, so projection segment reads for every split are registered before any split task is polled and the IO system can coalesce them. The filter evaluation had no such treatment: it was built inside the `MaskFuture`, so a filter over a column that is not projected trickled its reads in one split at a time. The filter evaluation cannot be hoisted in general, because the conjunct order and the mask fed to each conjunct are chosen at runtime from selectivity statistics. When the filter has a single conjunct there is no ordering to choose, so the whole pruning-then-filter chain can be built at task-construction time instead. The pruned mask is awaited before the filter evaluation so that a split which pruning has eliminated entirely still drops (and therefore cancels) its filter reads, and the dynamic-expression re-pruning check is preserved. Measured on TPC-H lineitem with a filter on `l_linenumber` projecting only `l_extendedprice`, pread64 counts drop from 302 to 177 at sf=10 and from 30 to 20 at sf=1. Row counts are unchanged on every query shape measured. Signed-off-by: Claude Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01D6qV3R62EBNgkd2Leq5YqZ --- vortex-layout/src/scan/tasks.rs | 171 +++++++++++++++++++++++--------- 1 file changed, 122 insertions(+), 49 deletions(-) diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs index 218efb64a0d..515efc54393 100644 --- a/vortex-layout/src/scan/tasks.rs +++ b/vortex-layout/src/scan/tasks.rs @@ -4,6 +4,7 @@ //! Split scanning task implementation. use std::ops::BitAnd; +use std::ops::Range; use std::sync::Arc; use bit_vec::BitVec; @@ -70,64 +71,70 @@ pub fn split_exec( 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()]; + // A single-conjunct filter has no adaptive ordering to decide at runtime, so the + // whole evaluation can be built up-front. + if filter.conjuncts().len() == 1 { + single_conjunct_mask(reader, filter, row_range, row_mask)? + } else { + 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()); - // 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); - if mask.all_false() { - return Ok(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())? .await?; mask = mask.bitand(&conjunct_mask); } - if mask.all_false() { - return Ok(mask); - } - let conjunct_mask = reader - .filter_evaluation(&row_range, conjunct, MaskFuture::ready(mask))? - .await?; - filter.report_selectivity(idx, conjunct_mask.density()); + // 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); + if mask.all_false() { + return Ok(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())? + .await?; + mask = mask.bitand(&conjunct_mask); + } + if mask.all_false() { + return Ok(mask); + } - // Filter evaluations return a mask already intersected with the input mask. - mask = conjunct_mask; - } + let conjunct_mask = reader + .filter_evaluation(&row_range, conjunct, MaskFuture::ready(mask))? + .await?; + filter.report_selectivity(idx, conjunct_mask.density()); - Ok(mask) - }) + // Filter evaluations return a mask already intersected with the input mask. + mask = conjunct_mask; + } + + Ok(mask) + }) + } } }; @@ -150,6 +157,72 @@ pub fn split_exec( Ok(array_fut.boxed()) } +/// Builds the filter mask for a filter made up of a single conjunct. +/// +/// With only one conjunct there is no conjunct ordering to decide at runtime, so the whole +/// pruning-then-filter chain can be constructed at task-construction time rather than when the +/// task is first polled. This registers the conjunct's segment reads for every split before any +/// split task runs, which lets the IO system coalesce them into larger reads. +/// +/// It matters most when the filter column is not part of the projection: the projection +/// evaluation is already built eagerly, so a filter over a projected column has its segments +/// registered either way, but a filter over an unprojected column otherwise trickles its reads +/// in one split at a time. +fn single_conjunct_mask( + reader: Arc, + filter: Arc, + row_range: Range, + row_mask: Mask, +) -> VortexResult { + let len = row_mask.len(); + let conjunct = filter.conjuncts()[0].clone(); + + // Store the latest version of the dynamic expression prior to pruning. We re-run the pruning + // if the version has changed by the time the task is polled. + let dynamic_version = filter.dynamic_updates(0).map(|du| du.version()); + let pruning_eval = reader.pruning_evaluation(&row_range, &conjunct, row_mask.clone())?; + + let pruned = MaskFuture::new(len, { + let reader = Arc::clone(&reader); + let filter = Arc::clone(&filter); + let conjunct = conjunct.clone(); + let row_range = row_range.clone(); + async move { + let mut mask = row_mask.bitand(&pruning_eval.await?); + + // If the dynamic expression has changed since pruning, re-run the pruning. + let current_version = filter.dynamic_updates(0).map(|du| du.version()); + if let Some(dv) = current_version + && dynamic_version.is_none_or(|v| v < dv) + && !mask.all_false() + { + let conjunct_mask = reader + .pruning_evaluation(&row_range, &conjunct, mask.clone())? + .await?; + mask = mask.bitand(&conjunct_mask); + } + + Ok(mask) + } + }); + + let filter_eval = reader.filter_evaluation(&row_range, &conjunct, pruned.clone())?; + + Ok(MaskFuture::new(len, async move { + // Awaiting the pruned mask first lets us drop the filter evaluation, cancelling its + // reads, when pruning has already eliminated the entire split. + let pruned = pruned.await?; + if pruned.all_false() { + return Ok(pruned); + } + + // Filter evaluations return a mask already intersected with the input mask. + let mask = filter_eval.await?; + filter.report_selectivity(0, mask.density()); + Ok(mask) + })) +} + /// Information needed to execute a single split task. /// /// Row selection is evaluated before creating a split task so it's not included From 8cef37d6b595f5659419d23bc27e932ede40b3ae Mon Sep 17 00:00:00 2001 From: Joe Isaacs Date: Mon, 10 Aug 2026 16:11:16 +0100 Subject: [PATCH 2/2] Chain filter evaluations for every conjunct (#9282) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Rationale for this change Stacked on #9279 — that PR handles a single conjunct as a special case, this one generalises it to any number and deletes the special case. **Draft, and not mergeable as it stands: it is a large regression on prunable queries** (`lineitem_prune` +354%, `lineitem_and` +193%). Opening it because the mechanism works exactly as intended for the case it targets, and the regression isolates a specific, fixable blocker. Detail below. `LayoutReader::filter_evaluation` registers its segment reads when it is *called*, but only awaits its input mask when it is *polled*. The existing loop builds one evaluation, awaits it fully, then builds the next — so reads trickle out one conjunct at a time, per split, and nothing can be coalesced. The trait documentation already describes the intended alternative: > It is recommended to defer awaiting the input mask for as long as possible (ideally, after all I/O is complete). This allows other conjuncts the opportunity to refine the mask as much as possible before it is used. That only makes sense if several conjuncts' evaluations are built and in flight at once, which the caller never did. ## What changes are included in this PR? `chained_filter_mask` replaces both the single-conjunct helper from #9279 and the multi-conjunct loop. Net −31 lines. Each conjunct's output `MaskFuture` is fed straight into the next at construction time, so the reads for the whole chain are registered up front while each conjunct still receives the mask its predecessor refined — no extra compute, and the `EXPR_EVAL_THRESHOLD` low-density path still applies. The evaluation order is drained from `FilterExpr::next_conjunct` up front rather than re-queried between conjuncts. That is safe: `next_conjunct` (`scan/filter.rs:93-97`) reads a precomputed `ordering` vector that is only recomputed inside `report_selectivity`, from histograms accumulated *across* splits. Within a single split the order was already fixed. Ordering still adapts across splits. The `all_false` short circuit between filter evaluations is deliberately **not** carried over — see below. ### Results TPC-H `lineitem`, warm page cache, local NVMe. Row counts identical across all three variants on every shape. `pread64` counts: | Query | sf=1 base | sf=1 #9279 | sf=1 chain | sf=10 base | sf=10 #9279 | sf=10 chain | | --- | --- | --- | --- | --- | --- | --- | | `lineitem_filter_only` | 30 | 20 | **20** | 302 | 177 | **177** | | `lineitem` | 14 | 14 | 14 | 109 | 109 | 109 | | `lineitem_and` (2 conjuncts) | 13 | 13 | 14 | 14 | 14 | **109** | | `lineitem_prune` | 8 | 8 | 14 | 9 | 9 | **91** | | `lineitem_wide` | 114 | 113 | 117 | 1165 | 1172 | 1163 | Execution time at sf=10, median of 7 interleaved rounds, each round the median of 5 executions: | Query | base (ms) | chain (ms) | change | | --- | --- | --- | --- | | `lineitem_filter_only` | 140.3 | 151.8 | +8.2% | | `lineitem` | 91.5 | 109.7 | +20.0% | | `lineitem_and` | 27.7 | 81.1 | **+192.8%** | | `lineitem_prune` | 12.2 | 55.4 | **+354.1%** | Both regressions reproduce exactly across repeated runs (preads 14/14 vs 109/109 and 9/9 vs 91/91; timing distributions fully separated — `lineitem_prune` 12–14ms vs 45–57ms). This is deterministic, not noise. ### Why it regresses, and what would fix it The old loop checked `mask.all_false()` *before constructing* each conjunct's `filter_evaluation`. On a heavily-pruned query most splits never reached that line, so their filter reads were never registered at all — which is why the baseline is 9 preads on `lineitem_prune`. Chaining necessarily registers every conjunct's reads before pruning has run, because getting the I/O in flight early is the entire point. On prunable queries that is 10× wasted reads. Eager registration and pruning-driven skipping are therefore in direct tension, and the resolution has to live inside `filter_evaluation` rather than at the call site. Currently `flat::filter_evaluation` does: ```rust let mut array = array.clone().await?; // decodes unconditionally let mask = mask.await?; ``` It awaits the array *before* the mask, so an all-false input mask still pays for the read and the decode. Polling both concurrently and returning early when the mask resolves all-false would let cancellation propagate back up the chain and make eager registration close to free. That change is a genuine trade rather than a pure win — awaiting the array first is also what lets a conjunct's decode overlap its predecessor's compute, and short-circuiting gives that overlap up. Which effect dominates depends on selectivity, so it wants measuring on its own. These numbers say the waste dominates for prunable queries by a wide margin, so it is worth measuring next. ## What APIs are changed? Are there any user-facing changes? None. No public API changes; `chained_filter_mask` is a private helper. Results are unchanged — same masks, same arrays, same row counts. ### Checks run - `cargo nextest run -p vortex-layout -p vortex-file -p vortex-scan` — 350 passed - `cargo clippy -p vortex-layout --all-targets --all-features` — clean - `cargo +nightly fmt --all` — clean Not run: full workspace tests, Python bindings, docs — this touches one Rust file with no API or documentation surface. Co-authored-by: Claude --- vortex-layout/src/scan/tasks.rs | 173 +++++++++++++------------------- 1 file changed, 71 insertions(+), 102 deletions(-) diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs index 515efc54393..1d8e378e7f1 100644 --- a/vortex-layout/src/scan/tasks.rs +++ b/vortex-layout/src/scan/tasks.rs @@ -71,70 +71,7 @@ pub fn split_exec( let filter = Arc::clone(filter); let row_range = row_range.clone(); - // A single-conjunct filter has no adaptive ordering to decide at runtime, so the - // whole evaluation can be built up-front. - if filter.conjuncts().len() == 1 { - single_conjunct_mask(reader, filter, row_range, row_mask)? - } else { - 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); - if mask.all_false() { - return Ok(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())? - .await?; - mask = mask.bitand(&conjunct_mask); - } - if mask.all_false() { - return Ok(mask); - } - - let conjunct_mask = reader - .filter_evaluation(&row_range, conjunct, MaskFuture::ready(mask))? - .await?; - filter.report_selectivity(idx, conjunct_mask.density()); - - // Filter evaluations return a mask already intersected with the input mask. - mask = conjunct_mask; - } - - Ok(mask) - }) - } + chained_filter_mask(reader, filter, row_range, row_mask)? } }; @@ -157,68 +94,100 @@ pub fn split_exec( Ok(array_fut.boxed()) } -/// Builds the filter mask for a filter made up of a single conjunct. +/// Builds the filter mask by chaining every conjunct's evaluation at task-construction time. +/// +/// [`LayoutReader::filter_evaluation`] registers its segment reads when it is *called*, but only +/// awaits its input mask when it is *polled*. Building the evaluations one at a time — awaiting +/// each before constructing the next — therefore trickles reads in one conjunct at a time, per +/// split. Feeding each conjunct's output [`MaskFuture`] straight into the next instead registers +/// the reads for the whole chain up front, so the IO system can coalesce them, while each +/// conjunct still receives the mask its predecessor refined. /// -/// With only one conjunct there is no conjunct ordering to decide at runtime, so the whole -/// pruning-then-filter chain can be constructed at task-construction time rather than when the -/// task is first polled. This registers the conjunct's segment reads for every split before any -/// split task runs, which lets the IO system coalesce them into larger reads. +/// This matters most for filter columns that are not projected. The projection evaluation is +/// already built eagerly, so a filter over a projected column has its segments registered either +/// way; a filter over an unprojected column otherwise has nothing registering them ahead of time. /// -/// It matters most when the filter column is not part of the projection: the projection -/// evaluation is already built eagerly, so a filter over a projected column has its segments -/// registered either way, but a filter over an unprojected column otherwise trickles its reads -/// in one split at a time. -fn single_conjunct_mask( +/// The evaluation order is taken from [`FilterExpr::next_conjunct`] up front rather than being +/// re-queried between conjuncts. That ordering is recomputed only when a *completed* conjunct +/// reports its selectivity, so within a single split it was already fixed; draining it here gives +/// up nothing but lets the chain be built before anything is awaited. Ordering still adapts +/// across splits. +fn chained_filter_mask( reader: Arc, filter: Arc, row_range: Range, row_mask: Mask, ) -> VortexResult { let len = row_mask.len(); - let conjunct = filter.conjuncts()[0].clone(); - - // Store the latest version of the dynamic expression prior to pruning. We re-run the pruning - // if the version has changed by the time the task is polled. - let dynamic_version = filter.dynamic_updates(0).map(|du| du.version()); - let pruning_eval = reader.pruning_evaluation(&row_range, &conjunct, row_mask.clone())?; + let conjunct_count = filter.conjuncts().len(); + + // Each pruning evaluation is fed the original split mask rather than the mask accumulated by + // the preceding conjuncts. Pruning masks are folded together with `bitand`, and intersection + // is associative and commutative, so the final mask is unchanged. + let mut dynamic_versions = Vec::with_capacity(conjunct_count); + let mut pruning_evals = Vec::with_capacity(conjunct_count); + for (idx, conjunct) in filter.conjuncts().iter().enumerate() { + // Store the latest version of the dynamic expression prior to pruning. We re-run the + // pruning if the version has changed by the time the task is polled. + dynamic_versions.push(filter.dynamic_updates(idx).map(|du| du.version())); + pruning_evals.push(reader.pruning_evaluation(&row_range, conjunct, row_mask.clone())?); + } let pruned = MaskFuture::new(len, { let reader = Arc::clone(&reader); let filter = Arc::clone(&filter); - let conjunct = conjunct.clone(); let row_range = row_range.clone(); async move { - let mut mask = row_mask.bitand(&pruning_eval.await?); - - // If the dynamic expression has changed since pruning, re-run the pruning. - let current_version = filter.dynamic_updates(0).map(|du| du.version()); - if let Some(dv) = current_version - && dynamic_version.is_none_or(|v| v < dv) - && !mask.all_false() - { - let conjunct_mask = reader - .pruning_evaluation(&row_range, &conjunct, mask.clone())? - .await?; - mask = mask.bitand(&conjunct_mask); + let mut mask = row_mask; + + for pruning_eval in pruning_evals { + if mask.all_false() { + // Dropping the remaining evaluations cancels their outstanding reads. + return Ok(mask); + } + mask = mask.bitand(&pruning_eval.await?); + } + + // Re-run the pruning for any conjunct whose dynamic expression has changed since. + for (idx, conjunct) in filter.conjuncts().iter().enumerate() { + if mask.all_false() { + return Ok(mask); + } + + 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) + { + let conjunct_mask = reader + .pruning_evaluation(&row_range, conjunct, mask.clone())? + .await?; + mask = mask.bitand(&conjunct_mask); + } } Ok(mask) } }); - let filter_eval = reader.filter_evaluation(&row_range, &conjunct, pruned.clone())?; + let mut remaining = BitVec::from_elem(conjunct_count, true); + let mut chain = Vec::with_capacity(conjunct_count); + let mut mask_fut = pruned; + while let Some(idx) = filter.next_conjunct(&remaining) { + remaining.set(idx, false); + mask_fut = reader.filter_evaluation(&row_range, &filter.conjuncts()[idx], mask_fut)?; + chain.push((idx, mask_fut.clone())); + } Ok(MaskFuture::new(len, async move { - // Awaiting the pruned mask first lets us drop the filter evaluation, cancelling its - // reads, when pruning has already eliminated the entire split. - let pruned = pruned.await?; - if pruned.all_false() { - return Ok(pruned); + // Filter evaluations return a mask already intersected with the input mask, so the tail + // of the chain is the fully refined mask. + let mask = mask_fut.await?; + + // Every link has resolved by the time the tail has, so these awaits are already complete. + for (idx, link) in chain { + filter.report_selectivity(idx, link.await?.density()); } - // Filter evaluations return a mask already intersected with the input mask. - let mask = filter_eval.await?; - filter.report_selectivity(0, mask.density()); Ok(mask) })) }