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) })) }