fix: harden vortex.zoned metadata parsing against malformed input; decimal sum fallback (#197)

dfa1 · claude · dfa1 · commit febfcc1d70fd · 2026-07-07T08:03:06.000+02:00
The aggregate-spec zone-map decoder added in #197 (commit 9e20edc) had two post-merge review findings. Overflow class: ProtoCursor guarded reads with `pos + len > bound`. A varint length up to Long.MAX_VALUE overflows `pos + len` to negative, so the `> bound` check is false and the guard passes — then `new byte[(int) len]` throws NegativeArraySizeException, or `pos` advances negative and the next segment read throws IndexOutOfBoundsException. This is untrusted file metadata (ADR 0003), so it must never surface a raw JDK exception. Every length/offset check (readAggregateSpecId message descent, readString, advance/skipField) is rewritten overflow-safe: the invariant pos <= bound holds throughout, so `bound - pos >= 0` never overflows and the guard becomes `len < 0 || len > bound - pos`. Malformed metadata now falls back to per-chunk stats. Decimal divergence: Rust's default_zoned_aggregate_fns (vortex-layout zoned/writer.rs) emits a `sum` column whenever Sum.return_dtype is Some — which is Some(Decimal(precision + 10)) for a Decimal column (sum/mod.rs) — while our sumDtype returns null for Decimal. Dropping that field with `continue` reconstructs {max,min,null_count} for an encoded {max,min,sum,null_count} table, so the positional decode reads null_count out of the sum buffer. The reconstructor now distinguishes "Rust also omits this" (nan_count over non-float per nan_count/mod.rs, min/max over an unsupported column per min_max minmax_supported_dtype — safe skip) from "Rust keeps a field we cannot map" (decimal sum, bounded_max/min nested state — return null so the whole column falls back to correct-but-unpruned per-chunk stats). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - Per-zone stats from current Rust writers (`vortex.zoned`, vortex-jni 0.76.0) decode again. The 0.76 zoned layout replaced the legacy `vortex.stats` bit-set metadata with an aggregate-function spec list and dropped the per-stat truncation flags; the reader now reconstructs the stats table from that spec list (min/max/sum/null_count), so `columnZoneStats` and aggregate push-down work against those files instead of throwing `ClassCastException`. ([#197](https://github.com/dfa1/vortex-java/pull/197))
+- Hardened the `vortex.zoned` metadata decoder against malformed input: an attacker-controlled length varint could overflow its `pos + len` bounds checks and crash with `NegativeArraySizeException`/`IndexOutOfBoundsException`; the checks are now overflow-safe and unparseable metadata falls back to per-chunk stats. Decimal columns — whose Rust zone table keeps a `sum` field this reader cannot map — now also fall back instead of decoding a misaligned table. ([#197](https://github.com/dfa1/vortex-java/pull/197))
 - CSV export renders nested struct columns as JSON object cells (`{"field":value,...}`, strings JSON-escaped, null fields as JSON `null`, nested structs recursed) instead of throwing `unsupported array type: StructArray`. ([#217](https://github.com/dfa1/vortex-java/issues/217))
 - Scanning a struct whose columns include a nested struct no longer fails chunk planning. A nested `vortex.struct` layout column (e.g. Raincloud `countries-of-the-world`'s `data` column) is now treated as a single full-range chunk source and decoded through the layout registry's new struct decoder into a `StructArray`, matching the Rust reference (a nested struct spans its parent's row range, not an independent chunking). `schema`/`count`/`inspect` already worked; plain scans and `select` of sibling columns now work too. CSV export of the struct column itself remains unsupported (a separate rendering limitation). ([#207](https://github.com/dfa1/vortex-java/issues/207))
 - CSV export renders unsigned integer columns (U8–U64) with their unsigned values — high-half values previously printed as two's-complement negatives (uci-wine `magnesium` U8 132 exported as -124), silent corruption found by the Raincloud conformance suite. ([#208](https://github.com/dfa1/vortex-java/issues/208))
diff --git a/docs/explanation.md b/docs/explanation.md
@@ -222,7 +222,7 @@ file-level `SegmentSpec[]` table). Five node types exist today:
 | ID              | Constant  | Children  | Role |
 |-----------------|-----------|-----------|------|
 | `vortex.struct` | `STRUCT`  | N         | Row type. One child per column. Root of every file. |
-| `vortex.stats`  | `ZONED`   | 1         | Wraps a child layout and carries per-chunk min/max as zone maps. Pruned at scan time when filter predicate falls outside `[min, max]`. |
+| `vortex.stats`  | `ZONED`   | 1         | Wraps a child layout and carries a per-zone stats table as a zone map. The legacy `vortex.stats` form declares its columns with a `Stat` bitset; the Rust >= 0.76 `vortex.zoned` form declares them with an aggregate-spec list, adding `sum`/`null_count` alongside `min`/`max`. Pruned at scan time when the filter predicate falls outside `[min, max]`. |
 | `vortex.chunked`| `CHUNKED` | M (+1)    | Row-group sequence. Optional stats child at index 0 when `metadata[0] == 1` (per-chunk stats sidecar); remaining children are the data chunks. |
 | `vortex.dict`   | `DICT`    | 2         | Dictionary-encoded leaf. `children[0]` = values layout, `children[1]` = codes layout. `metadata` holds the codes `PType` (varint, proto field 1). Decoder gathers values by code. |
 | `vortex.flat`   | `FLAT`    | 0         | Leaf. References one `SegmentSpec` via `segments[0]`. Decoded by the encoding named in the segment's `arraySpec`, not by `encodingId` itself — see below. |
diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java b/reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java
@@ -29,9 +29,12 @@
 ///       each get an extra trailing field `max_is_truncated` /
 ///       `min_is_truncated` of type `Bool` (non-nullable).
 ///
-/// `Sum` widening rules and Decimal handling are not yet implemented —
-/// when the column dtype has no resolvable stat dtype the stat is skipped so
-/// the inspector degrades to "no schema" rather than failing.
+/// `Sum` widening for Decimal columns is not yet modeled. The legacy bitset path
+/// ([#statsTableDtype(DType, List)]) skips any stat with no resolvable dtype so the inspector
+/// degrades to "no schema" rather than failing. The aggregate-spec path
+/// ([#aggregateStatsTableDtype(DType, MemorySegment)]) is positional, so it only skips fields Rust
+/// also omits and otherwise returns `null` (fall back to per-chunk stats) rather than emit a
+/// misaligned struct.
 public final class ZonedStatsSchema {
 
     /// Ordinal positions of [Stat] in the Rust enum — kept stable across
@@ -171,16 +174,52 @@ public static DType.Struct aggregateStatsTableDtype(DType columnDtype, MemorySeg
             }
             DType stype = statDtype(stat, columnDtype);
             if (stype == null) {
-                // Rust's schema drops aggregates with no state dtype for this column (e.g.
-                // nan_count over a non-float); skip to stay aligned with the encoded table.
-                continue;
+                if (rustAlsoOmits(stat, columnDtype)) {
+                    // Rust's aggregate_stats_table_dtype also drops this field (its aggregate's
+                    // state_dtype is None for this column), so skipping stays aligned with the
+                    // encoded table.
+                    continue;
+                }
+                // Rust keeps a field here that we cannot map — notably Sum over a Decimal column,
+                // whose state widens to Decimal(precision + 10) (writer.rs default_zoned_aggregate_fns
+                // only emits Sum when Sum.return_dtype is Some, which it is for Decimal). Dropping it
+                // would misalign the positional decode, so bail to per-chunk stats.
+                return null;
             }
             names.add(ColumnName.of(stat.fieldName()));
             types.add(stype.withNullable(true));
         }
         return new DType.Struct(List.copyOf(names), List.copyOf(types), false);
     }
 
+    /// Reports whether Rust's `aggregate_stats_table_dtype` also omits this aggregate for the given
+    /// column — i.e. the aggregate's `state_dtype` is `None` there, so dropping it keeps our
+    /// positional schema aligned with the encoded table. Any other null resolution means Rust keeps
+    /// a field we cannot describe, and the caller must bail instead.
+    ///
+    /// The two legitimate drops mirror the Rust reference
+    /// ([vortex-array/src/aggregate_fn/fns](https://github.com/spiraldb/vortex/tree/develop/vortex-array/src/aggregate_fn/fns)):
+    /// - `nan_count` returns `None` for non-float columns (`nan_count/mod.rs`);
+    /// - `min`/`max` return `None` when `minmax_supported_dtype` is false — for us that is only
+    ///       `DType.Null`, the sole column for which the min/max resolver yields null.
+    private static boolean rustAlsoOmits(Stat stat, DType columnDtype) {
+        return switch (stat) {
+            case NAN_COUNT -> !isFloatingColumn(columnDtype);
+            case MAX, MIN -> true;
+            default -> false;
+        };
+    }
+
+    private static boolean isFloatingColumn(DType columnDtype) {
+        if (columnDtype instanceof DType.Primitive p) {
+            return p.ptype().isFloating();
+        }
+        if (columnDtype instanceof DType.Extension ext) {
+            return isFloatingColumn(ext.storageDType());
+        }
+        return false;
+    }
+
     /// Reconstructs the per-zone stats-table dtype for the given column dtype
     /// and explicit stat list (already decoded from metadata).
     ///
@@ -400,7 +439,10 @@ boolean skipField(int wireType) {
         /// or `null` if the message is malformed or carries no id.
         String readAggregateSpecId() {
             long len = readVarint();
-            if (len < 0 || pos + len > end) {
+            // Overflow-safe bound: pos <= end holds throughout, so end - pos never overflows; a
+            // varint-derived len up to Long.MAX_VALUE would overflow pos + len to negative and slip
+            // past a pos + len > end guard, then advance pos past the segment.
+            if (len < 0 || len > end - pos) {
                 return null;
             }
             long messageEnd = pos + len;
@@ -426,7 +468,11 @@ String readAggregateSpecId() {
 
         private String readString(long limit) {
             long len = readVarint();
-            if (len < 0 || pos + len > limit) {
+            // Overflow-safe bound (see readAggregateSpecId): compare against limit - pos, never
+            // pos + len. If a prior varint pushed pos past limit, limit - pos is negative and any
+            // non-negative len is rejected, so we never size new byte[(int) len] from an
+            // attacker-controlled, overflowed length.
+            if (len < 0 || len > limit - pos) {
                 return null;
             }
             byte[] bytes = new byte[(int) len];
@@ -436,7 +482,10 @@ private String readString(long limit) {
         }
 
         private boolean advance(long count) {
-            if (count < 0 || pos + count > end) {
+            // Overflow-safe bound: pos <= end holds throughout, so end - pos never overflows,
+            // whereas pos + count could overflow to negative for a varint-derived count near
+            // Long.MAX_VALUE and slip past a pos + count > end guard.
+            if (count < 0 || count > end - pos) {
                 return false;
             }
             pos += count;
diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchemaTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchemaTest.java
@@ -256,17 +256,17 @@ void buildsNumericSchemaWithoutTruncationFlags() {
                     "vortex.max", "vortex.min", "vortex.sum", "vortex.nan_count", "vortex.null_count");
 
             // When
-            DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta);
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta);
 
             // Then — no `_is_truncated` fields (the new format has none) and nan_count is dropped
-            assertThat(schema.fieldNames().stream().map(ColumnName::value).toList())
+            assertThat(result.fieldNames().stream().map(ColumnName::value).toList())
                     .containsExactly("max", "min", "sum", "null_count");
-            assertThat(schema.fieldTypes()).containsExactly(
+            assertThat(result.fieldTypes()).containsExactly(
                     new DType.Primitive(PType.I64, true),
                     new DType.Primitive(PType.I64, true),
                     new DType.Primitive(PType.I64, true),
                     new DType.Primitive(PType.U64, true));
-            assertThat(schema.nullable()).isFalse();
+            assertThat(result.nullable()).isFalse();
         }
 
         @Test
@@ -275,12 +275,12 @@ void keepsNanCountForFloatColumn() {
             MemorySegment meta = aggregateMeta(1024, "vortex.max", "vortex.nan_count");
 
             // When
-            DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.F64, meta);
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.F64, meta);
 
             // Then
-            assertThat(schema.fieldNames().stream().map(ColumnName::value).toList())
+            assertThat(result.fieldNames().stream().map(ColumnName::value).toList())
                     .containsExactly("max", "nan_count");
-            assertThat(schema.fieldTypes()).element(1).isEqualTo(new DType.Primitive(PType.U64, true));
+            assertThat(result.fieldTypes()).element(1).isEqualTo(new DType.Primitive(PType.U64, true));
         }
 
         @Test
@@ -290,10 +290,27 @@ void bailsOnUnknownAggregate() {
             MemorySegment meta = aggregateMeta(4096, "vortex.bounded_max", "vortex.null_count");
 
             // When
-            DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.UTF8, meta);
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.UTF8, meta);
 
             // Then
-            assertThat(schema).isNull();
+            assertThat(result).isNull();
+        }
+
+        @Test
+        void bailsForDecimalSumColumn() {
+            // Given — Rust's default_zoned_aggregate_fns emits a `sum` column for a Decimal column
+            // (Sum.return_dtype widens to Decimal(precision + 10)), but this reader cannot map a
+            // decimal sum state. Reconstructing {max, min, null_count} while the encoded table is
+            // {max, min, sum, null_count} would read null_count out of the sum buffer, so the whole
+            // column must fall back to per-chunk (unpruned) stats rather than emit a mis-sized struct.
+            MemorySegment meta = aggregateMeta(8192,
+                    "vortex.max", "vortex.min", "vortex.sum", "vortex.null_count");
+
+            // When
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.decimal(20, 4), meta);
+
+            // Then
+            assertThat(result).isNull();
         }
 
         @Test
@@ -303,19 +320,19 @@ void bailsOnUnsupportedVersion() {
             meta.set(java.lang.foreign.ValueLayout.JAVA_BYTE, 0, (byte) 2);
 
             // When
-            DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta);
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta);
 
             // Then
-            assertThat(schema).isNull();
+            assertThat(result).isNull();
         }
 
         @Test
         void bailsOnNullMetadata() {
             // Given / When — a zoned layout with no metadata cannot be reconstructed
-            DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, null);
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, null);
 
             // Then
-            assertThat(schema).isNull();
+            assertThat(result).isNull();
         }
 
         @Test
@@ -324,11 +341,103 @@ void returnsEmptyStructWhenNoAggregatesPresent() {
             MemorySegment meta = aggregateMeta(8192);
 
             // When
-            DType.Struct schema = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta);
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(DType.I64, meta);
 
             // Then — an empty (non-null) struct; the caller falls back to per-chunk stats
-            assertThat(schema).isNotNull();
-            assertThat(schema.fieldNames()).isEmpty();
+            assertThat(result).isNotNull();
+            assertThat(result.fieldNames()).isEmpty();
+        }
+    }
+
+    @Nested
+    class MalformedAggregateMetadata {
+        // These blobs are untrusted file metadata. Each malformed shape must degrade to the
+        // fallback (null) return — never a raw JDK exception (NegativeArraySizeException /
+        // IndexOutOfBoundsException), which the pre-fix overflow-unsafe bounds checks allowed.
+
+        @Test
+        void bailsOnTruncatedVarint() {
+            // Given — the only field tag is an unterminated varint (continuation bit set, no next
+            // byte); the cursor must fail rather than read past the buffer.
+            java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream();
+            out.write(1); // envelope version
+            out.write(0x80); // dangling varint continuation byte
+
+            // When
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(
+                    DType.I64, MemorySegment.ofArray(out.toByteArray()));
+
+            // Then
+            assertThat(result).isNull();
+        }
+
+        @Test
+        void bailsOnOversizedStringLength() {
+            // Given — a framed AggregateSpecProto whose id-string length varint is Long.MAX_VALUE.
+            // The pre-fix `pos + len > limit` guard overflowed to negative and passed, then
+            // `new byte[(int) len]` threw NegativeArraySizeException.
+            java.io.ByteArrayOutputStream spec = new java.io.ByteArrayOutputStream();
+            writeTag(spec, 1, 2); // id, wire LEN
+            writeVarintLong(spec, Long.MAX_VALUE); // absurd string length
+            byte[] specBytes = spec.toByteArray();
+            java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream();
+            out.write(1); // envelope version
+            writeTag(out, 2, 2); // aggregate_specs, wire LEN
+            writeVarint(out, specBytes.length);
+            out.writeBytes(specBytes);
+
+            // When / Then — must not throw NegativeArraySizeException
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(
+                    DType.I64, MemorySegment.ofArray(out.toByteArray()));
+            assertThat(result).isNull();
+        }
+
+        @Test
+        void bailsOnOversizedSkippedFieldLength() {
+            // Given — a non-target length-delimited field whose length is Long.MAX_VALUE. The
+            // pre-fix advance guard `pos + count > end` overflowed and advanced pos negative,
+            // throwing IndexOutOfBoundsException on the next read.
+            java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream();
+            out.write(1); // envelope version
+            writeTag(out, 1, 2); // field 1 with wire LEN → skipped, not an aggregate spec
+            writeVarintLong(out, Long.MAX_VALUE);
+
+            // When / Then — must not throw IndexOutOfBoundsException
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(
+                    DType.I64, MemorySegment.ofArray(out.toByteArray()));
+            assertThat(result).isNull();
+        }
+
+        @Test
+        void bailsOnUnknownWireType() {
+            // Given — a field using wire type 3 (group start), which the cursor cannot skip.
+            java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream();
+            out.write(1); // envelope version
+            writeTag(out, 3, 3); // field 3, wire type 3 (unsupported)
+
+            // When
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(
+                    DType.I64, MemorySegment.ofArray(out.toByteArray()));
+
+            // Then
+            assertThat(result).isNull();
+        }
+
+        @Test
+        void bailsOnMessageLengthPastEnd() {
+            // Given — an aggregate_specs field whose declared length runs past the buffer end.
+            java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream();
+            out.write(1); // envelope version
+            writeTag(out, 2, 2); // aggregate_specs, wire LEN
+            writeVarint(out, 64); // claims 64 bytes...
+            out.write(0x01); // ...but only one follows
+
+            // When
+            DType.Struct result = ZonedStatsSchema.aggregateStatsTableDtype(
+                    DType.I64, MemorySegment.ofArray(out.toByteArray()));
+
+            // Then
+            assertThat(result).isNull();
         }
     }
 
@@ -372,4 +481,15 @@ private static void writeVarint(java.io.ByteArrayOutputStream out, int value) {
         }
         out.write(v);
     }
+
+    /// Writes an unsigned base-128 varint of a `long` — used to inject absurd (overflow-inducing)
+    /// lengths like `Long.MAX_VALUE` that a plain `int` varint cannot express.
+    private static void writeVarintLong(java.io.ByteArrayOutputStream out, long value) {
+        long v = value;
+        while ((v & ~0x7fL) != 0) {
+            out.write((int) ((v & 0x7f) | 0x80));
+            v >>>= 7;
+        }
+        out.write((int) v);
+    }
 }
