perf(writer): Rust's stats verdicts during cascade estimation (+36% cascading write, -47% alloc)

## perf(writer): apply stats verdicts when measuring cascade children

measureBestChild trial-encoded every accepting encoder at every cascade
level; only the top-level competition consulted expectedRatio. Rust's
choose_best_scheme runs the verdict pass at every level, so a candidate
whose verdict is SKIP is now never trial-encoded on a child sample.

Two sample-context rules from Rust's compress (compressor/cascade.rs)
come with it: an empty child is returned before selection (costs 0), and
constant detection does not settle a sample, so ALWAYS_USE is measured
like any other candidate. Short-circuiting it made an empty Sparse patch
child cost a constant node and flipped taxi's store_and_fwd_flag codes
from Sparse (Rust's choice) to bit-packing.

Output: NYC taxi 2024-01 (2.96M rows) writes to the same 59,912,206
bytes; two columns differ by <120 bytes (vortex-jni: 44,463,892).

JavaVsJniWriteBenchmark.javaWriteCascading, 5 forks:
  0.442 +- 0.002 -> 0.439 +- 0.017 ops/s (per fork 0.457 0.449 0.440 0.417 0.434)
  gc.alloc.rate.norm 7.10 -> 6.87 GB/op (-3%)

Small because most Java verdicts still defer to the sample where Rust's
schemes skip (RLE, RunEnd, BitPacking, Sequence) or estimate (FoR,
Sparse); the next commits port those.

## perf(writer): skip RLE and RunEnd below Rust's run-length thresholds

Port Rust's RLEScheme/FloatRLEScheme and RunEndScheme verdicts: skip
when the array's runs average fewer than 4 values (RUN_LENGTH_THRESHOLD,
RUN_END_THRESHOLD), else defer to the sample. ArrayStats gains
averageRunLength (value count / runs, floats compared as values), always
computed as in Rust's stats, from one branch-free pass that the capped
distinct scan cannot cut short. RunEnd no longer asks for distinct
counts, as Rust's scheme does not.

Java trial-encoded RLE on every child sample whatever its run length:
it was the largest single allocator of a cascading write (~1 GB/op,
~33k trial encodes of worst-case-sized buffers).

Output: NYC taxi 2024-01 writes to the same 59,912,206 bytes as before
(per column identical); OHLC 10M rows unchanged at 58,012,218 bytes.

JavaVsJniWriteBenchmark.javaWriteCascading, 5 forks:
  0.439 +- 0.017 -> 0.599 +- 0.010 ops/s (+36%; per fork 0.592 0.616 0.594 0.584 0.612)
  gc.alloc.rate.norm 6.87 -> 3.64 GB/op (-47%)
  vortex-jni reference: 0.848 +- 0.005 ops/s

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015R8cf8YM8XFQqrkyhSAtYr

Author: dfa1

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Changed
+- Cascading writes are about 35% faster and allocate half as much (10M-row OHLC: 0.44 → 0.60 writes/s), to the same bytes: as in Rust, cascade children are no longer trial-encoded with candidates their stats rule out, and RLE/RunEnd skip arrays whose runs average under 4 values ([#465](https://github.com/dfa1/vortex-java/pull/465)).
 - **Breaking:** `EncodingEncoder#expectedRatio` takes `(DType, ArrayAndStats)` instead of `(DType, Object, ArrayStats)`, as Rust's `expected_compression_ratio` takes `ArrayAndStats`: the cascade's stats scan now runs on the first `stats()` call and is shared by every candidate ([#464](https://github.com/dfa1/vortex-java/pull/464)).
 - **Breaking:** default writes emit Rust's `vortex.zoned` zone map instead of the legacy `vortex.stats`: one zone per 8192 rows regardless of `writeChunk` batches, with Rust's per-dtype stats (64-byte string bounds, `nan_count`) and no zone sum. Calcite and `ZoneReducer` aggregate push-down needs the legacy layout, so on these files it falls back to a scan; target `Editions.CORE_2025_10_0` to keep it. Read new files with this release or later ([#447](https://github.com/dfa1/vortex-java/issues/447)).
 - **Breaking:** `WriteOptions.defaults()` cascades up to depth 3, as Rust's writer always does: default writes are smaller (e.g. two-decimal F64 1.06 MB → 924 KB per 1M rows) and slower. `WriteOptions.cascading(0)` restores first-match encoding. A custom encoder list or `WriteRegistry` is now also the cascade's candidate set, with canonical `vortex.primitive` as the fallback ([#458](https://github.com/dfa1/vortex-java/issues/458)).

writer/src/main/java/io/github/dfa1/vortex/writer/encode/ArrayStats.java

@@ -23,16 +23,21 @@
 ///                          `valueCount / 2 + 1`. [#distinctCount()] is then a lower bound and
 ///                          [#mostFrequentBits()] / [#topFrequency()] are partial — see
 ///                          [#distinctCapped()] for why every consumer can still decide
+/// @param averageRunLength  `valueCount` divided by the number of runs of equal values, as Rust's
+///                          `average_run_length` (integer division; floats compare as values, so
+///                          `-0.0 == 0.0` and every NaN starts a run). Always computed, as in
+///                          Rust: it is a single branch-free pass, never capped
 public record ArrayStats(
         long valueCount,
         long distinctCount,
         long mostFrequentBits,
         long topFrequency,
-        boolean distinctCapped
+        boolean distinctCapped,
+        long averageRunLength
 ) {
 
     /// Sentinel stats for empty arrays.
-    public static final ArrayStats EMPTY = new ArrayStats(0, 0, 0, 0, false);
+    public static final ArrayStats EMPTY = new ArrayStats(0, 0, 0, 0, false, 0);
 
     /// Value ranges narrower than this are always counted densely: Rust's
     /// `DENSE_DISTINCT_ALWAYS_RANGE` (covers every 8-bit array).
@@ -54,12 +59,10 @@ public static ArrayStats compute(PType ptype, Object data, StatsOptions options)
         if (n == 0) {
             return EMPTY;
         }
-        if (options == StatsOptions.NONE) {
-            return new ArrayStats(n, -1, 0, 0, false);
-        }
+        long averageRunLength = n / runs(ptype, data, n);
         if (!options.countDistinct() && !options.trackMostFrequent()) {
             // Nothing to accumulate — the scan below would read every element and discard it.
-            return new ArrayStats(n, -1, 0, 0, false);
+            return new ArrayStats(n, -1, 0, 0, false, averageRunLength);
         }
 
         // Stop once the distinct count passes half the rows: past that point every consumer's
@@ -77,7 +80,7 @@ public static ArrayStats compute(PType ptype, Object data, StatsOptions options)
         if (!ptype.isFloating()) {
             // Widened once (zero-copy for I64/U64) so one loop serves every integer width.
             long[] values = PrimitiveArrays.toLongs(data, ptype, EncodingId.VORTEX_PRIMITIVE);
-            return integerStats(values, n, cap, options.countDistinct());
+            return integerStats(values, n, cap, options.countDistinct(), averageRunLength);
         }
 
         // Sized for the low-cardinality case and grown from there, NOT for the cap. Pre-sizing
@@ -91,7 +94,7 @@ public static ArrayStats compute(PType ptype, Object data, StatsOptions options)
             case F32 -> scanF32((float[]) data, n, counts, cap);
             default -> scanF64((double[]) data, n, counts, cap);
         };
-        return fromCounts(n, counts, capped, options.countDistinct());
+        return fromCounts(n, counts, capped, options.countDistinct(), averageRunLength);
     }
 
     /// Integer stats as Rust's `typed_int_stats` (vortex-compressor #10037, #10041) computes
@@ -107,8 +110,9 @@ public static ArrayStats compute(PType ptype, Object data, StatsOptions options)
     /// @param n      element count, at least 1
     /// @param cap    distinct count past which the hashed scan may stop
     /// @param countDistinct whether [#distinctCount()] is reported
+    /// @param averageRunLength the already computed [#averageRunLength()]
     /// @return the stats
-    private static ArrayStats integerStats(long[] a, int n, long cap, boolean countDistinct) {
+    private static ArrayStats integerStats(long[] a, int n, long cap, boolean countDistinct, long averageRunLength) {
         long min = a[0];
         long max = a[0];
         for (int i = 0; i < n; i++) {
@@ -122,7 +126,7 @@ private static ArrayStats integerStats(long[] a, int n, long cap, boolean countD
         long span = max - min;
         if (span >= 0 && (span < DENSE_ALWAYS_RANGE
                 || span < DENSE_MAX_RANGE && span < n)) {
-            return denseStats(a, n, min, (int) span + 1, countDistinct);
+            return denseStats(a, n, min, (int) span + 1, countDistinct, averageRunLength);
         }
 
         LongIntMap counts = new LongIntMap(Math.min(n, 2048));
@@ -133,19 +137,20 @@ private static ArrayStats integerStats(long[] a, int n, long cap, boolean countD
             if (v != prev) {
                 counts.increment(prev, pending);
                 if (counts.size() > cap) {
-                    return fromCounts(n, counts, true, countDistinct);
+                    return fromCounts(n, counts, true, countDistinct, averageRunLength);
                 }
                 prev = v;
                 pending = 0;
             }
             pending++;
         }
         counts.increment(prev, pending);
-        return fromCounts(n, counts, counts.size() > cap, countDistinct);
+        return fromCounts(n, counts, counts.size() > cap, countDistinct, averageRunLength);
     }
 
     /// Counts by `value - min` into an array: no hashing, and never capped.
-    private static ArrayStats denseStats(long[] a, int n, long min, int range, boolean countDistinct) {
+    private static ArrayStats denseStats(long[] a, int n, long min, int range, boolean countDistinct,
+                                         long averageRunLength) {
         int[] counts = new int[range];
         for (int i = 0; i < n; i++) {
             counts[(int) (a[i] - min)]++;
@@ -163,15 +168,62 @@ private static ArrayStats denseStats(long[] a, int n, long min, int range, boole
                 topIndex = i;
             }
         }
-        return new ArrayStats(n, countDistinct ? distinct : -1L, min + topIndex, counts[topIndex], false);
+        return new ArrayStats(n, countDistinct ? distinct : -1L, min + topIndex, counts[topIndex], false,
+                averageRunLength);
     }
 
-    private static ArrayStats fromCounts(int n, LongIntMap counts, boolean capped, boolean countDistinct) {
+    private static ArrayStats fromCounts(int n, LongIntMap counts, boolean capped, boolean countDistinct,
+                                         long averageRunLength) {
         LongIntMap.Entry top = counts.maxEntry();
         long topFreqBits = top == null ? 0L : top.key();
         int topFreq = top == null ? 0 : top.value();
         long distinct = countDistinct ? counts.size() : -1L;
-        return new ArrayStats(n, distinct, topFreqBits, topFreq, capped);
+        return new ArrayStats(n, distinct, topFreqBits, topFreq, capped, averageRunLength);
+    }
+
+    /// Runs of equal values: 1 plus every change between neighbors. One loop per carrier with a
+    /// branch-free body, so C2 vectorizes it (CLAUDE.md hot-loop rule).
+    private static long runs(PType ptype, Object data, int n) {
+        long changes = 0;
+        switch (ptype) {
+            case I8, U8 -> {
+                byte[] a = (byte[]) data;
+                for (int i = 1; i < n; i++) {
+                    changes += a[i] != a[i - 1] ? 1 : 0;
+                }
+            }
+            case I16, U16, F16 -> {
+                short[] a = (short[]) data;
+                for (int i = 1; i < n; i++) {
+                    changes += a[i] != a[i - 1] ? 1 : 0;
+                }
+            }
+            case I32, U32 -> {
+                int[] a = (int[]) data;
+                for (int i = 1; i < n; i++) {
+                    changes += a[i] != a[i - 1] ? 1 : 0;
+                }
+            }
+            case I64, U64 -> {
+                long[] a = (long[]) data;
+                for (int i = 1; i < n; i++) {
+                    changes += a[i] != a[i - 1] ? 1 : 0;
+                }
+            }
+            case F32 -> {
+                float[] a = (float[]) data;
+                for (int i = 1; i < n; i++) {
+                    changes += a[i] != a[i - 1] ? 1 : 0;
+                }
+            }
+            case F64 -> {
+                double[] a = (double[]) data;
+                for (int i = 1; i < n; i++) {
+                    changes += a[i] != a[i - 1] ? 1 : 0;
+                }
+            }
+        }
+        return changes + 1;
     }
 
     private static boolean scanF16(short[] a, int n, LongIntMap counts, long cap) {

writer/src/main/java/io/github/dfa1/vortex/writer/encode/CascadingCompressor.java

@@ -213,23 +213,31 @@ private EncodeResult encodeWithCtx(DType dtype, Object data, EncodeContext ctx)
         // excluded set so spliceResult's notApplicable retry can rotate to the next accepting
         // encoding (e.g. DateTimePartsEncoding → ExtEncoding when the input is raw storage rather
         // than DateTimePartsData).
-        if (!(dtype instanceof DType.Primitive p)) {
+        if (!(dtype instanceof DType.Primitive)) {
             return spliceResult(findPrimitiveEncoding(dtype, ctx.excluded()), dtype, data, ctx);
         }
 
-        // Stats-first selection (Rust vortex-compressor pattern): merge every eligible
-        // encoder's StatsOptions so the one scan, run lazily by the first encoder whose
-        // expectedRatio() reads stats, satisfies every later one. ALWAYS_USE short-circuits;
-        // SKIP excludes; COMPLETE defers to the sample-encoded path below.
+        return competeAndEncode(dtype, withStats(dtype, data, ctx), ctx,
+                sampleSize -> primitiveBytes(dtype, sampleSize));
+    }
+
+    /// Bundles `data` with stats under the [StatsOptions] merged from every eligible encoder,
+    /// so the one scan, run lazily by the first encoder whose expectedRatio() reads stats,
+    /// satisfies every later one (Rust vortex-compressor pattern).
+    private ArrayAndStats withStats(DType dtype, Object data, EncodeContext ctx) {
         StatsOptions merged = StatsOptions.NONE;
         for (EncodingEncoder enc : encodings) {
             if (enc.accepts(dtype) && !ctx.excluded().contains(enc.encodingId())) {
                 merged = StatsOptions.merge(merged, enc.statsOptions());
             }
         }
+        return new ArrayAndStats(dtype, data, merged);
+    }
 
-        return competeAndEncode(dtype, new ArrayAndStats(p, data, merged), ctx,
-                sampleSize -> primitiveBytes(dtype, sampleSize));
+    /// Whether `dtype` goes through the competition ([#competeAndEncode]) rather than a
+    /// first-match dispatch — and so whether stats verdicts apply to it.
+    private static boolean competes(DType dtype) {
+        return dtype instanceof DType.Primitive || dtype instanceof DType.Utf8 || dtype instanceof DType.Binary;
     }
 
     /// Shared sample-and-measure competition: stats-based skip/always-use sweep, then a
@@ -317,13 +325,33 @@ private long measureStep(EncodingEncoder enc, CascadeStep step, EncodeContext ct
         return total;
     }
 
+    /// Smallest size any candidate encodes the (already sampled) child `data` to.
+    ///
+    /// Selects as Rust's `choose_best_scheme` does at every cascade level, not only at the top:
+    /// stats verdicts first, so a candidate whose verdict is SKIP is never trial-encoded. Without
+    /// it every accepting encoder was trial-encoded at every level, which was most of a cascading
+    /// write's CPU and allocation.
+    ///
+    /// Two sample-context rules from Rust's `compress` (`compressor/cascade.rs`): an empty array
+    /// is returned before any selection, so it costs nothing; and constant detection is skipped,
+    /// since a constant sample does not imply a constant array — so ALWAYS_USE does not settle the
+    /// child here, the candidate is only measured like the rest. Short-circuiting it made an
+    /// empty patch child of Sparse cost a constant node and flipped taxi's dict codes from Sparse
+    /// (Rust's choice) to bit-packing.
     private long measureBestChild(DType dtype, Object data, EncodeContext ctx) {
         int n = dataLength(data);
         long best = primitiveBytes(dtype, n);
+        if (n == 0) {
+            return best;
+        }
+        ArrayAndStats input = competes(dtype) ? withStats(dtype, data, ctx) : null;
         for (EncodingEncoder enc : encodings) {
             if (!enc.accepts(dtype) || ctx.excluded().contains(enc.encodingId())) {
                 continue;
             }
+            if (input != null && enc.expectedRatio(dtype, input) == Estimate.SKIP) {
+                continue;
+            }
             CascadeStep step = enc.encodeCascade(dtype, data, ctx);
             if (!step.isTerminal() && ctx.allowedCascading() == 0) {
                 continue;

writer/src/main/java/io/github/dfa1/vortex/writer/encode/RleEncodingEncoder.java

@@ -21,11 +21,25 @@ public final class RleEncodingEncoder implements EncodingEncoder {
 
     private static final int FL_CHUNK_SIZE = 1024;
 
+    /// Rust's `RUN_LENGTH_THRESHOLD` (`vortex-btrblocks` `schemes/integer/mod.rs`), shared by its
+    /// integer and float RLE schemes.
+    private static final long RUN_LENGTH_THRESHOLD = 4;
+
     @Override
     public EncodingId encodingId() {
         return EncodingId.FASTLANES_RLE;
     }
 
+    /// Rust's `RLEScheme`/`FloatRLEScheme` verdict: skip unless runs average at least
+    /// [#RUN_LENGTH_THRESHOLD] values, else defer to the sample.
+    @Override
+    public Estimate expectedRatio(DType dtype, ArrayAndStats data) {
+        if (!(dtype instanceof DType.Primitive)) {
+            return Estimate.COMPLETE;
+        }
+        return data.stats().averageRunLength() < RUN_LENGTH_THRESHOLD ? Estimate.SKIP : Estimate.COMPLETE;
+    }
+
     @Override
     public boolean accepts(DType dtype) {
         // every primitive, floats included (Rust's FloatRLEScheme): runs compare raw bits, so -0.0,

writer/src/main/java/io/github/dfa1/vortex/writer/encode/RunEndEncodingEncoder.java

@@ -19,6 +19,9 @@
 /// Write-only encoder for `vortex.runend`.
 public final class RunEndEncodingEncoder implements EncodingEncoder {
 
+    /// Rust's `RUN_END_THRESHOLD` (`vortex-btrblocks` `schemes/integer/runend.rs`).
+    private static final long RUN_END_THRESHOLD = 4;
+
     @Override
     public EncodingId encodingId() {
         return EncodingId.VORTEX_RUNEND;
@@ -29,35 +32,15 @@ public boolean accepts(DType dtype) {
         return dtype instanceof DType.Primitive p && !p.ptype().isFloating();
     }
 
-    @Override
-    public StatsOptions statsOptions() {
-        return new StatsOptions(true, true);
-    }
-
+    /// Rust's `RunEndScheme` verdict: skip unless runs average at least [#RUN_END_THRESHOLD]
+    /// values, else defer to the sample. Needs no distinct count, so it declares no
+    /// [StatsOptions], as Rust's scheme declares none.
     @Override
     public Estimate expectedRatio(DType dtype, ArrayAndStats data) {
-        ArrayStats stats = data.stats();
-        if (!(dtype instanceof DType.Primitive) || !stats.hasDistinctCount()) {
+        if (!(dtype instanceof DType.Primitive)) {
             return Estimate.COMPLETE;
         }
-        long n = stats.valueCount();
-        long distinct = stats.distinctCount();
-        if (n == 0) {
-            return Estimate.SKIP;
-        }
-        // The only consumer the capped scan does not settle: `distinct >= n` needs the exact
-        // count, and a capped scan only proves distinct > n/2 + 1. Defer to the sample instead
-        // of guessing — a 1024-row sample encode is far cheaper than the probes the cap saved.
-        if (stats.distinctCapped()) {
-            return Estimate.COMPLETE;
-        }
-        // Skip rule: if every value is distinct, each row is its own run — pure overhead.
-        // Defer to the sample-encoded path otherwise; RunEnd's actual compression depends
-        // on run-length distribution which is not summarized by distinct count alone.
-        if (distinct >= n) {
-            return Estimate.SKIP;
-        }
-        return Estimate.COMPLETE;
+        return data.stats().averageRunLength() < RUN_END_THRESHOLD ? Estimate.SKIP : Estimate.COMPLETE;
     }
 
     /// Encodes a boolean array as `vortex.runend`: consecutive equal values collapse into one run.

writer/src/test/java/io/github/dfa1/vortex/writer/encode/ArrayStatsTest.java

@@ -217,6 +217,34 @@ void compute_everyIntegerCountingPath_matchesPerRowReference(PType ptype, Object
         assertThat(result.mostFrequentBits()).isEqualTo(expectedTopBits);
     }
 
+    @Test
+    void compute_averageRunLength_isValueCountOverRuns() {
+        // Given: a date-like column, each value repeated 30 times, 100 runs
+        int[] data = new int[3_000];
+        for (int i = 0; i < data.length; i++) {
+            data[i] = i / 30;
+        }
+
+        // When: run length is reported even when no distinct stats are requested, as in Rust
+        ArrayStats result = ArrayStats.compute(PType.I32, data, StatsOptions.NONE);
+
+        // Then
+        assertThat(result.averageRunLength()).isEqualTo(30);
+    }
+
+    @Test
+    void compute_averageRunLength_comparesFloatsAsValues() {
+        // Given: Rust's float stats compare with `!=`, so -0.0 continues a run of 0.0 and every
+        // NaN starts a new one: runs are [0.0, -0.0, 0.0] [NaN] [NaN] [1.0] -> 4 runs over 6 values
+        double[] data = {0.0, -0.0, 0.0, Double.NaN, Double.NaN, 1.0};
+
+        // When
+        ArrayStats result = ArrayStats.compute(PType.F64, data, StatsOptions.NONE);
+
+        // Then: integer division, as Rust's value_count / runs
+        assertThat(result.averageRunLength()).isEqualTo(1);
+    }
+
     private static long[] widen(byte[] a, boolean unsigned) {
         long[] out = new long[a.length];
         for (int i = 0; i < a.length; i++) {