fix(reader): harden RunEnd/Constant/Zoned/Pco against malformed input

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - Malformed files no longer crash the reader with a raw JDK exception when decoding VarBin, Dict, Bitpacked, ALP, Sparse, Chunked, or Struct columns — every case now fails as `VortexException`. ([ef982992](https://github.com/dfa1/vortex-java/commit/ef982992))
+- Same hardening for RunEnd, Constant, zone-map stats, and Pco columns — every case now fails as `VortexException`. ([12d7466c](https://github.com/dfa1/vortex-java/commit/12d7466c))
 
 ### Added
 

TODO.md

@@ -38,13 +38,7 @@ known gap, a contract audit, or supporting infra.
 Each encoding's `decode(DecodeContext)` should be exercised against crafted metadata that
 decodes but disagrees with the buffer payload. `bufferIndices[i] >= ctx.bufferCount()` (and the
 equivalent child-index check) is centralized in `DecodeContext.buffer(i)`/`decodeChild(i)`.
-VarBin, Dict, Bitpacked, ALP, Sparse, Chunked, and Struct are done — remaining gotchas:
-
-- [ ] **RLE / RunEnd**: `run_ends` non-monotonic; last `run_end` ≠ `row_count`.
-- [ ] **Constant**: protobuf scalar value missing or type-mismatched against declared `DType`.
-- [ ] **Zoned**: zone-map min > max; zone count ≠ child chunk count.
-- [ ] **Pco**: `bits_per_offset > 64`; `bin_count == 0` with non-empty page; per-page
-  `n` greater than `DEFAULT_MAX_PAGE_N`; ANS state values inconsistent with weight table.
+VarBin, Dict, Bitpacked, ALP, Sparse, Chunked, Struct, RunEnd, Constant, Zoned, and Pco are done.
 
 ### Resource caps
 

reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java

@@ -475,6 +475,15 @@ private List<ArrayStats> decodeZoneTable(ColumnName column) {
             return null;
         }
         long nZones = statsFlat.rowCount();
+        // statsFlat.rowCount() is an unvalidated field straight from the layout FlatBuffer
+        // (PostscriptParser never bounds it — zone count is deliberately decoupled from the
+        // data layout's chunk count, see this method's Javadoc). Below it sizes an ArrayList
+        // and drives a per-zone loop via an `(int) nZones` cast: a negative value throws a raw
+        // IllegalArgumentException from the ArrayList constructor instead of degrading to "no
+        // zone map" like every other unusable shape this method already falls back on.
+        if (nZones < 0 || nZones > Integer.MAX_VALUE) {
+            return null;
+        }
         SegmentSpec spec = file.footer().segmentSpecs().get(segIdx);
         try (Arena tableArena = Arena.ofConfined()) {
             Array decoded = file.decodeSegment(spec, statsDtype, nZones, tableArena);
@@ -485,7 +494,11 @@ private List<ArrayStats> decodeZoneTable(ColumnName column) {
             Array maxA = fieldOrNull(table, "max");
             Array sumA = fieldOrNull(table, "sum");
             Array nullCountA = fieldOrNull(table, "null_count");
-            List<ArrayStats> out = new ArrayList<>((int) nZones);
+            // Not pre-sized from nZones: it is bounded above only by Integer.MAX_VALUE (see the
+            // guard above), and a single ArrayList allocation at that scale is itself an
+            // OutOfMemoryError vector the security contract forbids. Growing incrementally keeps
+            // memory proportional to what the loop below actually produces.
+            List<ArrayStats> out = new ArrayList<>();
             for (long i = 0; i < nZones; i++) {
                 Object nullCount = boxedScalar(nullCountA, i);
                 out.add(new ArrayStats(

reader/src/main/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoder.java

@@ -114,6 +114,13 @@ private static Array constantPrimitive(DType outDtype, PType ptype, ProtoScalarV
 
     private static Array decodeDecimal(DType dtype, ProtoScalarValue scalar, long n) {
         byte[] elemBytes = scalar.bytes_value();
+        if (elemBytes == null) {
+            // A scalar whose oneof tag doesn't match the declared Decimal dtype (e.g. only
+            // int64_value set) leaves bytes_value() null; without this guard the length read
+            // below is a raw NullPointerException instead of a VortexException (ADR 0003).
+            throw new VortexException(EncodingId.VORTEX_CONSTANT,
+                    "constant decimal scalar missing bytes_value");
+        }
         int elemLen = elemBytes.length;
         // Decode the single scalar value via LazyDecimalArray (reuses its LE byte-order logic),
         // then wrap in a constant array — O(1) allocation regardless of row count.

reader/src/main/java/io/github/dfa1/vortex/reader/decode/PcoEncodingDecoder.java

@@ -7,6 +7,7 @@
 import io.github.dfa1.vortex.core.io.PTypeIO;
 import io.github.dfa1.vortex.core.proto.ProtoPcoChunkInfo;
 import io.github.dfa1.vortex.core.proto.ProtoPcoMetadata;
+import io.github.dfa1.vortex.core.proto.ProtoPcoPageInfo;
 import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.reader.array.BoolArray;
 import io.github.dfa1.vortex.reader.array.MaskedArray;
@@ -67,6 +68,28 @@ public Array decode(DecodeContext ctx) {
             }
         }
 
+        // Pages declare their own value counts (ProtoPcoPageInfo.n_values), independent of
+        // validCount. A crafted file can pair a huge or negative per-page count with a small
+        // rowCount: without this check, a negative count silently no-ops its loop while a
+        // desynced total either writes past rawLatents/compactOut (raw IndexOutOfBounds) or
+        // sizes rawAdjs from an attacker-controlled chunkN unrelated to any real buffer
+        // (OutOfMemoryError). Validating the total up front keeps every per-page/per-chunk
+        // access below implicitly bounded by validCount.
+        long totalPageValues = 0L;
+        for (ProtoPcoChunkInfo chunkInfo : meta.chunks()) {
+            for (ProtoPcoPageInfo page : chunkInfo.pages()) {
+                if (page.n_values() < 0) {
+                    throw new VortexException(EncodingId.VORTEX_PCO,
+                            "pco page n_values " + page.n_values() + " is negative");
+                }
+                totalPageValues += page.n_values();
+            }
+        }
+        if (totalPageValues != validCount) {
+            throw new VortexException(EncodingId.VORTEX_PCO,
+                    "pco total page values " + totalPageValues + " != expected valid row count " + validCount);
+        }
+
         MemorySegment rawLatents = ctx.arena().allocate(validCount * Long.BYTES);
 
         int nChunks = meta.chunks().size();
@@ -727,6 +750,14 @@ private static PcoBin[] readBins(LeBitReader r, int nBins, int ansSizeLog, int d
             int weight = (int) r.readBits(ansSizeLog) + 1;
             long lower = r.readBits(dtypeSize);
             int offsetBits = (int) r.readBits(offsetBitsWidth);
+            if (offsetBits > 64) {
+                // offsetBitsWidth is 5/6/7 bits wide (max value 31/63/127), wider than the
+                // 64-bit latent an offset can ever legally span; a page later reads this many
+                // bits per value via LeBitReader#readBits(int), whose own <=64 contract this
+                // would otherwise violate.
+                throw new VortexException(EncodingId.VORTEX_PCO,
+                        "pco bin offsetBits " + offsetBits + " exceeds max 64");
+            }
             bins[b] = new PcoBin(weight, lower, offsetBits);
         }
         return bins;

reader/src/main/java/io/github/dfa1/vortex/reader/decode/PcoTansDecoder.java

@@ -33,12 +33,17 @@ private PcoTansDecoder(int[] nextStateIdxBase, int[] bitsToRead,
     ///
     /// Port of `Spec::from_weights` + `Decoder::new` from pcodec.
     public static PcoTansDecoder build(int ansSizeLog, PcoBin[] bins) {
+        int tableSize = 1 << ansSizeLog;
         if (bins.length == 0) {
-            // Degenerate: no bins → 1-state table, all offsets zero.
-            return new PcoTansDecoder(new int[]{0}, new int[]{0}, new int[]{0}, new long[]{0L});
+            // Degenerate: no bins → every state decodes to offset zero. Sized to tableSize
+            // (not a fixed 1-state table): the initial ANS state indices a page carries are
+            // read with ansSizeLog bits (so any value in [0, tableSize) is possible) before
+            // this decoder is consulted — a corrupt file pairing zero bins with a nonzero
+            // ansSizeLog previously indexed a real 1-entry table out of bounds, a raw
+            // ArrayIndexOutOfBoundsException instead of a VortexException (ADR 0003).
+            return new PcoTansDecoder(new int[tableSize], new int[tableSize], new int[tableSize], new long[tableSize]);
         }
 
-        int tableSize = 1 << ansSizeLog;
         int[] weights = new int[bins.length];
         for (int i = 0; i < bins.length; i++) {
             weights[i] = bins[i].weight();

reader/src/main/java/io/github/dfa1/vortex/reader/decode/RunEndEncodingDecoder.java

@@ -52,9 +52,23 @@ public Array decode(DecodeContext ctx) {
         long offset = meta.offset();
 
         long n = ctx.rowCount();
+        if (numRuns < 0) {
+            throw new VortexException(EncodingId.VORTEX_RUNEND, "runend: negative num_runs " + numRuns);
+        }
+        if (numRuns == 0 && n > 0) {
+            // Zero runs cover no rows — a crafted file pairing that with a non-empty row
+            // count previously decoded "successfully" into a LazyRunEndXxxArray backed by
+            // an empty ends/values child, then threw a raw IndexOutOfBoundsException (or
+            // ArithmeticException via the % elementCount broadcast path) on first read
+            // instead of failing here as a VortexException.
+            throw new VortexException(EncodingId.VORTEX_RUNEND,
+                    "runend: zero runs cannot cover " + n + " row(s)");
+        }
         DType endsDtype = new DType.Primitive(endsPtype, false);
         Array endsArr = ctx.decodeChild(0, endsDtype, numRuns);
         Array endsData = endsArr instanceof MaskedArray m ? m.inner() : endsArr;
+        MemorySegment endsSeg = ctx.materialize(endsData);
+        validateEnds(endsSeg, endsPtype, numRuns, offset, n);
 
         // Values-side validity mirrors the Rust reference `ValidityVTable<RunEnd>`: a
         // RunEnd array's validity IS a RunEnd over the same ends whose per-run value is
@@ -72,7 +86,6 @@ public Array decode(DecodeContext ctx) {
         }
 
         if (ctx.dtype() instanceof DType.Utf8 || ctx.dtype() instanceof DType.Binary) {
-            MemorySegment endsSeg = ctx.materialize(endsData);
             Array result = expandStrings(endsSeg, VarBinArray.toOffsetMode((VarBinArray) valuesData, ctx.arena()),
                     endsPtype, numRuns, offset, n, ctx.dtype(), ctx.arena());
             return withRunValidity(result, valuesValidity, endsData, n, offset);
@@ -119,6 +132,41 @@ private static Array withRunValidity(Array result, BoolArray valuesValidity, Arr
         return new MaskedArray(result, rowValidity);
     }
 
+    /// Validates `ends` against the format's write-side contract that the reference reader does
+    /// not itself enforce — the spec's note on this encoding is explicit: "a conformant reader
+    /// SHOULD validate \[strict-increase and the two-children shape\] itself rather than assume
+    /// them" (`encoding-format/dict-runend-sparse.md` §RunEnd). One O(numRuns) pass checks:
+    /// `ends` strictly increasing; `ends[0] >= offset` when sliced; and `ends[numRuns-1] >=
+    /// offset + n` — the runs must cover the full requested window (trailing runs beyond it are
+    /// legal per the offset-aware slicing model and simply go unused, so this is `>=`, not `==`).
+    /// Every violation here previously decoded without error and either silently repeated the
+    /// last run's value past where the data actually ends, or (for `ends[0] < offset`) resolved a
+    /// negative index.
+    private static void validateEnds(MemorySegment endsSeg, PType endsPtype, long numRuns, long offset, long n) {
+        long endsCap = SegmentBroadcast.capacity(endsSeg, endsPtype.byteSize());
+        if (endsCap <= 0) {
+            throw new VortexException(EncodingId.VORTEX_RUNEND,
+                    "runend: empty ends buffer for " + numRuns + " run(s)");
+        }
+        long prev = readUnsigned(endsSeg, 0, endsPtype);
+        if (offset != 0 && prev < offset) {
+            throw new VortexException(EncodingId.VORTEX_RUNEND,
+                    "runend: ends[0]=" + prev + " < offset " + offset);
+        }
+        for (long i = 1; i < numRuns; i++) {
+            long end = readUnsigned(endsSeg, i % endsCap, endsPtype);
+            if (end <= prev) {
+                throw new VortexException(EncodingId.VORTEX_RUNEND,
+                        "runend: ends not strictly increasing at run " + i + " (" + end + " <= " + prev + ")");
+            }
+            prev = end;
+        }
+        if (prev < offset + n) {
+            throw new VortexException(EncodingId.VORTEX_RUNEND,
+                    "runend: last end " + prev + " does not cover offset+n=" + (offset + n));
+        }
+    }
+
     private static Array expandStrings(
             MemorySegment endsSeg, VarBinArray.OffsetMode valuesArr,
             PType endsPtype, long numRuns, long offset, long n,

reader/src/test/java/io/github/dfa1/vortex/reader/ScanIteratorZoneCountAdversarialTest.java

@@ -0,0 +1,104 @@
+package io.github.dfa1.vortex.reader;
+
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.LayoutId;
+import io.github.dfa1.vortex.reader.layout.Layout;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.lang.foreign.Arena;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
+
+/// A `vortex.stats` (zoned) layout's zone-map table row count is its own layout metadata field
+/// (`statsFlat.rowCount()`), never bounds-checked at parse time or cross-checked against the
+/// data layout's actual chunk count (see [ScanIterator#columnZoneStats] Javadoc — the two are
+/// deliberately decoupled). [ScanIterator] previously cast that attacker-controlled row count
+/// straight to `int` to size an `ArrayList`: a negative value threw a raw
+/// `IllegalArgumentException` and a value just over `Integer.MAX_VALUE` wrapped to negative on
+/// the cast, both instead of the documented "fall back to per-chunk stats" behavior.
+@ExtendWith(MockitoExtension.class)
+class ScanIteratorZoneCountAdversarialTest {
+
+    private static final ColumnName COLUMN = ColumnName.of("v");
+    private static final DType.Struct SCHEMA = new DType.Struct(List.of(COLUMN), List.of(DType.I64), false);
+
+    @Mock
+    private VortexHandle file;
+
+    @ParameterizedTest
+    @ValueSource(longs = {-1L, Long.MIN_VALUE, ((long) Integer.MAX_VALUE) + 1L, Long.MAX_VALUE})
+    void corruptZoneCount_fallsBackInsteadOfCrashing(long corruptZoneCount) {
+        // Given — a one-chunk file whose zone-map table declares a corrupt row count
+        Layout root = rootLayout(corruptZoneCount);
+        Footer footer = new Footer(List.of(), List.of(),
+                List.of(new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE),
+                        new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE)),
+                List.of());
+        given(file.dtype()).willReturn(SCHEMA);
+        given(file.layout()).willReturn(root);
+        given(file.footer()).willReturn(footer);
+
+        // When
+        List<ArrayStats> result;
+        try (ScanIterator sut = new ScanIterator(file, ScanOptions.columns("v"))) {
+            result = sut.columnZoneStats("v");
+        }
+
+        // Then — degrades to the per-chunk fallback (one empty entry per chunk), no raw exception
+        assertThat(result).hasSize(1);
+        assertThat(result.getFirst()).isEqualTo(ArrayStats.empty());
+    }
+
+    @Test
+    void plausibleZoneCount_isNotRejected() {
+        // Given — a small, legitimate-looking zone count on an otherwise-corrupt (headerless)
+        // stats segment, which still degrades gracefully once decoding is attempted
+        Layout root = rootLayout(1L);
+        Footer footer = new Footer(List.of(), List.of(),
+                List.of(new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE),
+                        new SegmentSpec(0, 0, (byte) 0, CompressionScheme.NONE)),
+                List.of());
+        given(file.dtype()).willReturn(SCHEMA);
+        given(file.layout()).willReturn(root);
+        given(file.footer()).willReturn(footer);
+
+        // When
+        List<ArrayStats> result;
+        try (ScanIterator sut = new ScanIterator(file, ScanOptions.columns("v"))) {
+            result = sut.columnZoneStats("v");
+        }
+
+        // Then — the guard only rejects implausible counts; this one reaches the normal decode
+        // path (which itself falls back gracefully on the segment's missing content)
+        assertThat(result).hasSize(1);
+    }
+
+    /// Builds `Struct(v) -> Zoned[Flat(data, empty segment 0), Flat(stats, rowCount=zoneCount,
+    /// segment 1)]`. The data flat's zero-length segment makes the per-chunk fallback resolve to
+    /// [ArrayStats#empty()] without needing real FlatBuffer bytes.
+    private static Layout rootLayout(long zoneCount) {
+        Layout dataFlat = new Layout(LayoutId.FLAT, 5, null, List.of(), List.of(0));
+        Layout statsFlat = new Layout(LayoutId.FLAT, zoneCount, minStatBitset(), List.of(), List.of(1));
+        Layout zoned = new Layout(LayoutId.STATS, 5, null, List.of(dataFlat, statsFlat), List.of());
+        return new Layout(LayoutId.STRUCT, 5, null, List.of(zoned), List.of());
+    }
+
+    /// `vortex.stats` metadata: 4-byte zone length (unused here) + a bitset with the `MIN` bit
+    /// (ordinal 4) set, so [io.github.dfa1.vortex.reader.layout.ZonedStatsSchema#statsTableDtype]
+    /// resolves a non-empty schema and the code under test proceeds past its early-return guards.
+    private static MemorySegment minStatBitset() {
+        MemorySegment seg = Arena.ofAuto().allocate(5);
+        seg.set(ValueLayout.JAVA_BYTE, 4, (byte) 0x10);
+        return seg;
+    }
+}