From 5e9d4b4f5db5316825d83af52de9e14f3ed6c7d8 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 15:52:52 +0000 Subject: [PATCH] Chain filter evaluations for every conjunct Generalises the previous commit from a single conjunct to any number. `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, and nothing can be coalesced. Feeding each conjunct's output `MaskFuture` straight into the next registers the reads for the whole chain up front while each conjunct still receives the mask its predecessor refined, so no extra work is done. This is the usage the trait documentation already recommends: "defer awaiting the input mask for as long as possible ... this allows other conjuncts the opportunity to refine the mask as much as possible before it is used." The evaluation order is drained from `FilterExpr::next_conjunct` up front rather than 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 gives up nothing, and ordering still adapts across splits. The `all_false` short circuit between filter evaluations is deliberately not added back. Awaiting the array before the mask is what lets a conjunct's decode overlap its predecessor's compute, so short circuiting would trade that overlap for skipped work -- a trade that depends on selectivity and needs measuring separately. Signed-off-by: Claude Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01D6qV3R62EBNgkd2Leq5YqZ --- 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) })) }