From 0cb8b3f6783fafe8c3008af7965de5ddf19914ea Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 13:28:24 +0000 Subject: [PATCH] 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