fix(reader): decode vortex.zoned aggregate-spec zone maps (vortex-jni 0.76.0)

Bumping vortex-jni to 0.76.0 broke RustWritesJavaReadsIntegrationTest
#jniWriter_perZoneSum_readFromZoneMapTable with a ClassCastException
(DType$Bool -> DType$Primitive) while decoding the per-zone stats table.

0.76.0's zoned layout (spiraldb/vortex #7938) splits the format in two: the
canonical `vortex.zoned` id now stores an ordered aggregate-function spec list
in its metadata (proto: version byte + ZonedMetadataProto) and its stats table
drops the legacy per-stat `_is_truncated` Bool flags, while the legacy
`vortex.stats` alias keeps the Stat-bitset metadata. Our reader treated the two
ids as pure aliases and always reconstructed the legacy schema, so the extra
Bool fields no longer matched the encoded table and the positional struct decode
mismatched.

The reader now dispatches on the layout id: `vortex.zoned` goes through a new
ZonedStatsSchema.aggregateStatsTableDtype that parses the aggregate specs and
maps each known aggregate (max/min/sum/null_count/nan_count/...) to its stored
stat dtype, with no truncation flags; `vortex.stats` keeps the bitset path.
Unknown/nested-state aggregates (e.g. vortex.bounded_max) bail to the per-chunk
fallback rather than decode a misaligned table. The CLI stats inspector uses the
same dispatch. This is a genuine compatibility fix on top of the dependency
bump, not a rebase-conflict resolution (the rebase onto main was clean).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dfa1

CHANGELOG.md

@@ -9,6 +9,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))
 - 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))

cli/src/main/java/io/github/dfa1/vortex/cli/tui/VortexInspectorTui.java

@@ -4,6 +4,7 @@
 import io.github.dfa1.vortex.reader.SegmentSpec;
 import io.github.dfa1.vortex.core.model.DType;
 import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.LayoutId;
 import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.cli.tui.term.Ansi;
 import io.github.dfa1.vortex.cli.tui.term.Key;
@@ -673,8 +674,12 @@ private void runStatsLoad(InspectorTree.Node anchor) {
                     statsCache.put(anchor, new DataState.Failed("no column dtype"));
                     return;
                 }
-                DType.Struct statsDtype = ZonedStatsSchema.statsTableDtype(columnDtype, anchorLayout.metadata());
-                if (statsDtype.fieldNames().isEmpty()) {
+                // `vortex.zoned` (Rust >= 0.76) carries aggregate-spec metadata; the legacy
+                // `vortex.stats` alias carries a Stat bitset — reconstruct the schema accordingly.
+                DType.Struct statsDtype = anchorLayout.layoutId() == LayoutId.ZONED
+                        ? ZonedStatsSchema.aggregateStatsTableDtype(columnDtype, anchorLayout.metadata())
+                        : ZonedStatsSchema.statsTableDtype(columnDtype, anchorLayout.metadata());
+                if (statsDtype == null || statsDtype.fieldNames().isEmpty()) {
                     statsCache.put(anchor, new DataState.Failed("no stats present in metadata"));
                     return;
                 }

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

@@ -3,6 +3,7 @@
 import static io.github.dfa1.vortex.core.io.VortexFormat.LE_INT;
 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.core.io.IoBounds;
 import io.github.dfa1.vortex.core.error.VortexException;
 import io.github.dfa1.vortex.reader.array.Array;
@@ -395,10 +396,11 @@ public List<ArrayStats> columnZoneStats(String column) {
         return out;
     }
 
-    /// Decodes the column's `vortex.stats` zone-map table into one [ArrayStats] per zone, or
-    /// returns `null` when the column has no zone map (so the caller falls back to per-chunk
-    /// node stats). The table is a single flat segment encoding a struct with a subset of the
-    /// `min`/`max`/`sum`/`null_count` fields (see [ZonedStatsSchema]); it is decoded into a
+    /// Decodes the column's zone-map table into one [ArrayStats] per zone, or returns `null` when
+    /// the column has no zone map (so the caller falls back to per-chunk node stats). The table is
+    /// a single flat segment encoding a struct with a subset of the `min`/`max`/`sum`/`null_count`
+    /// fields; the schema is reconstructed from the layout metadata (legacy `vortex.stats` bitset
+    /// or newer `vortex.zoned` aggregate specs — see [ZonedStatsSchema]). It is decoded into a
     /// short-lived confined arena and the scalar values are boxed out before the arena closes.
     private List<ArrayStats> decodeZoneTable(ColumnName column) {
         Layout zoned = findZonedLayout(file.layout(), column);
@@ -417,7 +419,15 @@ private List<ArrayStats> decodeZoneTable(ColumnName column) {
         if (segIdx < 0 || segIdx >= file.footer().segmentSpecs().size()) {
             return null;
         }
-        DType.Struct statsDtype = ZonedStatsSchema.statsTableDtype(columnDtype, zoned.metadata());
+        // The canonical `vortex.zoned` id (Rust >= 0.76) stores an aggregate-spec metadata blob;
+        // the legacy `vortex.stats` alias (what vortex-java writes) stores a Stat bitset. They
+        // reconstruct the table schema differently — dispatch on the layout id.
+        DType.Struct statsDtype = zoned.layoutId() == LayoutId.ZONED
+                ? ZonedStatsSchema.aggregateStatsTableDtype(columnDtype, zoned.metadata())
+                : ZonedStatsSchema.statsTableDtype(columnDtype, zoned.metadata());
+        if (statsDtype == null || statsDtype.fieldNames().isEmpty()) {
+            return null;
+        }
         long nZones = statsFlat.rowCount();
         SegmentSpec spec = file.footer().segmentSpecs().get(segIdx);
         try (Arena tableArena = Arena.ofConfined()) {

reader/src/main/java/io/github/dfa1/vortex/reader/layout/ZonedStatsSchema.java

@@ -135,6 +135,52 @@ public static DType.Struct statsTableDtype(DType columnDtype, MemorySegment meta
         return statsTableDtype(columnDtype, presentStats(metadata));
     }
 
+    /// Reconstructs the per-zone stats-table dtype for a newer `vortex.zoned` layout, whose
+    /// metadata carries an ordered list of aggregate-function specs instead of the legacy
+    /// [Stat] bitset.
+    ///
+    /// The shape mirrors Rust's `aggregate_stats_table_dtype`
+    /// ([vortex-layout/src/layouts/zoned/schema.rs](https://github.com/spiraldb/vortex/blob/develop/vortex-layout/src/layouts/zoned/schema.rs)):
+    /// for every aggregate whose state dtype is defined for the column, append one
+    /// `(name, nullable state-dtype)` field, in spec order — no `_is_truncated` flags. Field
+    /// names are the aggregate's canonical [Stat#fieldName()] so the reader can look them up by
+    /// stat; the struct decode itself is positional, so alignment with the encoded table is what
+    /// matters.
+    ///
+    /// Returns `null` when the metadata is not a decodable aggregate-spec blob or names an
+    /// aggregate this reader cannot map to a [Stat] (e.g. `vortex.bounded_max`, whose state is a
+    /// nested struct). Bailing keeps the positional schema faithful: the caller falls back to
+    /// per-chunk stats rather than decoding a misaligned table.
+    ///
+    /// @param columnDtype the column's logical dtype (the `data` child's dtype)
+    /// @param metadata    raw `vortex.zoned` layout metadata
+    /// @return reconstructed non-nullable struct dtype, or `null` when it cannot be reconstructed
+    public static DType.Struct aggregateStatsTableDtype(DType columnDtype, MemorySegment metadata) {
+        List<String> aggregateIds = aggregateIds(metadata);
+        if (aggregateIds == null) {
+            return null;
+        }
+        List<ColumnName> names = new ArrayList<>(aggregateIds.size());
+        List<DType> types = new ArrayList<>(aggregateIds.size());
+        for (String aggregateId : aggregateIds) {
+            Stat stat = statForAggregate(aggregateId);
+            if (stat == null) {
+                // Unknown aggregate — the encoded table has a field here we cannot describe, so
+                // any reconstruction would be positionally misaligned. Bail to the fallback path.
+                return null;
+            }
+            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;
+            }
+            names.add(ColumnName.of(stat.fieldName()));
+            types.add(stype.withNullable(true));
+        }
+        return new DType.Struct(List.copyOf(names), List.copyOf(types), false);
+    }
+
     /// Reconstructs the per-zone stats-table dtype for the given column dtype
     /// and explicit stat list (already decoded from metadata).
     ///
@@ -231,4 +277,170 @@ private static DType sumDtype(DType columnDtype) {
         // so the stat is dropped from the schema rather than causing a decode mismatch.
         return null;
     }
+
+    /// First byte of `vortex.zoned` metadata: the protobuf envelope version Rust writes.
+    private static final int AGGREGATE_METADATA_VERSION = 1;
+
+    /// Maps a well-known aggregate-function id (Rust `AggregateFnId`) to the [Stat] whose stored
+    /// dtype and canonical field name it uses in the zone-map table, or `null` when the reader
+    /// has no faithful mapping (nested-state aggregates like `vortex.bounded_max`, or functions
+    /// not stored as a scalar stat).
+    private static Stat statForAggregate(String aggregateId) {
+        return switch (aggregateId) {
+            case "vortex.max" -> Stat.MAX;
+            case "vortex.min" -> Stat.MIN;
+            case "vortex.sum" -> Stat.SUM;
+            case "vortex.null_count" -> Stat.NULL_COUNT;
+            case "vortex.nan_count" -> Stat.NAN_COUNT;
+            case "vortex.is_constant" -> Stat.IS_CONSTANT;
+            case "vortex.is_sorted" -> Stat.IS_SORTED;
+            case "vortex.is_strict_sorted" -> Stat.IS_STRICT_SORTED;
+            case "vortex.uncompressed_size_in_bytes" -> Stat.UNCOMPRESSED_SIZE_IN_BYTES;
+            default -> null;
+        };
+    }
+
+    /// Extracts the ordered aggregate-function ids from `vortex.zoned` metadata.
+    ///
+    /// The metadata is a version byte (value [#AGGREGATE_METADATA_VERSION]) followed by a
+    /// protobuf `ZonedMetadataProto { uint32 zone_len = 1; repeated AggregateSpecProto
+    /// aggregate_specs = 2; }`, where `AggregateSpecProto { string id = 1; bytes options = 2; }`.
+    /// Only the `id` of each spec is needed to reconstruct the table schema.
+    ///
+    /// Returns `null` when the blob is empty, carries an unsupported version, or is not
+    /// well-formed protobuf — signalling the caller to fall back rather than trust a partial parse.
+    private static List<String> aggregateIds(MemorySegment metadata) {
+        if (metadata == null || metadata.byteSize() < 1) {
+            return null;
+        }
+        if ((metadata.get(ValueLayout.JAVA_BYTE, 0) & 0xff) != AGGREGATE_METADATA_VERSION) {
+            return null;
+        }
+        ProtoCursor cursor = new ProtoCursor(metadata, 1, metadata.byteSize());
+        List<String> ids = new ArrayList<>();
+        while (cursor.hasRemaining()) {
+            long tag = cursor.readVarint();
+            if (tag < 0) {
+                return null;
+            }
+            int fieldNumber = (int) (tag >>> 3);
+            int wireType = (int) (tag & 0x7);
+            if (fieldNumber == 2 && wireType == ProtoCursor.WIRE_LEN) {
+                String id = cursor.readAggregateSpecId();
+                if (id == null) {
+                    return null;
+                }
+                ids.add(id);
+            } else if (!cursor.skipField(wireType)) {
+                return null;
+            }
+        }
+        return List.copyOf(ids);
+    }
+
+    /// Minimal forward cursor over a protobuf byte range within a [MemorySegment]. Reads only the
+    /// varints and length-delimited fields the zoned metadata uses; any malformed or unexpected
+    /// wire shape is reported as a failure so the caller can bail rather than misread.
+    private static final class ProtoCursor {
+        static final int WIRE_VARINT = 0;
+        static final int WIRE_I64 = 1;
+        static final int WIRE_LEN = 2;
+        static final int WIRE_I32 = 5;
+
+        private final MemorySegment segment;
+        private long pos;
+        private final long end;
+
+        ProtoCursor(MemorySegment segment, long start, long end) {
+            this.segment = segment;
+            this.pos = start;
+            this.end = end;
+        }
+
+        boolean hasRemaining() {
+            return pos < end;
+        }
+
+        /// Reads a base-128 varint, or `-1` when it runs past the end or exceeds 64 bits.
+        long readVarint() {
+            long result = 0;
+            int shift = 0;
+            while (pos < end) {
+                int b = segment.get(ValueLayout.JAVA_BYTE, pos) & 0xff;
+                pos++;
+                if (shift < 64) {
+                    result |= (long) (b & 0x7f) << shift;
+                }
+                if ((b & 0x80) == 0) {
+                    return result;
+                }
+                shift += 7;
+                if (shift > 63) {
+                    return -1;
+                }
+            }
+            return -1;
+        }
+
+        /// Skips a field of the given wire type, returning `false` on an unknown type or overrun.
+        boolean skipField(int wireType) {
+            return switch (wireType) {
+                case WIRE_VARINT -> readVarint() >= 0;
+                case WIRE_I64 -> advance(8);
+                case WIRE_I32 -> advance(4);
+                case WIRE_LEN -> {
+                    long len = readVarint();
+                    yield len >= 0 && advance(len);
+                }
+                default -> false;
+            };
+        }
+
+        /// Reads an `AggregateSpecProto` length-delimited message and returns its `id` (field 1),
+        /// or `null` if the message is malformed or carries no id.
+        String readAggregateSpecId() {
+            long len = readVarint();
+            if (len < 0 || pos + len > end) {
+                return null;
+            }
+            long messageEnd = pos + len;
+            String id = null;
+            while (pos < messageEnd) {
+                long tag = readVarint();
+                if (tag < 0) {
+                    return null;
+                }
+                int fieldNumber = (int) (tag >>> 3);
+                int wireType = (int) (tag & 0x7);
+                if (fieldNumber == 1 && wireType == WIRE_LEN) {
+                    id = readString(messageEnd);
+                    if (id == null) {
+                        return null;
+                    }
+                } else if (!skipField(wireType)) {
+                    return null;
+                }
+            }
+            return id;
+        }
+
+        private String readString(long limit) {
+            long len = readVarint();
+            if (len < 0 || pos + len > limit) {
+                return null;
+            }
+            byte[] bytes = new byte[(int) len];
+            MemorySegment.copy(segment, ValueLayout.JAVA_BYTE, pos, bytes, 0, (int) len);
+            pos += len;
+            return new String(bytes, java.nio.charset.StandardCharsets.UTF_8);
+        }
+
+        private boolean advance(long count) {
+            if (count < 0 || pos + count > end) {
+                return false;
+            }
+            pos += count;
+            return true;
+        }
+    }
 }