diff --git a/vortex-layout/src/scan/filter.rs b/vortex-layout/src/scan/filter.rs index 2cacfbecf89..2cb36068913 100644 --- a/vortex-layout/src/scan/filter.rs +++ b/vortex-layout/src/scan/filter.rs @@ -114,7 +114,7 @@ impl FilterExpr { // 1. The ordering is a heuristic optimization, not a correctness requirement // 2. The selectivity values are statistical estimates that change gradually // 3. Any ordering will produce correct results, just with different performance - let all_selectivity = self + let Some(all_selectivity) = self .conjunct_selectivity .iter() .map(|histogram| { @@ -123,10 +123,14 @@ impl FilterExpr { .quantile(self.selectivity_quantile) .map_err(|e| vortex_err!("{e}")) // Only errors when the quantile is out of range .vortex_expect("quantile out of range") - // If the sketch is empty, its selectivity is 0. - .unwrap_or_default() }) - .collect::>(); + .collect::>>() + else { + // Preserve the input order until every conjunct has been observed. Treating an unseen + // conjunct as perfectly selective can move expensive predicates ahead of selective + // predicates based only on which concurrent split finishes first. + return; + }; { let ordering = self.ordering.read(); @@ -192,4 +196,18 @@ mod tests { ); Ok(()) } + + #[test] + fn waits_for_all_conjuncts_before_reordering() -> VortexResult<()> { + let dtype = DType::Bool(Nullability::Nullable); + let bound = and(root(), not(root())).bind(&dtype)?; + let filter = FilterExpr::new(bound); + + filter.report_selectivity(0, 0.9); + assert_eq!(*filter.ordering.read(), vec![0, 1]); + + filter.report_selectivity(1, 0.1); + assert_eq!(*filter.ordering.read(), vec![1, 0]); + Ok(()) + } } diff --git a/vortex-layout/src/scan/tasks.rs b/vortex-layout/src/scan/tasks.rs index 218efb64a0d..6fdd3ee0a60 100644 --- a/vortex-layout/src/scan/tasks.rs +++ b/vortex-layout/src/scan/tasks.rs @@ -117,10 +117,14 @@ pub fn split_exec( return Ok(mask); } + let input_true_count = mask.true_count(); let conjunct_mask = reader .filter_evaluation(&row_range, conjunct, MaskFuture::ready(mask))? .await?; - filter.report_selectivity(idx, conjunct_mask.density()); + filter.report_selectivity( + idx, + conditional_selectivity(input_true_count, conjunct_mask.true_count()), + ); // Filter evaluations return a mask already intersected with the input mask. mask = conjunct_mask; @@ -150,6 +154,12 @@ pub fn split_exec( Ok(array_fut.boxed()) } +fn conditional_selectivity(input_true_count: usize, output_true_count: usize) -> f64 { + debug_assert!(input_true_count > 0); + debug_assert!(output_true_count <= input_true_count); + output_true_count as f64 / input_true_count as f64 +} + /// Information needed to execute a single split task. /// /// Row selection is evaluated before creating a split task so it's not included @@ -163,3 +173,13 @@ pub struct TaskContext { /// Function that maps into an A. pub mapper: Arc VortexResult + Send + Sync>, } + +#[cfg(test)] +mod tests { + use super::conditional_selectivity; + + #[test] + fn selectivity_is_relative_to_the_input_mask() { + assert_eq!(conditional_selectivity(20, 5), 0.25); + } +}