feat(reader): surface per-zone stats from the zone-map table (ADR 0013 §6)

Add ScanIterator.columnZoneStats(col): one ArrayStats per zone with
min/max/sum/null-count, the read-side feed for aggregate push-down.

Sum is decoded from the column's vortex.stats zone-map table rather than
per-flat node stats — matching Rust, whose flat writer retains only
pre-computed stats (flat/writer.rs) and emits SUM only in the zoned table
(zoned/writer.rs). Falls back to per-chunk node stats (sum null) when a
column has no zone map.

- ArrayStats gains a sum component; fromFbs decodes it (forward-compat).
- ZonedStatsSchema moves inspector -> reader so the read path can
  reconstruct the stats-table dtype; cli/inspector imports updated.
- VortexWriter is unchanged functionally (comment only); sum continues to
  live in the zone-map table.

Verified both interop directions, incl. a new test folding per-zone sums
from a Rust-written file to the exact column total.

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

Author: dfa1

TODO.md

@@ -96,6 +96,10 @@ Per-encoding gotchas:
 - [ ] **Compute primitives — masks, kernels, no-materialise** — pushdown filter/compare/aggregate
   kernels operating on Lazy arrays without materialising. See [ADR-0013](docs/adr/0013-compute-primitives.md)
   (Proposed). Gate: a concrete downstream consumer (e.g. the vortex-arrow bridge or filter pushdown).
+  Done: §6 read-side surface — `ScanIterator.columnZoneStats(col)` exposes per-zone
+  min/max/sum/null count, decoding sum from the `vortex.stats` zone-map table (matches files from
+  Rust, whose flat writer omits per-flat sum). Next: `Mask`/`Predicate`/kernel vocab and the
+  two-tier whole-zone+residual reduce; rewire calcite `VortexAggregates.SUM` off full-scan.
 
 ## Encodings
 

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

@@ -9,7 +9,7 @@
 import io.github.dfa1.vortex.cli.tui.term.Terminal;
 import io.github.dfa1.vortex.inspect.ByteSize;
 import io.github.dfa1.vortex.inspect.InspectorTree;
-import io.github.dfa1.vortex.inspect.ZonedStatsSchema;
+import io.github.dfa1.vortex.reader.ZonedStatsSchema;
 import io.github.dfa1.vortex.reader.VortexHandle;
 import io.github.dfa1.vortex.reader.Chunk;
 import io.github.dfa1.vortex.reader.ScanIterator;

inspector/src/main/java/io/github/dfa1/vortex/inspect/VortexInspector.java

@@ -102,7 +102,7 @@ private static ArrayStats aggregateStats(InspectorTree.Node node) {
         if (min == null && max == null) {
             return ArrayStats.empty();
         }
-        return new ArrayStats(min, max, null, null, null, null);
+        return new ArrayStats(min, max, null, null, null, null, null);
     }
 
     @SuppressWarnings({"unchecked", "rawtypes"})

inspector/src/test/java/io/github/dfa1/vortex/inspect/VortexInspectorTest.java

@@ -133,9 +133,9 @@ void render_aggregatesMinMaxAcrossChunks() {
         Layout structLayout = new Layout("vortex.struct", 1000, null, List.of(chunked), List.of());
 
         InspectorTree.Node c1 = new InspectorTree.Node(chunk1, Optional.empty(), Set.of(),
-                new ArrayStats(10L, 50L, null, null, null, null), List.of());
+                new ArrayStats(10L, 50L, null, null, null, null, null), List.of());
         InspectorTree.Node c2 = new InspectorTree.Node(chunk2, Optional.empty(), Set.of(),
-                new ArrayStats(5L, 100L, null, null, null, null), List.of());
+                new ArrayStats(5L, 100L, null, null, null, null, null), List.of());
         InspectorTree.Node chunkedN = new InspectorTree.Node(chunked, Optional.of("id"),
                 Set.of("vortex.flat"), ArrayStats.empty(), List.of(c1, c2));
         InspectorTree.Node rootN = new InspectorTree.Node(structLayout, Optional.empty(),

integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java

@@ -14,6 +14,7 @@
 import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.reader.array.DoubleArray;
 import io.github.dfa1.vortex.reader.array.LongArray;
+import io.github.dfa1.vortex.reader.ArrayStats;
 import io.github.dfa1.vortex.reader.ReadRegistry;
 import io.github.dfa1.vortex.reader.VortexReader;
 import org.apache.arrow.c.ArrowArray;
@@ -269,6 +270,35 @@ void jniWriter_javaReader_singleChunk(@TempDir Path tmp) throws IOException {
         }
     }
 
+    @Test
+    void jniWriter_perZoneSum_readFromZoneMapTable(@TempDir Path tmp) throws IOException {
+        // Given — a Rust-written file large enough that the JNI writer emits a multi-zone column.
+        // Sum lives only in Rust's vortex.stats zone-map table (its flat writer doesn't retain it),
+        // so this proves the Java reader decodes that table for per-zone SUM (ADR 0013 §6 parity).
+        int n = 200_000;
+        long[] ids = new long[n];
+        double[] vals = new double[n];
+        for (int i = 0; i < n; i++) {
+            ids[i] = i;
+            vals[i] = i;
+        }
+        Path file = tmp.resolve("jni_zones.vtx");
+        writeJni(file, ids, vals);
+        long expected = (long) n * (n - 1) / 2; // Σ 0..n-1
+
+        // When — fold the per-zone SUM rows the reader surfaces from the zone-map table
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll());
+             var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all())) {
+            List<ArrayStats> zones = iter.columnZoneStats("id");
+
+            // Then — every zone carries a SUM (came from Rust's table, not a Java-side recompute)
+            // and the whole-zone fold equals the column total.
+            assertThat(zones).isNotEmpty().allSatisfy(z -> assertThat(z.sum()).isNotNull());
+            long total = zones.stream().mapToLong(z -> (Long) z.sum()).sum();
+            assertThat(total).isEqualTo(expected);
+        }
+    }
+
     @Test
     void jniWriter_javaReader_multipleChunks(@TempDir Path tmp) throws IOException {
         // Given

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

@@ -11,19 +11,21 @@
 ///
 /// @param min            minimum value in the array, or `null` if unknown
 /// @param max            maximum value in the array, or `null` if unknown
+/// @param sum            sum of values (`Long` for integer columns, `Double` for floats), or `null` if unknown
 /// @param trueCount      number of `true` values for bool columns, or `null` if unknown
 /// @param nullCount      number of null values, or `null` if unknown
 /// @param isSorted       `true` if the array is sorted in ascending order, or `null` if unknown
 /// @param isStrictSorted `true` if the array is strictly sorted (no duplicates), or `null` if unknown
 public record ArrayStats(
         Object min,
         Object max,
+        Object sum,
         Long trueCount,
         Long nullCount,
         Boolean isSorted,
         Boolean isStrictSorted
 ) {
-    private static final ArrayStats EMPTY = new ArrayStats(null, null, null, null, null, null);
+    private static final ArrayStats EMPTY = new ArrayStats(null, null, null, null, null, null, null);
 
     /// Returns an empty stats instance with all fields set to `null`.
     ///
@@ -43,11 +45,12 @@ public static ArrayStats fromFbs(io.github.dfa1.vortex.core.fbs.FbsArrayStats fb
         }
         Object min = decodeScalar(fbs.minAsSegment());
         Object max = decodeScalar(fbs.maxAsSegment());
+        Object sum = decodeScalar(fbs.sumAsSegment());
         Long nullCount = fbs.hasNullCount() ? fbs.nullCount() : null;
-        if (min == null && max == null && nullCount == null) {
+        if (min == null && max == null && sum == null && nullCount == null) {
             return EMPTY;
         }
-        return new ArrayStats(min, max, null, nullCount, null, null);
+        return new ArrayStats(min, max, sum, null, nullCount, null, null);
     }
 
     private static Object decodeScalar(MemorySegment seg) {

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

@@ -395,6 +395,150 @@ public long[] chunkRowCounts() {
         return out;
     }
 
+    /// Returns the per-zone statistics for one column, one entry per zone in scan order.
+    ///
+    /// When the column carries a `vortex.stats` (zone-map) layout, the rows come from that
+    /// table — min/max/sum/null count per zone — decoded once from the small stats segment
+    /// without touching any data segment. This is the source Rust populates for `SUM`, so the
+    /// values match files written by either implementation. When no zone map is present the
+    /// list falls back to each chunk's embedded `ArrayStats` (min/max/null count; `sum` is
+    /// `null`, since the flat writer does not retain it). Either way the result has one entry
+    /// per chunk/zone, positionally aligned with [#chunkRowCounts()]; a column that is absent
+    /// or carries no stats yields [ArrayStats#empty()] per zone.
+    ///
+    /// This is the read-side surface for aggregate push-down (ADR 0013 §6): a reduction can
+    /// fold whole zones from these rows and fall back to a streaming decode only for the
+    /// boundary zones a predicate partially selects.
+    ///
+    /// Like [#chunkRowCounts()], filter pruning and `ScanOptions.limit()` are not applied —
+    /// the list reflects the raw layout shape.
+    ///
+    /// @param column the column name
+    /// @return per-zone stats in zone order; empty list if the file has no chunks
+    public List<ArrayStats> columnZoneStats(String column) {
+        if (chunks == null) {
+            initialize();
+        }
+        List<ArrayStats> fromTable = decodeZoneTable(column);
+        if (fromTable != null) {
+            return fromTable;
+        }
+        // No zone-map table — surface each chunk's embedded ArrayStats (sum absent).
+        List<ArrayStats> out = new ArrayList<>(chunks.size());
+        for (ChunkSpec spec : chunks) {
+            Layout flat = spec.layoutFor(column);
+            out.add(flat == null ? ArrayStats.empty() : readFlatStats(flat));
+        }
+        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
+    /// short-lived confined arena and the scalar values are boxed out before the arena closes.
+    private List<ArrayStats> decodeZoneTable(String column) {
+        Layout zoned = findZonedLayout(file.layout(), column);
+        if (zoned == null || zoned.children().size() < 2) {
+            return null;
+        }
+        Layout statsFlat = zoned.children().get(1);
+        if (!statsFlat.isFlat() || statsFlat.segments().isEmpty()) {
+            return null;
+        }
+        DType columnDtype = columnDType(column);
+        if (columnDtype == null) {
+            return null;
+        }
+        int segIdx = statsFlat.segments().getFirst();
+        if (segIdx < 0 || segIdx >= file.footer().segmentSpecs().size()) {
+            return null;
+        }
+        DType.Struct statsDtype = ZonedStatsSchema.statsTableDtype(columnDtype, zoned.metadata());
+        long nZones = statsFlat.rowCount();
+        SegmentSpec spec = file.footer().segmentSpecs().get(segIdx);
+        try (Arena tableArena = Arena.ofConfined()) {
+            Array decoded = file.decodeFlatSegment(spec, statsDtype, nZones, tableArena);
+            if (!(decoded instanceof StructArray table)) {
+                return null;
+            }
+            Array minA = fieldOrNull(table, "min");
+            Array maxA = fieldOrNull(table, "max");
+            Array sumA = fieldOrNull(table, "sum");
+            Array nullCountA = fieldOrNull(table, "null_count");
+            List<ArrayStats> out = new ArrayList<>((int) nZones);
+            for (long i = 0; i < nZones; i++) {
+                Object nullCount = boxedScalar(nullCountA, i);
+                out.add(new ArrayStats(
+                        boxedScalar(minA, i),
+                        boxedScalar(maxA, i),
+                        boxedScalar(sumA, i),
+                        null,
+                        nullCount == null ? null : ((Number) nullCount).longValue(),
+                        null, null));
+            }
+            return out;
+        }
+    }
+
+    /// Finds the first `vortex.stats` layout in the subtree of `column`'s top-level layout, or
+    /// `null` when the column is not zone-mapped.
+    private Layout findZonedLayout(Layout root, String column) {
+        if (!(file.dtype() instanceof DType.Struct struct) || !root.isStruct()) {
+            return null;
+        }
+        int idx = struct.fieldNames().indexOf(column);
+        if (idx < 0 || idx >= root.children().size()) {
+            return null;
+        }
+        return firstZoned(root.children().get(idx));
+    }
+
+    private static Layout firstZoned(Layout layout) {
+        if (layout.isZoned()) {
+            return layout;
+        }
+        for (Layout child : layout.children()) {
+            Layout found = firstZoned(child);
+            if (found != null) {
+                return found;
+            }
+        }
+        return null;
+    }
+
+    private static Array fieldOrNull(StructArray table, String field) {
+        if (((DType.Struct) table.dtype()).fieldNames().contains(field)) {
+            return table.field(field);
+        }
+        return null;
+    }
+
+    /// Reads the boxed scalar at index `i` from a (possibly nullable) stats column, or `null`
+    /// when the array is absent or the position is invalid.
+    private static Object boxedScalar(Array array, long i) {
+        if (array == null) {
+            return null;
+        }
+        if (array instanceof MaskedArray masked) {
+            if (!masked.isValid(i)) {
+                return null;
+            }
+            return boxedScalar(masked.inner(), i);
+        }
+        return switch (array) {
+            case LongArray a -> a.getLong(i);
+            case IntArray a -> a.getInt(i);
+            case DoubleArray a -> a.getDouble(i);
+            case FloatArray a -> a.getFloat(i);
+            case ShortArray a -> a.getShort(i);
+            case ByteArray a -> a.getByte(i);
+            case BoolArray a -> a.getBoolean(i);
+            case VarBinArray a -> a.getString(i);
+            default -> null;
+        };
+    }
+
     /// Runs `action` on each remaining chunk inside a try-with-resources
     /// block so every chunk's [Arena] is released before the next iteration.
     /// Prefer this over a manual `while (hasNext()) { next(); `} loop

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

@@ -180,8 +180,8 @@ public Map<String, ArrayStats> columnStats() {
     private ArrayStats aggregateStats(List<Layout> flats) {
         Object globalMin = null;
         Object globalMax = null;
-        // Sum is meaningful only when every chunk carries a null count; one missing makes the
-        // column total unknown (null), so don't report a partial sum.
+        // Null count is meaningful only when every chunk carries it; one missing makes the column
+        // total unknown (null), so don't report a partial count.
         long totalNullCount = 0L;
         boolean allHaveNullCount = !flats.isEmpty();
         for (Layout flat : flats) {
@@ -202,7 +202,9 @@ private ArrayStats aggregateStats(List<Layout> flats) {
         if (globalMin == null && globalMax == null && nullCount == null) {
             return ArrayStats.empty();
         }
-        return new ArrayStats(globalMin, globalMax, null, nullCount, null, null);
+        // Sum is left null at the file level: it lives in the per-zone stats table, surfaced by
+        // ScanIterator.columnZoneStats rather than folded here.
+        return new ArrayStats(globalMin, globalMax, null, null, nullCount, null, null);
     }
 
     private ArrayStats readFlatStats(Layout flat) {

…fa1/vortex/inspect/ZonedStatsSchema.java …dfa1/vortex/reader/ZonedStatsSchema.javainspector/src/main/java/io/github/dfa1/vortex/inspect/ZonedStatsSchema.java renamed to reader/src/main/java/io/github/dfa1/vortex/reader/ZonedStatsSchema.java inspector/src/main/java/io/github/dfa1/vortex/inspect/ZonedStatsSchema.java renamed to reader/src/main/java/io/github/dfa1/vortex/reader/ZonedStatsSchema.java

@@ -1,4 +1,4 @@
-package io.github.dfa1.vortex.inspect;
+package io.github.dfa1.vortex.reader;
 
 import io.github.dfa1.vortex.core.model.DType;
 
@@ -14,15 +14,15 @@
 ///
 /// The shape is sourced from the Rust reference implementation:
 /// - Metadata format
-///       (<a href="https://github.com/spiraldb/vortex/blob/develop/vortex-layout/src/layouts/zoned/mod.rs">
-///       vortex-layout/src/layouts/zoned/mod.rs</a> — `ZonedMetadata`):
+///       ([vortex-layout/src/layouts/zoned/mod.rs](https://github.com/spiraldb/vortex/blob/develop/vortex-layout/src/layouts/zoned/mod.rs)
+///       — `ZonedMetadata`):
 ///       bytes [0..4) are the zone length as a little-endian `u32`;
 ///       remaining bytes form a `Stat` bitset (LSB-first per byte). Each
 ///       set bit at index `i` indicates that the [Stat] with that
 ///       ordinal is present in the auxiliary stats table.
 /// - Schema construction
-///       (<a href="https://github.com/spiraldb/vortex/blob/develop/vortex-layout/src/layouts/zoned/schema.rs">
-///       vortex-layout/src/layouts/zoned/schema.rs</a> — `stats_table_dtype`):
+///       ([vortex-layout/src/layouts/zoned/schema.rs](https://github.com/spiraldb/vortex/blob/develop/vortex-layout/src/layouts/zoned/schema.rs)
+///       — `stats_table_dtype`):
 ///       for each present stat in ordinal order, append a struct field with the
 ///       stat's name and the stat's nullable dtype. `Max` and `Min`
 ///       each get an extra trailing field `max_is_truncated` /
@@ -167,8 +167,7 @@ public static DType.Struct statsTableDtype(DType columnDtype, List<Stat> present
     ///
     /// Mirrors Rust's `Stat::dtype(&DType)` plus the aggregate-function
     /// return-type rules (see
-    /// <a href="https://github.com/spiraldb/vortex/blob/develop/vortex-array/src/aggregate_fn/fns">
-    /// vortex-array/src/aggregate_fn/fns</a>):
+    /// [vortex-array/src/aggregate_fn/fns](https://github.com/spiraldb/vortex/blob/develop/vortex-array/src/aggregate_fn/fns)):
     /// - min/max → same dtype as column (except DType.Null → null);
     /// - is_constant/is_sorted/is_strict_sorted → non-nullable Bool;
     /// - null_count → non-nullable U64;

…vortex/inspect/ZonedStatsSchemaTest.java …/vortex/reader/ZonedStatsSchemaTest.javainspector/src/test/java/io/github/dfa1/vortex/inspect/ZonedStatsSchemaTest.java renamed to reader/src/test/java/io/github/dfa1/vortex/reader/ZonedStatsSchemaTest.java inspector/src/test/java/io/github/dfa1/vortex/inspect/ZonedStatsSchemaTest.java renamed to reader/src/test/java/io/github/dfa1/vortex/reader/ZonedStatsSchemaTest.java

@@ -1,4 +1,4 @@
-package io.github.dfa1.vortex.inspect;
+package io.github.dfa1.vortex.reader;
 
 import java.lang.foreign.MemorySegment;