Generalize the deferred-estimation early-exit to the cost model

.github/workflows/ci.yml

@@ -452,6 +452,38 @@ jobs:
           alert-title: "Rust tests (linux-arm64) failed on develop"
           deduplication-key: ci-rust-test-linux-arm64-failure
 
+  btrblocks-golden:
+    name: "Rust (btrblocks golden corpus)"
+    timeout-minutes: 10
+    runs-on: >-
+      ${{ github.repository == 'vortex-data/vortex'
+          && format('runs-on={0}/runner=amd64-medium/image=ubuntu24-full-x64-pre-v2/extras=s3-cache/tag=btrblocks-golden', github.run_id)
+          || 'ubuntu-latest' }}
+    steps:
+      - uses: runs-on/action@v2
+        if: github.repository == 'vortex-data/vortex'
+        with:
+          sccache: s3
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7
+      - uses: ./.github/actions/setup-prebuild
+        with:
+          enable-sccache: "true"
+      # The golden suite pins the compressor's decisions per feature variant, and the
+      # variants are mutually exclusive at compile time: the `default` variant only exists
+      # without `unstable_encodings` (which changes ALL_SCHEMES), so the `--all-features`
+      # workspace test jobs cannot run it. Run each feature combination explicitly.
+      - name: Golden corpus (default features)
+        run: |
+          cargo nextest run --cargo-profile ci --locked --no-fail-fast -p vortex-btrblocks --test golden
+      - name: Golden corpus (unstable_encodings)
+        run: |
+          cargo nextest run --cargo-profile ci --locked --no-fail-fast -p vortex-btrblocks --test golden \
+            --features unstable_encodings
+      - name: Golden corpus (compact, unstable_encodings + zstd + pco)
+        run: |
+          cargo nextest run --cargo-profile ci --locked --no-fail-fast -p vortex-btrblocks --test golden \
+            --features unstable_encodings,zstd,pco
+
   build-java:
     name: "Java"
     runs-on: >-

vortex-btrblocks/Cargo.toml

@@ -41,6 +41,7 @@ vortex-zstd = { workspace = true, optional = true }
 
 [dev-dependencies]
 divan = { workspace = true }
+insta = { workspace = true }
 rstest = { workspace = true }
 test-with = { workspace = true }
 vortex-array = { workspace = true, features = ["_test-harness"] }

vortex-btrblocks/REUSE.toml

@@ -0,0 +1,7 @@
+version = 1
+
+# `insta` snapshot files do not allow leading comment lines.
+[[annotations]]
+path = "tests/snapshots/**.snap"
+SPDX-FileCopyrightText = "Copyright the Vortex contributors"
+SPDX-License-Identifier = "Apache-2.0"

vortex-btrblocks/src/schemes/integer/delta.rs

@@ -14,7 +14,6 @@ use vortex_compressor::builtins::IntDictScheme;
 use vortex_compressor::builtins::StringDictScheme;
 use vortex_compressor::estimate::CompressionEstimate;
 use vortex_compressor::estimate::DeferredEstimate;
-use vortex_compressor::estimate::EstimateScore;
 use vortex_compressor::estimate::EstimateVerdict;
 use vortex_compressor::scheme::AncestorExclusion;
 use vortex_compressor::scheme::ChildSelection;
@@ -133,14 +132,13 @@ impl Scheme for DeltaScheme {
         // delta-encodes the array and measures the residual range.
         let min_ratio = self.min_ratio;
         CompressionEstimate::Deferred(DeferredEstimate::Callback(Box::new(
-            move |_compressor, data, best_so_far, _ctx, exec_ctx| {
+            move |_compressor, data, threshold, _ctx, exec_ctx| {
                 let primitive = data.array().clone().execute::<PrimitiveArray>(exec_ctx)?;
                 let full_width = primitive.ptype().bit_width() as f64;
 
                 // Delta's best case is residuals collapsing to a single bit. If even that, after
                 // the penalty, can't beat the incumbent, skip before doing the encode work.
-                let threshold = best_so_far.and_then(EstimateScore::finite_ratio);
-                if threshold.is_some_and(|t| full_width * DELTA_PENALTY <= t) {
+                if threshold.best_case_ratio_cannot_win(full_width * DELTA_PENALTY) {
                     return Ok(EstimateVerdict::Skip);
                 }
 
@@ -197,3 +195,130 @@ impl Scheme for DeltaScheme {
         Delta::try_new(compressed_bases, compressed_deltas, 0, len).map(IntoArray::into_array)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::LazyLock;
+
+    use rand::RngExt;
+    use rand::SeedableRng;
+    use rand::rngs::StdRng;
+    use vortex_array::IntoArray;
+    use vortex_array::VortexSessionExecute;
+    use vortex_array::arrays::Constant;
+    use vortex_array::arrays::ConstantArray;
+    use vortex_array::dtype::Nullability;
+    use vortex_array::scalar::Scalar;
+    use vortex_array::validity::Validity;
+    use vortex_buffer::Buffer;
+    use vortex_session::VortexSession;
+
+    use super::*;
+
+    static SESSION: LazyLock<VortexSession> = LazyLock::new(vortex_array::array_session);
+
+    /// Immediate fixed-ratio competitor; its `compress` emits a tiny constant array so the
+    /// winner is observable from the output encoding.
+    #[derive(Debug)]
+    struct FixedRatioScheme {
+        ratio: f64,
+    }
+
+    impl Scheme for FixedRatioScheme {
+        fn scheme_name(&self) -> &'static str {
+            "test.fixed_ratio"
+        }
+
+        fn matches(&self, canonical: &Canonical) -> bool {
+            canonical.dtype().is_int()
+        }
+
+        fn expected_compression_ratio(
+            &self,
+            _data: &ArrayAndStats,
+            _compress_ctx: CompressorContext,
+            _exec_ctx: &mut ExecutionCtx,
+        ) -> CompressionEstimate {
+            CompressionEstimate::Verdict(EstimateVerdict::Ratio(self.ratio))
+        }
+
+        fn compress(
+            &self,
+            _compressor: &CascadingCompressor,
+            data: &ArrayAndStats,
+            _compress_ctx: CompressorContext,
+            _exec_ctx: &mut ExecutionCtx,
+        ) -> VortexResult<ArrayRef> {
+            Ok(ConstantArray::new(
+                Scalar::primitive(0u64, Nullability::NonNullable),
+                data.array_len(),
+            )
+            .into_array())
+        }
+    }
+
+    /// Near-monotone u64 data: Delta's residual span is a few bits, so its real penalized
+    /// ratio is comfortably above 2.0 but nowhere near its 64-bit best case.
+    fn monotone_jitter_u64() -> ArrayRef {
+        let mut rng = StdRng::seed_from_u64(42);
+        let mut value = 1_700_000_000_000u64;
+        let values: Buffer<u64> = (0..4096)
+            .map(|_| {
+                value += 900 + rng.random_range(0..200);
+                value
+            })
+            .collect();
+        PrimitiveArray::new(values, Validity::NonNullable).into_array()
+    }
+
+    /// With the incumbent below Delta's achievable ratio, the callback must proceed past
+    /// the best-case threshold check and win with its measured estimate.
+    #[test]
+    fn delta_wins_over_low_threshold() -> VortexResult<()> {
+        static COMPETITOR: FixedRatioScheme = FixedRatioScheme { ratio: 2.0 };
+        static DELTA: DeltaScheme = DeltaScheme::new(1.25);
+        let compressor = CascadingCompressor::new(vec![&COMPETITOR, &DELTA]);
+
+        let mut exec_ctx = SESSION.create_execution_ctx();
+        let compressed = compressor.compress(&monotone_jitter_u64(), &mut exec_ctx)?;
+
+        assert!(compressed.is::<Delta>());
+        Ok(())
+    }
+
+    /// With the incumbent at exactly Delta's best case (`full_width * DELTA_PENALTY`, the
+    /// same expression the callback computes), Delta must not be chosen — the same decision
+    /// the pre-threshold-handle `max_ratio <= best` skip produced on this input.
+    #[test]
+    fn delta_loses_at_best_case_tie() -> VortexResult<()> {
+        static COMPETITOR: FixedRatioScheme = FixedRatioScheme {
+            ratio: 64.0 * DELTA_PENALTY,
+        };
+        static DELTA: DeltaScheme = DeltaScheme::new(1.25);
+        let compressor = CascadingCompressor::new(vec![&COMPETITOR, &DELTA]);
+
+        let mut exec_ctx = SESSION.create_execution_ctx();
+        let compressed = compressor.compress(&monotone_jitter_u64(), &mut exec_ctx)?;
+
+        assert!(compressed.is::<Constant>());
+        Ok(())
+    }
+
+    /// Wide random residuals put Delta's penalized ratio below its `min_ratio`, so with no
+    /// competitor the array must stay canonical.
+    #[test]
+    fn delta_skips_below_min_ratio() -> VortexResult<()> {
+        static DELTA: DeltaScheme = DeltaScheme::new(1.25);
+        let compressor = CascadingCompressor::new(vec![&DELTA]);
+
+        let mut rng = StdRng::seed_from_u64(43);
+        let values: Buffer<u64> = (0..4096).map(|_| rng.random::<u64>()).collect();
+        let array = PrimitiveArray::new(values, Validity::NonNullable).into_array();
+
+        let mut exec_ctx = SESSION.create_execution_ctx();
+        let compressed = compressor.compress(&array, &mut exec_ctx)?;
+
+        assert!(!compressed.is::<Delta>());
+        Ok(())
+    }
+}

vortex-btrblocks/src/schemes/integer/sequence.rs

@@ -12,7 +12,6 @@ use vortex_compressor::builtins::IntDictScheme;
 use vortex_compressor::builtins::StringDictScheme;
 use vortex_compressor::estimate::CompressionEstimate;
 use vortex_compressor::estimate::DeferredEstimate;
-use vortex_compressor::estimate::EstimateScore;
 use vortex_compressor::estimate::EstimateVerdict;
 use vortex_compressor::scheme::AncestorExclusion;
 use vortex_compressor::scheme::ChildSelection;
@@ -94,16 +93,15 @@ impl Scheme for SequenceScheme {
         // TODO(connor): `sequence_encode` allocates the encoded array just to confirm feasibility.
         // A cheaper `is_sequence` probe would let us skip the allocation entirely.
         CompressionEstimate::Deferred(DeferredEstimate::Callback(Box::new(
-            |_compressor, data, best_so_far, _ctx, exec_ctx| {
+            |_compressor, data, threshold, _ctx, exec_ctx| {
                 // `SequenceArray` stores exactly two scalars (base and multiplier), so the best
                 // achievable compression ratio is `array_len / 2`.
                 let compressed_size = 2usize;
                 let max_ratio = data.array_len() as f64 / compressed_size as f64;
 
                 // If we cannot beat the best so far, then we do not want to even try sequence
                 // encoding the data.
-                let threshold = best_so_far.and_then(EstimateScore::finite_ratio);
-                if threshold.is_some_and(|t| max_ratio <= t) {
+                if threshold.best_case_ratio_cannot_win(max_ratio) {
                     return Ok(EstimateVerdict::Skip);
                 }
 
@@ -134,3 +132,106 @@ impl Scheme for SequenceScheme {
             .ok_or_else(|| vortex_err!("cannot sequence encode array"))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::LazyLock;
+
+    use vortex_array::ExecutionCtx;
+    use vortex_array::IntoArray;
+    use vortex_array::VortexSessionExecute;
+    use vortex_array::arrays::Constant;
+    use vortex_array::arrays::ConstantArray;
+    use vortex_array::arrays::PrimitiveArray;
+    use vortex_array::dtype::Nullability;
+    use vortex_array::scalar::Scalar;
+    use vortex_array::validity::Validity;
+    use vortex_buffer::Buffer;
+    use vortex_compressor::estimate::EstimateVerdict;
+    use vortex_sequence::Sequence;
+    use vortex_session::VortexSession;
+
+    use super::*;
+
+    static SESSION: LazyLock<VortexSession> = LazyLock::new(vortex_array::array_session);
+
+    /// Number of values in the test arrays; Sequence's best case is `LEN / 2`.
+    const LEN: usize = 2048;
+
+    /// Immediate fixed-ratio competitor; its `compress` emits a tiny constant array so the
+    /// winner is observable from the output encoding.
+    #[derive(Debug)]
+    struct FixedRatioScheme {
+        ratio: f64,
+    }
+
+    impl Scheme for FixedRatioScheme {
+        fn scheme_name(&self) -> &'static str {
+            "test.fixed_ratio"
+        }
+
+        fn matches(&self, canonical: &Canonical) -> bool {
+            canonical.dtype().is_int()
+        }
+
+        fn expected_compression_ratio(
+            &self,
+            _data: &ArrayAndStats,
+            _compress_ctx: CompressorContext,
+            _exec_ctx: &mut ExecutionCtx,
+        ) -> CompressionEstimate {
+            CompressionEstimate::Verdict(EstimateVerdict::Ratio(self.ratio))
+        }
+
+        fn compress(
+            &self,
+            _compressor: &CascadingCompressor,
+            data: &ArrayAndStats,
+            _compress_ctx: CompressorContext,
+            _exec_ctx: &mut ExecutionCtx,
+        ) -> VortexResult<ArrayRef> {
+            Ok(ConstantArray::new(
+                Scalar::primitive(0i64, Nullability::NonNullable),
+                data.array_len(),
+            )
+            .into_array())
+        }
+    }
+
+    /// An exact arithmetic sequence, encodable by [`SequenceScheme`].
+    fn arithmetic_sequence() -> ArrayRef {
+        let values: Buffer<i64> = (0..LEN as i64).map(|i| 10_000 + 7 * i).collect();
+        PrimitiveArray::new(values, Validity::NonNullable).into_array()
+    }
+
+    /// With the incumbent below Sequence's best case, the callback must proceed past the
+    /// threshold check, trial-encode, and win.
+    #[test]
+    fn sequence_wins_over_low_threshold() -> VortexResult<()> {
+        static COMPETITOR: FixedRatioScheme = FixedRatioScheme { ratio: 2.0 };
+        let compressor = CascadingCompressor::new(vec![&COMPETITOR, &SequenceScheme]);
+
+        let mut exec_ctx = SESSION.create_execution_ctx();
+        let compressed = compressor.compress(&arithmetic_sequence(), &mut exec_ctx)?;
+
+        assert!(compressed.is::<Sequence>());
+        Ok(())
+    }
+
+    /// With the incumbent at exactly Sequence's best case (`LEN / 2`), Sequence must not be
+    /// chosen — the same decision the pre-threshold-handle `max_ratio <= best` skip produced
+    /// on this input.
+    #[test]
+    fn sequence_loses_at_best_case_tie() -> VortexResult<()> {
+        static COMPETITOR: FixedRatioScheme = FixedRatioScheme {
+            ratio: LEN as f64 / 2.0,
+        };
+        let compressor = CascadingCompressor::new(vec![&COMPETITOR, &SequenceScheme]);
+
+        let mut exec_ctx = SESSION.create_execution_ctx();
+        let compressed = compressor.compress(&arithmetic_sequence(), &mut exec_ctx)?;
+
+        assert!(compressed.is::<Constant>());
+        Ok(())
+    }
+}