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

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>

Author: dfa1

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - 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))
 - Scans of files whose columns use different chunk grids no longer fail with `mixed per-column chunking beyond 1-vs-N is not supported`. The scan planner now splits at the merged boundary grid — the sorted union of every column's chunk boundaries, matching the Rust reference — and decodes each column's covering chunk once, slicing it zero-copy to each window. This handles both nested grids (Raincloud `emotions-dataset-for-nlp`, where `label`'s coarse chunks nest inside `text`'s finer grid) and disjoint grids (`uci-beijing-multi-site-air-quality`, where numeric and `station` boundaries do not nest). Aligned N-vs-N and 1-vs-N scans keep their existing slice-free fast path. ([#221](https://github.com/dfa1/vortex-java/issues/221))
+- 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))

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. |

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;