fix: stabilize conjunct selectivity

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::<Vec<_>>();
+            .collect::<Option<Vec<_>>>()
+        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(())
+    }
 }

vortex-layout/src/scan/tasks.rs

@@ -117,10 +117,14 @@ pub fn split_exec<A: 'static + Send>(
                         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<A: 'static + Send>(
     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<A> {
     /// Function that maps into an A.
     pub mapper: Arc<dyn Fn(ArrayRef) -> VortexResult<A> + 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);
+    }
+}