Add CostModel::lower_bound; compressor prunes hopeless deferred work

CostModel gains lower_bound(scheme, data) -> Option<Cost>: a bound on
the best cost any candidate from the scheme could achieve. Pass 2 of
choose_best_scheme now skips a scheme's deferred sampling or callback
outright when the bound is not strictly below the best cost so far.
This is the hook that later lets decode-heavy models skip expensive
FSST sampling when it cannot win.

The default implementation returns None (never prune), which SizeCost
keeps, so behavior is unchanged (zero golden snapshot churn). The plan
sketched the signature as returning Cost with a never-prunes default,
but Cost must be finite and its units are model-defined, so no sentinel
'never prunes' value exists; Option<Cost> expresses the default
honestly. Unit tests cover pruning behind a best candidate and the
no-best case where pruning must not fire.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: Connor Tsui <connor.tsui20@gmail.com>

Author: claude

vortex-compressor/src/compressor.rs

@@ -485,6 +485,16 @@ impl CascadingCompressor {
         // short-circuit with `Skip` when they cannot beat it.
         for (scheme, deferred_estimate) in deferred {
             let _span = trace::scheme_eval_span(scheme.id()).entered();
+
+            // Model-side pruning: skip the deferred work outright when even this scheme's
+            // best case cannot strictly beat the best candidate so far.
+            if let Some((best_cost, _)) = best.as_ref()
+                && let Some(lower_bound) = self.cost_model.lower_bound(scheme.id(), data)
+                && lower_bound >= *best_cost
+            {
+                continue;
+            }
+
             let threshold: Option<EstimateScore> =
                 best.as_ref().map(|(_, candidate)| candidate.score);
             match deferred_estimate {
@@ -1382,6 +1392,76 @@ mod tests {
         Ok(())
     }
 
+    /// Delegates pricing to [`SizeCost`] but bounds every scheme at a cost that can never
+    /// win, so all deferred work is pruned as soon as any best candidate exists.
+    #[derive(Debug)]
+    struct PruneAllDeferredModel;
+
+    impl CostModel for PruneAllDeferredModel {
+        fn cost(&self, candidate: &Candidate) -> Option<Cost> {
+            SizeCost.cost(candidate)
+        }
+
+        fn canonical_cost(&self, data: &ArrayAndStats, n_values: u64) -> Cost {
+            SizeCost.canonical_cost(data, n_values)
+        }
+
+        fn lower_bound(&self, _scheme: SchemeId, _data: &ArrayAndStats) -> Option<Cost> {
+            Some(Cost::new(f64::MAX))
+        }
+    }
+
+    #[test]
+    fn lower_bound_prunes_deferred_work_behind_a_best() -> VortexResult<()> {
+        // `CallbackRatioScheme`'s deferred `Ratio(3.0)` would beat `DirectRatioScheme`'s
+        // immediate `Ratio(2.0)` (see `callback_ratio_competes_numerically`), but the model
+        // bounds it above the running best, so its deferred work is pruned.
+        let compressor = CascadingCompressor::new(vec![&DirectRatioScheme, &CallbackRatioScheme])
+            .with_cost_model(Arc::new(PruneAllDeferredModel));
+        let schemes: [&'static dyn Scheme; 2] = [&DirectRatioScheme, &CallbackRatioScheme];
+        let data = estimate_test_data();
+        let mut exec_ctx = SESSION.create_execution_ctx();
+
+        let winner = compressor.choose_best_scheme(
+            &schemes,
+            &data,
+            CompressorContext::new(),
+            &mut exec_ctx,
+        )?;
+
+        assert!(matches!(
+            winner,
+            Some((scheme, WinnerEstimate::Score { score: EstimateScore::FiniteCompression(2.0), .. }))
+                if scheme.id() == DirectRatioScheme.id()
+        ));
+        Ok(())
+    }
+
+    #[test]
+    fn lower_bound_never_prunes_without_a_best() -> VortexResult<()> {
+        // With no best candidate there is nothing to prune against, so the deferred
+        // callback still runs even under a prune-everything bound.
+        let compressor = CascadingCompressor::new(vec![&CallbackRatioScheme])
+            .with_cost_model(Arc::new(PruneAllDeferredModel));
+        let schemes: [&'static dyn Scheme; 1] = [&CallbackRatioScheme];
+        let data = estimate_test_data();
+        let mut exec_ctx = SESSION.create_execution_ctx();
+
+        let winner = compressor.choose_best_scheme(
+            &schemes,
+            &data,
+            CompressorContext::new(),
+            &mut exec_ctx,
+        )?;
+
+        assert!(matches!(
+            winner,
+            Some((scheme, WinnerEstimate::Score { score: EstimateScore::FiniteCompression(3.0), .. }))
+                if scheme.id() == CallbackRatioScheme.id()
+        ));
+        Ok(())
+    }
+
     #[test]
     fn all_null_array_compresses_to_constant() -> VortexResult<()> {
         let array = PrimitiveArray::new(

vortex-compressor/src/cost/mod.rs

@@ -42,6 +42,7 @@ use std::fmt::Debug;
 pub use size::SizeCost;
 
 pub use crate::candidate::Candidate;
+use crate::scheme::SchemeId;
 use crate::stats::ArrayAndStats;
 
 /// An opaque, totally ordered cost. Lower is better.
@@ -95,4 +96,18 @@ pub trait CostModel: Debug + Send + Sync + 'static {
     /// Cost of leaving the array canonical — the baseline every candidate must strictly
     /// beat to be selected.
     fn canonical_cost(&self, data: &ArrayAndStats, n_values: u64) -> Cost;
+
+    /// A lower bound on the cost any candidate from `scheme` could possibly achieve on
+    /// `data`, or `None` (the default) if the model offers no bound.
+    ///
+    /// The compressor skips a scheme's deferred estimation work (sampling, callbacks)
+    /// outright when the bound is not strictly below the best cost observed so far, since
+    /// even the scheme's best case could not win selection. Models that can bound a
+    /// scheme's payoff from cheap facts (e.g. minimum plausible output bytes at zero decode
+    /// cost) can prune expensive sampling here; a sound bound never exceeds the cost of any
+    /// candidate the scheme could actually produce.
+    fn lower_bound(&self, scheme: SchemeId, data: &ArrayAndStats) -> Option<Cost> {
+        let _ = (scheme, data);
+        None
+    }
 }