diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7c9465a3c..75c8d344a 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,15 @@ All notable changes to **vortex-java** are documented here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Fixed
+
+- Zone-map `MIN`/`MAX` is now per-zone nullable instead of all-or-nothing: one chunk without stats (e.g. an all-null chunk) used to blank `MIN`/`MAX` for every chunk in the column, silently disabling zone-map pruning for the whole column rather than just that chunk — matching Rust, which always wraps zone-map stats nullable (`vortex-layout/src/layouts/zoned/schema.rs`). ([#378](https://github.com/dfa1/vortex-java/issues/378))
+- `FrameOfReferenceEncodingEncoder` and `DictEncodingEncoder`'s primitive path now surface `MIN`/`MAX` stats instead of hardcoding `null`; combined with cascading compression (`WriteOptions.cascading`/`allowedCascading`), a numeric column whose chunks won the FOR or Dict competition previously lost zone-map pruning for those chunks with no indication. ([#379](https://github.com/dfa1/vortex-java/issues/379))
+- `ScanIterator`'s zone-map pruning check now decodes the column's compact zone-map table (once per scan, cached) instead of fetching each chunk's own full data segment just to read its embedded stats — over HTTP, checking whether a chunk could be pruned used to cost exactly as much as fetching it outright, so pruning saved nothing for the checked column itself (only for other, unfiltered columns in the same row window). ([#380](https://github.com/dfa1/vortex-java/issues/380))
+- `MaskedEncodingEncoder` (nullable columns) now computes `MIN`/`MAX` from only the row-valid elements instead of trusting whichever inner encoder ran over the dense, placeholder-filled values array — a `long[]`/`int[]`/... has no way to represent "no value", so invalid slots carry a placeholder (commonly `0`) that was silently folding into the reported stats; a column with any nulls could report a wrong non-null `MIN`/`MAX` (e.g. `0`) instead of excluding those rows, corrupting both pruning and any consumer trusting the zone-map's actual values. Found while verifying the #378 fix against the real Rust reader on the same data. ([#381](https://github.com/dfa1/vortex-java/pull/381))
+
 ## [0.14.1] — 2026-09-06
 
 ### Fixed
diff --git a/core/src/main/java/io/github/dfa1/vortex/core/compute/PrimitiveArrays.java b/core/src/main/java/io/github/dfa1/vortex/core/compute/PrimitiveArrays.java
index 1091443ed..2bbb0a77b 100644
--- a/core/src/main/java/io/github/dfa1/vortex/core/compute/PrimitiveArrays.java
+++ b/core/src/main/java/io/github/dfa1/vortex/core/compute/PrimitiveArrays.java
@@ -15,8 +15,9 @@
 ///
 /// [#toLongs(Object, PType, EncodingId)] and [#fromLongs(long[], PType, SegmentAllocator)] are
 /// inverses: the first widens any 8–64 bit integer array to a `long[]`, the second writes a
-/// `long[]` back to a little-endian off-heap segment of the target width. Floating-point ptypes
-/// are not handled here — they reinterpret to raw bits or take type-specific encode paths instead.
+/// `long[]` back to a little-endian off-heap segment of the target width. Both are integer-only —
+/// floating-point ptypes reinterpret to raw bits or take type-specific encode paths instead.
+/// [#compact(PType, Object, boolean[])] covers every primitive ptype, integer and floating alike.
 public final class PrimitiveArrays {
 
     private PrimitiveArrays() {
@@ -109,4 +110,97 @@ public static MemorySegment fromLongs(long[] longs, PType ptype, SegmentAllocato
         }
         return seg;
     }
+
+    /// Copies only the elements at `true` positions in `mask` from `data`, preserving `ptype`'s
+    /// storage array shape (`byte[]` for I8/U8, `short[]` for I16/U16/F16, `int[]` for I32/U32,
+    /// `long[]` for I64/U64, `float[]` for F32, `double[]` for F64). Covers every primitive ptype,
+    /// unlike [#toLongs(Object, PType, EncodingId)]/[#fromLongs(long[], PType, SegmentAllocator)],
+    /// which are integer-only.
+    ///
+    /// Used to strip a nullable column's dense, placeholder-filled values down to its real
+    /// (row-valid) values before an operation that must not see the placeholder — a `byte[]`/
+    /// `int[]`/... has no way to represent "no value", so a nullable column's invalid slots carry
+    /// some placeholder chosen by the caller, commonly `0`; running e.g. a min/max scan over the
+    /// raw array would fold that placeholder in as if it were real data.
+    ///
+    /// @param ptype the logical primitive type of `data`
+    /// @param data  the value array; its runtime type must match `ptype`
+    /// @param mask  per-element validity, aligned with `data` (`true` = keep)
+    /// @return a new array of the same runtime type as `data`, holding only the `true`-masked elements
+    public static Object compact(PType ptype, Object data, boolean[] mask) {
+        int n = 0;
+        for (boolean v : mask) {
+            if (v) {
+                n++;
+            }
+        }
+        return switch (ptype) {
+            case I8, U8 -> {
+                byte[] src = (byte[]) data;
+                byte[] out = new byte[n];
+                int j = 0;
+                for (int i = 0; i < src.length; i++) {
+                    if (mask[i]) {
+                        out[j++] = src[i];
+                    }
+                }
+                yield out;
+            }
+            case I16, U16, F16 -> {
+                short[] src = (short[]) data;
+                short[] out = new short[n];
+                int j = 0;
+                for (int i = 0; i < src.length; i++) {
+                    if (mask[i]) {
+                        out[j++] = src[i];
+                    }
+                }
+                yield out;
+            }
+            case I32, U32 -> {
+                int[] src = (int[]) data;
+                int[] out = new int[n];
+                int j = 0;
+                for (int i = 0; i < src.length; i++) {
+                    if (mask[i]) {
+                        out[j++] = src[i];
+                    }
+                }
+                yield out;
+            }
+            case I64, U64 -> {
+                long[] src = (long[]) data;
+                long[] out = new long[n];
+                int j = 0;
+                for (int i = 0; i < src.length; i++) {
+                    if (mask[i]) {
+                        out[j++] = src[i];
+                    }
+                }
+                yield out;
+            }
+            case F32 -> {
+                float[] src = (float[]) data;
+                float[] out = new float[n];
+                int j = 0;
+                for (int i = 0; i < src.length; i++) {
+                    if (mask[i]) {
+                        out[j++] = src[i];
+                    }
+                }
+                yield out;
+            }
+            case F64 -> {
+                double[] src = (double[]) data;
+                double[] out = new double[n];
+                int j = 0;
+                for (int i = 0; i < src.length; i++) {
+                    if (mask[i]) {
+                        out[j++] = src[i];
+                    }
+                }
+                yield out;
+            }
+        };
+    }
 }
diff --git a/core/src/test/java/io/github/dfa1/vortex/core/compute/PrimitiveArraysTest.java b/core/src/test/java/io/github/dfa1/vortex/core/compute/PrimitiveArraysTest.java
index aef3128e1..b18541229 100644
--- a/core/src/test/java/io/github/dfa1/vortex/core/compute/PrimitiveArraysTest.java
+++ b/core/src/test/java/io/github/dfa1/vortex/core/compute/PrimitiveArraysTest.java
@@ -170,4 +170,131 @@ private static long readElement(MemorySegment seg, PType ptype, int i) {
             default -> throw new IllegalArgumentException("not an integer ptype: " + ptype);
         };
     }
+
+    @Test
+    void compact_i8_keepsOnlyValidElements() {
+        // Given a leading and a trailing invalid slot around real values
+        byte[] data = {9, 10, 20, 9};
+        boolean[] mask = {false, true, true, false};
+
+        // When
+        byte[] result = (byte[]) PrimitiveArrays.compact(PType.I8, data, mask);
+
+        // Then
+        assertThat(result).containsExactly((byte) 10, (byte) 20);
+    }
+
+    @Test
+    void compact_i16_keepsOnlyValidElements() {
+        // Given
+        short[] data = {9, 10, 20, 9};
+        boolean[] mask = {false, true, true, false};
+
+        // When
+        short[] result = (short[]) PrimitiveArrays.compact(PType.I16, data, mask);
+
+        // Then
+        assertThat(result).containsExactly((short) 10, (short) 20);
+    }
+
+    @Test
+    void compact_f16_keepsOnlyValidElements() {
+        // Given — F16 shares I16/U16's short[] storage shape
+        short[] data = {9, 10, 20, 9};
+        boolean[] mask = {false, true, true, false};
+
+        // When
+        short[] result = (short[]) PrimitiveArrays.compact(PType.F16, data, mask);
+
+        // Then
+        assertThat(result).containsExactly((short) 10, (short) 20);
+    }
+
+    @Test
+    void compact_i32_keepsOnlyValidElements() {
+        // Given
+        int[] data = {9, 10, 20, 9};
+        boolean[] mask = {false, true, true, false};
+
+        // When
+        int[] result = (int[]) PrimitiveArrays.compact(PType.I32, data, mask);
+
+        // Then
+        assertThat(result).containsExactly(10, 20);
+    }
+
+    @Test
+    void compact_i64_keepsOnlyValidElements() {
+        // Given
+        long[] data = {9L, 10L, 20L, 9L};
+        boolean[] mask = {false, true, true, false};
+
+        // When
+        long[] result = (long[]) PrimitiveArrays.compact(PType.I64, data, mask);
+
+        // Then
+        assertThat(result).containsExactly(10L, 20L);
+    }
+
+    @Test
+    void compact_f32_keepsOnlyValidElements() {
+        // Given
+        float[] data = {9f, 10f, 20f, 9f};
+        boolean[] mask = {false, true, true, false};
+
+        // When
+        float[] result = (float[]) PrimitiveArrays.compact(PType.F32, data, mask);
+
+        // Then
+        assertThat(result).containsExactly(10f, 20f);
+    }
+
+    @Test
+    void compact_f64_keepsOnlyValidElements() {
+        // Given
+        double[] data = {9.0, 10.0, 20.0, 9.0};
+        boolean[] mask = {false, true, true, false};
+
+        // When
+        double[] result = (double[]) PrimitiveArrays.compact(PType.F64, data, mask);
+
+        // Then
+        assertThat(result).containsExactly(10.0, 20.0);
+    }
+
+    @Test
+    void compact_allValid_returnsEveryElement() {
+        // Given no placeholder slots at all
+        long[] data = {1L, 2L, 3L};
+        boolean[] mask = {true, true, true};
+
+        // When
+        long[] result = (long[]) PrimitiveArrays.compact(PType.I64, data, mask);
+
+        // Then
+        assertThat(result).containsExactly(1L, 2L, 3L);
+    }
+
+    @Test
+    void compact_allInvalid_returnsEmptyArray() {
+        // Given — no real value exists anywhere, e.g. an all-null zone/chunk
+        long[] data = {0L, 0L, 0L};
+        boolean[] mask = {false, false, false};
+
+        // When
+        long[] result = (long[]) PrimitiveArrays.compact(PType.I64, data, mask);
+
+        // Then
+        assertThat(result).isEmpty();
+    }
+
+    @Test
+    void compact_empty_returnsEmptyArray() {
+        // Given no elements at all
+        // When
+        long[] result = (long[]) PrimitiveArrays.compact(PType.I64, new long[0], new boolean[0]);
+
+        // Then
+        assertThat(result).isEmpty();
+    }
 }
diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java
index 733a623a1..ceceb7dae 100644
--- a/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java
+++ b/integration/src/test/java/io/github/dfa1/vortex/integration/JavaWritesRustReadsIntegrationTest.java
@@ -536,6 +536,51 @@ void javaWriter_jniReader_zoneMapped_multipleZones(@TempDir Path tmp) throws IOE
         assertThat(decodedVals).containsExactly(vals);
     }
 
+    @Test
+    void javaWriter_jniReader_zoneMapped_allNullChunkStillRoundTrips(@TempDir Path tmp) throws IOException {
+        // Given — #378 regression: a nullable I64 column across 3 zone-mapped chunks where the
+        // middle chunk is entirely null (no chunk-level min/max to record). Before the fix, one
+        // stats-less chunk anywhere in the column dropped MIN/MAX from the zone-map for every
+        // chunk, not just the offending one; Rust's own zoned schema always wraps MIN/MAX nullable
+        // per zone (vortex-layout/src/layouts/zoned/schema.rs), so an all-or-nothing Java writer was
+        // silently out of step with the format it claims to speak, even though every file it wrote
+        // still parsed. This pins that the fixed writer round-trips through the real Rust reader:
+        // it isn't enough that vortex-java's own reader tolerates the shape it now writes.
+        Path file = tmp.resolve("java_zoned_null_chunk.vtx");
+        DType.Struct schema = new DType.Struct(
+                List.of(ColumnName.of("v")), List.of(new DType.Primitive(PType.I64, true)), false);
+        WriteOptions zoneMapped = new WriteOptions(4, true, 0.90, 0, false, false, MemorySize.ofMiB(256), Map.of());
+        Long[] data = {
+                0L, 1L, 2L, 3L,
+                null, null, null, null,
+                100L, 101L, 102L, 103L};
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var sut = VortexWriter.create(ch, schema, zoneMapped)) {
+            for (int start = 0; start < data.length; start += 4) {
+                sut.writeChunk(Map.of(ColumnName.of("v"), Arrays.copyOfRange(data, start, start + 4)));
+            }
+        }
+
+        // Then — Rust parses the zone-map layout and returns every row, nulls included
+        String uri = file.toAbsolutePath().toUri().toString();
+        DataSource ds = DataSource.open(SESSION, uri);
+        Scan scan = ds.scan(ScanOptions.of());
+        var values = new ArrayList<Long>();
+        while (scan.hasNext()) {
+            Partition partition = scan.next();
+            try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) {
+                while (reader.loadNextBatch()) {
+                    VectorSchemaRoot root = reader.getVectorSchemaRoot();
+                    BigIntVector vec = (BigIntVector) root.getVector("v");
+                    for (int i = 0; i < root.getRowCount(); i++) {
+                        values.add(vec.isNull(i) ? null : vec.get(i));
+                    }
+                }
+            }
+        }
+        assertThat(values).containsExactly(data);
+    }
+
     @Test
     void javaWriter_jniReader_i32Column(@TempDir Path tmp) throws IOException {
         // Given
diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java b/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java
index ca5983287..561c94204 100644
--- a/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java
+++ b/reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java
@@ -94,6 +94,14 @@ public final class ScanIterator implements Iterator<Chunk>, AutoCloseable {
     // its own name and dtype from the schema, unlike the synthetic placeholder.
     private boolean singleColumnIsSyntheticWrapper;
     private Map<ColumnName, DType> columnDtypes;
+    // Total physical chunk count per column, captured at #initialize() from the same collectFlats()
+    // walk that built the scan grid. Guards #zoneIndexByOrdinal against a malformed or foreign
+    // "vortex.stats"-tagged file whose zone table doesn't actually carry one zone per chunk.
+    private Map<ColumnName, Integer> columnChunkCounts;
+    // Decoded zone-map table per column, fetched at most once per scan (one segment read instead of
+    // one per pruning check) — see #zoneStatsFor(ColumnName). Absent entries (no zone map, or the
+    // decode failed) are not cached: those paths cost no I/O, so recomputing is cheap.
+    private Map<ColumnName, List<ArrayStats>> zoneStatsCache;
     private int chunkIndex;
     private int peekedChunkIdx = -1;
     private long rowsReturned;
@@ -193,6 +201,7 @@ static List<ChunkSpec> buildChunks(Map<ColumnName, List<Layout>> columnFlats) {
             long windowRows = boundaries[w + 1] - windowStart;
             Layout[] layouts = new Layout[numCols];
             long[] sliceOffsets = new long[numCols];
+            int[] chunkOrdinals = new int[numCols];
             for (int j = 0; j < numCols; j++) {
                 long[] starts = colStarts[j];
                 int c = cursor[j];
@@ -209,8 +218,9 @@ static List<ChunkSpec> buildChunks(Map<ColumnName, List<Layout>> columnFlats) {
                 }
                 layouts[j] = flats.get(c);
                 sliceOffsets[j] = windowStart - starts[c];
+                chunkOrdinals[j] = c;
             }
-            result.add(new ChunkSpec(windowRows, colNames, layouts, sliceOffsets));
+            result.add(new ChunkSpec(windowStart, windowRows, colNames, layouts, sliceOffsets, chunkOrdinals));
         }
         return List.copyOf(result);
     }
@@ -653,6 +663,10 @@ private void initialize() {
         projectedNames = List.copyOf(columnDtypes.keySet());
         projectedDtypes = List.copyOf(columnDtypes.values());
         lastCoveringFlats = new Layout[projectedNames.size()];
+        columnChunkCounts = new HashMap<>();
+        for (Map.Entry<ColumnName, List<Layout>> e : columnFlats.entrySet()) {
+            columnChunkCounts.put(e.getKey(), e.getValue().size());
+        }
         chunks = buildChunks(columnFlats);
     }
 
@@ -846,12 +860,101 @@ private boolean canPruneChunk(ChunkSpec chunk, RowFilter filter) {
                 if (flat == null) {
                     yield false;
                 }
-                ArrayStats stats = readFlatStats(flat);
+                ArrayStats stats = zoneStats(chunk, col, flat.rowCount());
+                if (stats == null) {
+                    stats = readFlatStats(flat);
+                }
                 yield canPrune(predicate, stats, flat.rowCount(), columnDType(col));
             }
         };
     }
 
+    /// Returns `col`'s zone-map stats for the physical chunk covering `chunk`'s window, or `null`
+    /// when the column has no usable zone map — in which case the caller falls back to the chunk's
+    /// own embedded stats. Reading here decodes (and caches) one small zone-map segment per column
+    /// for the whole scan, instead of [#readFlatStats(Layout)]'s per-chunk read of the chunk's full
+    /// data segment — the difference that makes pruning over HTTP actually cheaper than not pruning.
+    ///
+    /// The two `vortex.zoned`-family layouts need different lookup strategies, dispatched by
+    /// [Layout#layoutId()]: this writer's legacy `vortex.stats` always emits exactly one zone per
+    /// physical chunk, in chunk order ([#zoneIndexByOrdinal]), regardless of each chunk's row count
+    /// (`options.chunkSize()` bounds a chunk's maximum size, not its actual size — a batch smaller
+    /// than the cap becomes its own, smaller chunk — so its declared zone-length metadata does not
+    /// describe a uniform stride and cannot drive arithmetic). The newer Rust `vortex.zoned` instead
+    /// declares a genuinely independent, uniform zone length ([#zoneIndexByLength]) with no fixed
+    /// relationship to chunk boundaries at all; a chunk's row range must fit entirely inside a single
+    /// zone for that zone's stats to safely describe it — never a partial overlap, or rows the chunk
+    /// covers would be invisible to the stats used to judge it.
+    private ArrayStats zoneStats(ChunkSpec chunk, ColumnName col, long chunkRowCount) {
+        Layout zoned = findZonedLayout(file.layout(), col);
+        if (zoned == null) {
+            return null;
+        }
+        Integer zoneIdx = zoned.layoutId() == LayoutId.ZONED
+                ? zoneIndexByLength(chunk, col, chunkRowCount, zoned.metadata())
+                : zoneIndexByOrdinal(chunk, col);
+        if (zoneIdx == null) {
+            return null;
+        }
+        List<ArrayStats> zones = zoneStatsFor(col);
+        return zones == null || zoneIdx >= zones.size() ? null : zones.get(zoneIdx);
+    }
+
+    /// Locates the zone covering a chunk under this writer's legacy `vortex.stats` layout, where
+    /// zone order always matches physical chunk order 1:1 (`VortexWriter#flushZoneMaps`) — so the
+    /// chunk's ordinal position among `col`'s own chunks IS the zone index. Guarded by a chunk-count
+    /// match against the decoded table, in case a malformed or foreign `vortex.stats`-tagged file
+    /// does not actually follow that convention.
+    private Integer zoneIndexByOrdinal(ChunkSpec chunk, ColumnName col) {
+        Integer chunkCount = columnChunkCounts.get(col);
+        if (chunkCount == null) {
+            return null;
+        }
+        List<ArrayStats> zones = zoneStatsFor(col);
+        if (zones == null || zones.size() != chunkCount) {
+            return null;
+        }
+        int ordinal = chunk.chunkOrdinalFor(col);
+        return ordinal < 0 ? null : ordinal;
+    }
+
+    /// Locates the zone covering a chunk under Rust's `vortex.zoned` layout, whose declared zone
+    /// length is a genuine uniform stride independent of chunk boundaries. Returns `null` unless the
+    /// chunk's whole row range — `[chunkStart, chunkStart + chunkRowCount)`, `chunkStart` recovered
+    /// from `chunk.windowStart() - chunk.sliceOffsetFor(col)` — fits inside a single zone's range.
+    private static Integer zoneIndexByLength(ChunkSpec chunk, ColumnName col, long chunkRowCount, MemorySegment metadata) {
+        long zoneLen = ZonedStatsSchema.aggregateZoneLength(metadata);
+        if (zoneLen <= 0) {
+            return null;
+        }
+        long sliceOffset = chunk.sliceOffsetFor(col);
+        if (sliceOffset < 0) {
+            return null;
+        }
+        long chunkStart = chunk.windowStart() - sliceOffset;
+        long zoneIdx = chunkStart / zoneLen;
+        long zoneEnd = (zoneIdx + 1) * zoneLen;
+        if (chunkStart + chunkRowCount > zoneEnd || zoneIdx > Integer.MAX_VALUE) {
+            return null;
+        }
+        return (int) zoneIdx;
+    }
+
+    private List<ArrayStats> zoneStatsFor(ColumnName col) {
+        if (zoneStatsCache == null) {
+            zoneStatsCache = new HashMap<>();
+        }
+        List<ArrayStats> cached = zoneStatsCache.get(col);
+        if (cached != null) {
+            return cached;
+        }
+        List<ArrayStats> decoded = decodeZoneTable(col);
+        if (decoded != null) {
+            zoneStatsCache.put(col, decoded);
+        }
+        return decoded;
+    }
+
     /// Tests whether `predicate`, compiled against a chunk's zone-map statistics, can prove that no
     /// row in the chunk can match — in which case the chunk is skipped. Pruning is strictly
     /// conservative: every branch returns `false` (do not prune) when a needed statistic is missing
@@ -967,14 +1070,34 @@ public SegmentSpec segmentSpec(int index) {
 
     @SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared.
     record ChunkSpec(
-            long rowCount, ColumnName[] columnNames, Layout[] columnLayouts, long[] sliceOffsets) {
+            long windowStart, long rowCount, ColumnName[] columnNames, Layout[] columnLayouts,
+            long[] sliceOffsets, int[] chunkOrdinals) {
         Layout layoutFor(ColumnName col) {
+            int i = indexFor(col);
+            return i < 0 ? null : columnLayouts[i];
+        }
+
+        /// The offset of this window into `col`'s covering chunk, or `-1` if `col` is not part of
+        /// this window. `windowStart() - sliceOffsetFor(col)` is that chunk's absolute row start.
+        long sliceOffsetFor(ColumnName col) {
+            int i = indexFor(col);
+            return i < 0 ? -1 : sliceOffsets[i];
+        }
+
+        /// The ordinal position (0-based) of this window's covering chunk within `col`'s own
+        /// physical chunk list, or `-1` if `col` is not part of this window.
+        int chunkOrdinalFor(ColumnName col) {
+            int i = indexFor(col);
+            return i < 0 ? -1 : chunkOrdinals[i];
+        }
+
+        private int indexFor(ColumnName col) {
             for (int i = 0; i < columnNames.length; i++) {
                 if (columnNames[i].equals(col)) {
-                    return columnLayouts[i];
+                    return i;
                 }
             }
-            return null;
+            return -1;
         }
     }
 }
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
index 267d7a0f4..0b88e2d43 100644
--- 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
@@ -93,6 +93,38 @@ public static long zoneLength(MemorySegment metadata) {
         return Integer.toUnsignedLong(metadata.get(LE_INT, 0));
     }
 
+    /// Returns the zone length declared in a newer `vortex.zoned` layout's aggregate-spec metadata
+    /// (field 1, `uint32 zone_len`, of the `ZonedMetadataProto` described in [#aggregateIds]), or
+    /// `0` when the blob is empty, carries an unsupported version, or is not well-formed protobuf.
+    ///
+    /// @param metadata raw `vortex.zoned` layout metadata, possibly `null`
+    /// @return decoded zone length, or `0` when absent
+    public static long aggregateZoneLength(MemorySegment metadata) {
+        if (metadata == null || metadata.byteSize() < 1) {
+            return 0L;
+        }
+        if ((metadata.get(ValueLayout.JAVA_BYTE, 0) & 0xff) != AGGREGATE_METADATA_VERSION) {
+            return 0L;
+        }
+        ProtoCursor cursor = new ProtoCursor(metadata, 1, metadata.byteSize());
+        while (cursor.hasRemaining()) {
+            long tag = cursor.readVarint();
+            if (tag < 0) {
+                return 0L;
+            }
+            int fieldNumber = (int) (tag >>> 3);
+            int wireType = (int) (tag & 0x7);
+            if (fieldNumber == 1 && wireType == ProtoCursor.WIRE_VARINT) {
+                long len = cursor.readVarint();
+                return len < 0 ? 0L : len;
+            }
+            if (!cursor.skipField(wireType)) {
+                return 0L;
+            }
+        }
+        return 0L;
+    }
+
     /// Returns the stats present in the layout metadata bitset, in ordinal order.
     ///
     /// Unknown bits (set at an index past [Stat#values()]'s length, which
diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/ScanIteratorChunkGridTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/ScanIteratorChunkGridTest.java
index 5cd40c957..b5399b2d2 100644
--- a/reader/src/test/java/io/github/dfa1/vortex/reader/ScanIteratorChunkGridTest.java
+++ b/reader/src/test/java/io/github/dfa1/vortex/reader/ScanIteratorChunkGridTest.java
@@ -59,6 +59,51 @@ void singleFlatColumnSharesTheChunkedGrid() {
         assertWholeChunk(result.get(1), B, 4);
     }
 
+    @Test
+    void windowStartMinusSliceOffsetRecoversEachChunksAbsoluteRowStart() {
+        // Given the same 1-vs-N shape as singleFlatColumnSharesTheChunkedGrid: one full-column flat
+        // [8] (A, a single physical chunk starting at row 0) beside a chunked column [4, 4] (B, a
+        // second physical chunk starting at row 4).
+        var columnFlats = flats(A, new long[]{8}, B, new long[]{4, 4});
+
+        // When
+        List<ChunkSpec> result = ScanIterator.buildChunks(columnFlats);
+
+        // Then `windowStart() - sliceOffsetFor(col)` recovers each column's covering chunk's
+        // absolute row start: A's single chunk starts at row 0 in both windows (its one physical
+        // chunk spans both, so the derived start must not drift with the window), while B's second
+        // window is covered by its second chunk, starting at row 4. This is the invariant zone-map
+        // pruning's fast path (#380) relies on to locate which zone actually covers a given chunk.
+        assertThat(chunkStart(result.get(0), A)).isZero();
+        assertThat(chunkStart(result.get(0), B)).isZero();
+        assertThat(chunkStart(result.get(1), A)).isZero();
+        assertThat(chunkStart(result.get(1), B)).isEqualTo(4L);
+    }
+
+    private static long chunkStart(ChunkSpec spec, ColumnName column) {
+        return spec.windowStart() - spec.sliceOffsetFor(column);
+    }
+
+    @Test
+    void chunkOrdinalsTrackPositionWithinEachColumnsOwnChunkList() {
+        // Given the same 1-vs-N shape: one full-column flat [8] (A, a single physical chunk) beside
+        // a chunked column [4, 4] (B, two physical chunks).
+        var columnFlats = flats(A, new long[]{8}, B, new long[]{4, 4});
+
+        // When
+        List<ChunkSpec> result = ScanIterator.buildChunks(columnFlats);
+
+        // Then A's ordinal stays 0 in both windows — its one physical chunk spans both, so the
+        // ordinal must not increment per window — while B's ordinal advances to its second chunk.
+        // This is the invariant this writer's own zone-map pruning fast path (#380) relies on: for
+        // the legacy vortex.stats layout, zone order always matches physical chunk order 1:1, so a
+        // window's chunk ordinal directly indexes the decoded zone table.
+        assertThat(result.get(0).chunkOrdinalFor(A)).isZero();
+        assertThat(result.get(0).chunkOrdinalFor(B)).isZero();
+        assertThat(result.get(1).chunkOrdinalFor(A)).isZero();
+        assertThat(result.get(1).chunkOrdinalFor(B)).isEqualTo(1);
+    }
+
     @Test
     void nestedBoundariesSliceTheCoarseColumnAtTheFineGrid() {
         // Given the emotions-dataset-for-nlp shape scaled down: a coarse column [8, 8, 8, 4] (like
diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java b/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java
index 6aa6db488..e5c0681b9 100644
--- a/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java
+++ b/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java
@@ -833,13 +833,12 @@ private void flushZoneMaps() throws IOException {
             }
             DType colDtype = columnDtype(colName);
             DType minMaxDtype = ZoneMapStatCodec.zoneMinMaxDtype(colDtype);
-            boolean hasMinMax = minMaxDtype != null && chunks.stream().allMatch(ChunkRef::hasStats);
             DType sumDtype = ZoneMapStatCodec.zoneSumDtype(colDtype);
             long[] nullCounts = new long[chunks.size()];
             for (int i = 0; i < chunks.size(); i++) {
                 nullCounts[i] = chunks.get(i).nullCount();
             }
-            emitZoneMap(colName, hasMinMax ? minMaxDtype : null,
+            emitZoneMap(colName, minMaxDtype,
                     chunks.stream().map(ChunkRef::statsMin).toList(),
                     chunks.stream().map(ChunkRef::statsMax).toList(),
                     sumDtype, chunks.stream().map(ChunkRef::statsSum).toList(),
@@ -852,11 +851,8 @@ private void flushZoneMaps() throws IOException {
             DictColRef ref = e.getValue();
             DType colDtype = columnDtype(e.getKey());
             DType minMaxDtype = ZoneMapStatCodec.zoneMinMaxDtype(colDtype);
-            boolean hasMinMax = minMaxDtype != null
-                    && ref.chunkStatsMin().stream().allMatch(java.util.Objects::nonNull)
-                    && ref.chunkStatsMax().stream().allMatch(java.util.Objects::nonNull);
             long[] nullCounts = ref.chunkNullCounts().stream().mapToLong(Long::longValue).toArray();
-            emitZoneMap(e.getKey(), hasMinMax ? minMaxDtype : null,
+            emitZoneMap(e.getKey(), minMaxDtype,
                     ref.chunkStatsMin(), ref.chunkStatsMax(),
                     ZoneMapStatCodec.zoneSumDtype(colDtype), ref.chunkStatsSum(), nullCounts);
         }
@@ -869,8 +865,10 @@ private DType columnDtype(ColumnName colName) {
     /// Writes one `vortex.stats` zone-map for `colName`: one zone per chunk, with NULL_COUNT always,
     /// MAX/MIN (plus always-false `_is_truncated` flags) when `minMaxDtype` is non-null, and SUM when
     /// `sumDtype` is non-null. `minBytes`/`maxBytes`/`sumBytes` hold each zone's serialized scalar —
-    /// read only when the matching dtype is set; a `null` `sumBytes` entry marks an overflowed zone
-    /// (recorded as a null sum). Field/bit order follows ZonedStatsSchema: MAX(3), MIN(4), SUM(5),
+    /// read only when the matching dtype is set; a `null` entry marks that specific zone's stat as
+    /// unknown (e.g. an all-null chunk, an overflowed sum, or an encoder that does not surface a
+    /// min/max) rather than dropping the stat for the whole column — MIN/MAX/SUM are nullable per
+    /// zone, matching Rust. Field/bit order follows ZonedStatsSchema: MAX(3), MIN(4), SUM(5),
     /// NULL_COUNT(6).
     private void emitZoneMap(ColumnName colName, DType minMaxDtype, List<byte[]> minBytes, List<byte[]> maxBytes,
                              DType sumDtype, List<byte[]> sumBytes, long[] nullCounts) throws IOException {
@@ -883,15 +881,19 @@ private void emitZoneMap(ColumnName colName, DType minMaxDtype, List<byte[]> min
         List<Object> fields = new java.util.ArrayList<>();
         if (minMaxDtype != null) {
             boolean[] notTruncated = new boolean[nZones];
+            boolean[] maxValid = new boolean[nZones];
+            Object maxValues = ZoneMapStatCodec.zoneStatValues(minMaxDtype, maxBytes, maxValid);
             names.add("max");
             types.add(minMaxDtype);
-            fields.add(new NullableData(ZoneMapStatCodec.zoneStatValues(minMaxDtype, maxBytes), allValid.clone()));
+            fields.add(new NullableData(maxValues, maxValid));
             names.add("max_is_truncated");
             types.add(DType.BOOL);
             fields.add(notTruncated);
+            boolean[] minValid = new boolean[nZones];
+            Object minValues = ZoneMapStatCodec.zoneStatValues(minMaxDtype, minBytes, minValid);
             names.add("min");
             types.add(minMaxDtype);
-            fields.add(new NullableData(ZoneMapStatCodec.zoneStatValues(minMaxDtype, minBytes), allValid.clone()));
+            fields.add(new NullableData(minValues, minValid));
             names.add("min_is_truncated");
             types.add(DType.BOOL);
             fields.add(notTruncated.clone());
@@ -1194,9 +1196,6 @@ private record SegRef(long offset, long len) {
     @SuppressWarnings("java:S6218")
     private record ChunkRef(int segIdx, long rowCount, byte[] statsMin, byte[] statsMax,
             byte[] statsSum, long nullCount) {
-        boolean hasStats() {
-            return statsMin != null && statsMax != null;
-        }
     }
 
     /// Per-column zone-map: the flat segment holding the per-zone stats table, the zone
diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/ZoneMapStatCodec.java b/writer/src/main/java/io/github/dfa1/vortex/writer/ZoneMapStatCodec.java
index 35abe3a4d..898a6a4de 100644
--- a/writer/src/main/java/io/github/dfa1/vortex/writer/ZoneMapStatCodec.java
+++ b/writer/src/main/java/io/github/dfa1/vortex/writer/ZoneMapStatCodec.java
@@ -85,11 +85,15 @@ static byte[] columnSum(DType dtype, Object data) {
     }
 
     /// Builds the per-zone min (or max) values array for the resolved min/max `dtype`, decoding each
-    /// zone's serialized [ProtoScalarValue] stat into the array shape its encoder expects.
-    static Object zoneStatValues(DType minMaxDtype, List<byte[]> statBytes) throws IOException {
+    /// zone's serialized [ProtoScalarValue] stat into the array shape its encoder expects. A `null`
+    /// entry in `statBytes` (the chunk's encoder did not surface a min/max, e.g. an all-null chunk,
+    /// or an encoder that does not compute one) marks that zone invalid in `valid` rather than
+    /// failing the whole column's zone-map — matching Rust, which stores min/max as a nullable
+    /// per-zone stat.
+    static Object zoneStatValues(DType minMaxDtype, List<byte[]> statBytes, boolean[] valid) throws IOException {
         return switch (minMaxDtype) {
-            case DType.Primitive p -> statColumn(p.ptype(), statBytes);
-            case DType.Utf8 _ -> statStringColumn(statBytes);
+            case DType.Primitive p -> statColumn(p.ptype(), statBytes, valid);
+            case DType.Utf8 _ -> statStringColumn(statBytes, valid);
             default -> throw new IllegalStateException("no zone stat values for " + minMaxDtype);
         };
     }
@@ -118,58 +122,69 @@ static Object sumColumn(DType sumDtype, List<byte[]> sumBytes, boolean[] valid)
 
     /// Builds the per-zone string array by decoding each zone's serialized string [ProtoScalarValue]
     /// stat. Used for Utf8 columns whose `vortex.varbin` encoder records full string min/max scalars.
-    private static String[] statStringColumn(List<byte[]> statBytes) throws IOException {
+    /// A `null` entry in `statBytes` sets `valid[i]` to `false` and fills the slot with `""` (a
+    /// sum-neutral-style placeholder never read back, since [NullableData] carries validity
+    /// separately from values).
+    private static String[] statStringColumn(List<byte[]> statBytes, boolean[] valid) throws IOException {
         String[] out = new String[statBytes.size()];
         for (int i = 0; i < out.length; i++) {
-            out[i] = decodeScalar(statBytes.get(i)).string_value();
+            valid[i] = statBytes.get(i) != null;
+            out[i] = valid[i] ? decodeScalar(statBytes.get(i)).string_value() : "";
         }
         return out;
     }
 
     /// Builds the per-zone values array in the storage shape the primitive encoder expects, decoding
-    /// each zone's serialized [ProtoScalarValue] stat.
-    private static Object statColumn(PType ptype, List<byte[]> statBytes) throws IOException {
+    /// each zone's serialized [ProtoScalarValue] stat. A `null` entry in `statBytes` sets `valid[i]`
+    /// to `false` and fills the slot with a zero placeholder, never read back.
+    private static Object statColumn(PType ptype, List<byte[]> statBytes, boolean[] valid) throws IOException {
         int n = statBytes.size();
         return switch (ptype) {
             case I8, U8 -> {
                 byte[] a = new byte[n];
                 for (int i = 0; i < n; i++) {
-                    a[i] = (byte) scalarLong(statBytes.get(i));
+                    valid[i] = statBytes.get(i) != null;
+                    a[i] = valid[i] ? (byte) scalarLong(statBytes.get(i)) : 0;
                 }
                 yield a;
             }
             case I16, U16 -> {
                 short[] a = new short[n];
                 for (int i = 0; i < n; i++) {
-                    a[i] = (short) scalarLong(statBytes.get(i));
+                    valid[i] = statBytes.get(i) != null;
+                    a[i] = valid[i] ? (short) scalarLong(statBytes.get(i)) : 0;
                 }
                 yield a;
             }
             case I32, U32 -> {
                 int[] a = new int[n];
                 for (int i = 0; i < n; i++) {
-                    a[i] = (int) scalarLong(statBytes.get(i));
+                    valid[i] = statBytes.get(i) != null;
+                    a[i] = valid[i] ? (int) scalarLong(statBytes.get(i)) : 0;
                 }
                 yield a;
             }
             case I64, U64 -> {
                 long[] a = new long[n];
                 for (int i = 0; i < n; i++) {
-                    a[i] = scalarLong(statBytes.get(i));
+                    valid[i] = statBytes.get(i) != null;
+                    a[i] = valid[i] ? scalarLong(statBytes.get(i)) : 0L;
                 }
                 yield a;
             }
             case F32 -> {
                 float[] a = new float[n];
                 for (int i = 0; i < n; i++) {
-                    a[i] = (float) scalarDouble(statBytes.get(i));
+                    valid[i] = statBytes.get(i) != null;
+                    a[i] = valid[i] ? (float) scalarDouble(statBytes.get(i)) : 0f;
                 }
                 yield a;
             }
             case F64 -> {
                 double[] a = new double[n];
                 for (int i = 0; i < n; i++) {
-                    a[i] = scalarDouble(statBytes.get(i));
+                    valid[i] = statBytes.get(i) != null;
+                    a[i] = valid[i] ? scalarDouble(statBytes.get(i)) : 0.0;
                 }
                 yield a;
             }
@@ -177,7 +192,8 @@ private static Object statColumn(PType ptype, List<byte[]> statBytes) throws IOE
                 // F16 min/max are serialized as f32 scalars; re-pack to float16 storage.
                 short[] a = new short[n];
                 for (int i = 0; i < n; i++) {
-                    a[i] = Float.floatToFloat16((float) scalarDouble(statBytes.get(i)));
+                    valid[i] = statBytes.get(i) != null;
+                    a[i] = valid[i] ? Float.floatToFloat16((float) scalarDouble(statBytes.get(i))) : 0;
                 }
                 yield a;
             }
diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoder.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoder.java
index ff52f7be7..e630dd2ef 100644
--- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoder.java
+++ b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoder.java
@@ -78,7 +78,9 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
                 new EncodeNode[]{valuesNode, codesNode},
                 new int[0]);
 
-        return new EncodeResult(rootNode, List.of(d.valuesBuf(), codesBuf), null, null);
+        byte[][] stats = PrimitiveEncodingEncoder.minMaxStats(((DType.Primitive) dtype).ptype(), data);
+        return new EncodeResult(rootNode, List.of(d.valuesBuf(), codesBuf),
+                PrimitiveEncodingEncoder.minOf(stats), PrimitiveEncodingEncoder.maxOf(stats));
     }
 
     @Override
@@ -98,7 +100,9 @@ public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext ctx) {
 
         DType codesDtype = new DType.Primitive(codePType, false);
         ChildSlot slot = new ChildSlot(codesDtype, d.codesArr(), 1);
-        return new CascadeStep(partialRoot, List.of(d.valuesBuf()), List.of(slot), null, null, true);
+        byte[][] stats = PrimitiveEncodingEncoder.minMaxStats(((DType.Primitive) dtype).ptype(), data);
+        return new CascadeStep(partialRoot, List.of(d.valuesBuf()), List.of(slot),
+                PrimitiveEncodingEncoder.minOf(stats), PrimitiveEncodingEncoder.maxOf(stats), true);
     }
 
     /// Cascading Utf8 dict: emit the codes leaf but expose the distinct-values pool as an open Utf8
diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoder.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoder.java
index c4691ae06..e98adfaa0 100644
--- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoder.java
+++ b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoder.java
@@ -39,7 +39,9 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
 
         EncodeNode child = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 0);
         EncodeNode root = new EncodeNode(EncodingId.FASTLANES_FOR, meta, new EncodeNode[]{child}, new int[0]);
-        return new EncodeResult(root, List.of(residuals), null, null);
+        byte[][] stats = PrimitiveEncodingEncoder.minMaxStats(ptype, data);
+        return new EncodeResult(root, List.of(residuals),
+                PrimitiveEncodingEncoder.minOf(stats), PrimitiveEncodingEncoder.maxOf(stats));
     }
 
     @Override
@@ -62,7 +64,9 @@ public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext encodeC
 
         EncodeNode partialRoot = new EncodeNode(EncodingId.FASTLANES_FOR, meta, new EncodeNode[1], new int[0]);
         ChildSlot slot = new ChildSlot(dtype, residualsAsNativeArray(longs, ref, ptype), 0);
-        return new CascadeStep(partialRoot, List.of(), List.of(slot), null, null, true);
+        byte[][] stats = PrimitiveEncodingEncoder.minMaxStats(ptype, data);
+        return new CascadeStep(partialRoot, List.of(), List.of(slot),
+                PrimitiveEncodingEncoder.minOf(stats), PrimitiveEncodingEncoder.maxOf(stats), true);
     }
 
     private static long computeRef(long[] longs, int n) {
diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoder.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoder.java
index 6b4c8d87f..e6f6b5132 100644
--- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoder.java
+++ b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoder.java
@@ -4,6 +4,7 @@
 import io.github.dfa1.vortex.core.error.VortexException;
 
 import io.github.dfa1.vortex.core.model.EncodingId;
+import io.github.dfa1.vortex.core.compute.PrimitiveArrays;
 
 import java.lang.foreign.MemorySegment;
 import java.util.ArrayList;
@@ -59,7 +60,37 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
                 null,
                 new EncodeNode[]{valuesResult.rootNode(), validityNode},
                 new int[0]);
-        return new EncodeResult(root, buffers, valuesResult.statsMin(), valuesResult.statsMax());
+        byte[][] stats = maskedMinMaxStats(nonNullable, values, validity);
+        byte[] statsMin = stats == null ? null : stats[0];
+        byte[] statsMax = stats == null ? null : stats[1];
+        return new EncodeResult(root, buffers, statsMin, statsMax);
+    }
+
+    /// Computes MIN/MAX over only the row-valid elements of `values`, never `valuesResult`'s own
+    /// stats. A primitive `NullableData` values array is always dense (a `long[]`/`int[]`/... has
+    /// no way to represent "no value"), so its invalid slots carry some placeholder — commonly `0`
+    /// — supplied by the caller; blindly reusing whichever inner encoder's stats (e.g.
+    /// [PrimitiveEncodingEncoder], [FrameOfReferenceEncodingEncoder], [DictEncodingEncoder]) ran over
+    /// that dense array folds the placeholder into MIN/MAX as if it were real data. A `String[]`
+    /// carries a real `null` at invalid positions (validity is redundant there, [#denseValues] only
+    /// substitutes `""` in a separate copy fed to the cascade, never in `values` itself), so
+    /// [VarBinEncodingEncoder#minMaxStats(String[])]'s existing null-skip is already correct and is
+    /// reused as-is. Binary and nested (List/FixedSizeList) values have no zone-map min/max
+    /// ([io.github.dfa1.vortex.writer.ZoneMapStatCodec#zoneMinMaxDtype]), so `null` there is a no-op.
+    ///
+    /// @param nonNullable the values' own dtype (validity stripped)
+    /// @param values      the dense values array from the [NullableData] carrier
+    /// @param validity    per-row validity, aligned with `values`
+    /// @return a two-element `{min, max}` array of encoded scalars, or `null` when no row is valid
+    private static byte[][] maskedMinMaxStats(DType nonNullable, Object values, boolean[] validity) {
+        if (nonNullable instanceof DType.Primitive p) {
+            Object compacted = PrimitiveArrays.compact(p.ptype(), values, validity);
+            return PrimitiveEncodingEncoder.minMaxStats(p.ptype(), compacted);
+        }
+        if (values instanceof String[] strings) {
+            return VarBinEncodingEncoder.minMaxStats(strings);
+        }
+        return null;
     }
 
     /// Encodes the non-null values of a masked column.
diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/PrimitiveEncodingEncoder.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/PrimitiveEncodingEncoder.java
index eadc6e260..30abf552a 100644
--- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/PrimitiveEncodingEncoder.java
+++ b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/PrimitiveEncodingEncoder.java
@@ -286,6 +286,24 @@ public static byte[][] minMaxStats(PType ptype, Object data) {
         };
     }
 
+    /// The min half of a [#minMaxStats] result, or `null` when `stats` itself is `null` (empty
+    /// data). Saves callers the repeated `stats == null ? null : stats[0]` idiom.
+    ///
+    /// @param stats a [#minMaxStats] result, possibly `null`
+    /// @return the serialized min scalar, or `null`
+    public static byte[] minOf(byte[][] stats) {
+        return stats == null ? null : stats[0];
+    }
+
+    /// The max half of a [#minMaxStats] result, or `null` when `stats` itself is `null` (empty
+    /// data). Saves callers the repeated `stats == null ? null : stats[1]` idiom.
+    ///
+    /// @param stats a [#minMaxStats] result, possibly `null`
+    /// @return the serialized max scalar, or `null`
+    public static byte[] maxOf(byte[][] stats) {
+        return stats == null ? null : stats[1];
+    }
+
     /// Computes the serialized SUM [io.github.dfa1.vortex.core.proto.ProtoScalarValue] for a raw primitive
     /// array, in the widened shape Rust uses for zone-map sums: signed ints → `i64`, unsigned ints
     /// → `u64`, floats → `f64`. Returns `null` on integer overflow (Rust drops the zone's sum) and
diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/WriterZoneMapHttpPruningTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterZoneMapHttpPruningTest.java
new file mode 100644
index 000000000..959a8f870
--- /dev/null
+++ b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterZoneMapHttpPruningTest.java
@@ -0,0 +1,150 @@
+package io.github.dfa1.vortex.writer;
+
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.MemorySize;
+import io.github.dfa1.vortex.reader.ReadRegistry;
+import io.github.dfa1.vortex.reader.RowFilter;
+import io.github.dfa1.vortex.reader.ScanOptions;
+import io.github.dfa1.vortex.reader.VortexHttpReader;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpHeaders;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+import javax.net.ssl.SSLSession;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+
+/// Regression coverage for #380: zone-map pruning over HTTP must consult the compact zone-map
+/// table (one small segment for the whole column) instead of fetching each chunk's own full data
+/// segment just to check whether it can be skipped — otherwise "checking" a chunk costs as much
+/// over the network as simply not pruning it at all.
+///
+/// Lives here (not in `reader`) because building the fixture needs [VortexWriter]; `vortex-writer`
+/// already depends on `vortex-reader` in test scope for round-trip assertions like this one.
+@ExtendWith(MockitoExtension.class)
+class WriterZoneMapHttpPruningTest {
+
+    @Mock
+    private HttpClient client;
+
+    private static final URI FILE_URI = URI.create("http://example.com/pruning.vortex");
+
+    @Test
+    void pruningEveryChunk_fetchesTheZoneTableOnce_notOncePerChunk(@TempDir Path tmp) throws Exception {
+        // Given a plain (non-cascading) I64 column across 4 chunks of 2 rows, each chunk in a
+        // disjoint value range: [0,1], [10,11], [20,21], [30,31].
+        DType.Struct schema = new DType.Struct(List.of(ColumnName.of("v")), List.of(DType.I64), false);
+        WriteOptions opts = new WriteOptions(2, true, 0.90, 0, false, false, MemorySize.ofMiB(256), Map.of());
+        Path file = tmp.resolve("pruning.vtx");
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var sut = VortexWriter.create(ch, schema, opts)) {
+            sut.writeChunk(Map.of(ColumnName.of("v"), new long[]{0L, 1L}));
+            sut.writeChunk(Map.of(ColumnName.of("v"), new long[]{10L, 11L}));
+            sut.writeChunk(Map.of(ColumnName.of("v"), new long[]{20L, 21L}));
+            sut.writeChunk(Map.of(ColumnName.of("v"), new long[]{30L, 31L}));
+        }
+        byte[] bytes = Files.readAllBytes(file);
+        AtomicInteger requestCount = new AtomicInteger();
+        given(client.send(any(), any())).willAnswer(inv -> {
+            requestCount.incrementAndGet();
+            HttpRequest req = inv.getArgument(0);
+            return rangeResponse(req, bytes);
+        });
+
+        // When — a filter no chunk can satisfy (every chunk's max is well below 1000). The file is
+        // tiny, so #open() resolves footer/layout/dtype from a single tail fetch; only the scan
+        // below should issue further requests.
+        try (var reader = VortexHttpReader.open(FILE_URI, ReadRegistry.loadAll(), client)) {
+            int afterOpen = requestCount.get();
+            try (var iter = reader.scan(ScanOptions.all().withFilter(RowFilter.gt("v", 1000L)))) {
+                assertThat(iter.hasNext()).isFalse();
+            }
+
+            // Then exactly one additional request — decoding the shared zone-map table once —
+            // accounts for pruning all 4 chunks. Before the #380 fix, checking each chunk fetched
+            // that chunk's own full data segment to read its embedded stats (4 requests), so
+            // pruning cost as much over HTTP as not pruning at all.
+            assertThat(requestCount.get() - afterOpen).isEqualTo(1);
+        }
+    }
+
+    private static HttpResponse<byte[]> rangeResponse(HttpRequest req, byte[] file) {
+        String rangeHeader = req.headers().firstValue("Range").orElseThrow();
+        String spec = rangeHeader.substring("bytes=".length());
+        String[] parts = spec.split("-", 2);
+        int from = parts[0].isEmpty()
+                ? Math.max(0, file.length - Integer.parseInt(parts[1]))
+                : Integer.parseInt(parts[0]);
+        int to = parts[0].isEmpty() ? file.length - 1 : Math.min(Integer.parseInt(parts[1]), file.length - 1);
+        byte[] body = Arrays.copyOfRange(file, from, to + 1);
+        String contentRange = "bytes " + from + "-" + to + "/" + file.length;
+        return response(206, contentRange, body);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static HttpResponse<byte[]> response(int status, String contentRange, byte[] body) {
+        return new HttpResponse<>() {
+            @Override
+            public int statusCode() {
+                return status;
+            }
+
+            @Override
+            public byte[] body() {
+                return body;
+            }
+
+            @Override
+            public HttpHeaders headers() {
+                Map<String, List<String>> map = contentRange == null
+                        ? Map.of()
+                        : Map.of("content-range", List.of(contentRange));
+                return HttpHeaders.of(map, (k, v) -> true);
+            }
+
+            @Override
+            public HttpRequest request() {
+                return null;
+            }
+
+            @Override
+            public Optional<HttpResponse<byte[]>> previousResponse() {
+                return Optional.empty();
+            }
+
+            @Override
+            public Optional<SSLSession> sslSession() {
+                return Optional.empty();
+            }
+
+            @Override
+            public URI uri() {
+                return FILE_URI;
+            }
+
+            @Override
+            public HttpClient.Version version() {
+                return HttpClient.Version.HTTP_1_1;
+            }
+        };
+    }
+}
diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/WriterZoneMapTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterZoneMapTest.java
index e6b3ae1c3..7f6d93093 100644
--- a/writer/src/test/java/io/github/dfa1/vortex/writer/WriterZoneMapTest.java
+++ b/writer/src/test/java/io/github/dfa1/vortex/writer/WriterZoneMapTest.java
@@ -212,10 +212,12 @@ void noChunks_emitsNoZoneMap(@TempDir Path tmp) throws IOException {
     }
 
     @Test
-    void chunkWithoutStats_emitsNullCountOnlyZoneMap(@TempDir Path tmp) throws IOException {
+    void chunkWithoutStats_marksOnlyThatZoneInvalid(@TempDir Path tmp) throws IOException {
         // Given a column with one normal chunk and one empty chunk (no min/max stats): MIN/MAX is
-        // dropped (it requires every chunk to carry stats), but NULL_COUNT and SUM are still emitted
-        // — SUM is independent (the empty zone's sum is simply null).
+        // still emitted for the column (I64 always resolves a min/max dtype) — only the empty
+        // chunk's own zone is recorded as invalid, not the whole column's MIN/MAX. This is the
+        // #378 fix: a single stats-less chunk (e.g. all-null) must not blank out zone-map pruning
+        // for every other chunk in the column.
         DType.Struct schema = new DType.Struct(
                 List.of(ColumnName.of("v")), List.of(DType.I64), false);
         WriteOptions opts = new WriteOptions(2, true, 0.90, 0, false, false, MemorySize.ofMiB(256), Map.of());
@@ -226,12 +228,27 @@ void chunkWithoutStats_emitsNullCountOnlyZoneMap(@TempDir Path tmp) throws IOExc
             sut.writeChunk(Map.of(ColumnName.of("v"), new long[]{}));
         }
 
-        // When / Then — zoned with the SUM+NULL_COUNT bitset (bits 5+6 = 0x60), no MIN/MAX
+        // When / Then — zoned with the full MAX+MIN+SUM+NULL_COUNT bitset (0x78): zone 0 carries a
+        // valid min/max from its data, zone 1 (the empty chunk) carries an invalid (null) min/max.
         try (VortexReader reader = VortexReader.open(file)) {
             Layout column = reader.layout().children().get(0);
             assertThat(column.isZoned()).isTrue();
             MemorySegment meta = column.metadata();
-            assertThat(meta.get(ValueLayout.JAVA_BYTE, 4)).isEqualTo((byte) 0x60);
+            assertThat(meta.get(ValueLayout.JAVA_BYTE, 4)).isEqualTo((byte) 0x78);
+
+            Layout zonesFlat = column.children().get(1);
+            SegmentSpec spec = reader.footer().segmentSpecs().get(zonesFlat.segments().getFirst());
+            try (Arena arena = Arena.ofConfined()) {
+                StructArray stats = (StructArray) reader.decodeSegment(spec, numericStatsTableDtype(), 2, arena);
+                MaskedArray min = (MaskedArray) stats.field("min");
+                MaskedArray max = (MaskedArray) stats.field("max");
+                assertThat(min.isValid(0)).isTrue();
+                assertThat(max.isValid(0)).isTrue();
+                assertThat(((LongArray) min.inner()).getLong(0)).isEqualTo(1L);
+                assertThat(((LongArray) max.inner()).getLong(0)).isEqualTo(2L);
+                assertThat(min.isValid(1)).isFalse();
+                assertThat(max.isValid(1)).isFalse();
+            }
         }
     }
 
diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoderTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoderTest.java
index 3958c125d..2612ccc59 100644
--- a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoderTest.java
+++ b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoderTest.java
@@ -3,10 +3,12 @@
 import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.reader.array.VarBinArray;
 import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.PType;
 import io.github.dfa1.vortex.core.testing.DTypes;
 import io.github.dfa1.vortex.reader.decode.DecodeContext;
 
 import io.github.dfa1.vortex.core.io.VortexFormat;
+import io.github.dfa1.vortex.core.proto.ProtoScalarValue;
 import io.github.dfa1.vortex.reader.ReadRegistry;
 import io.github.dfa1.vortex.reader.decode.TestRegistry;
 import io.github.dfa1.vortex.core.proto.ProtoDictMetadata;
@@ -18,6 +20,7 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import java.io.IOException;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
 import java.util.stream.Stream;
@@ -200,4 +203,54 @@ void encodeCascade_utf8_exposesCodesAndValuesAsOpenChildren() {
         assertThat(values.childDtype()).isEqualTo(DTypes.UTF8);
         assertThat((String[]) values.childData()).containsExactly("apple", "banana", "cherry");
     }
+
+    @Test
+    void encode_primitive_statsCarryMinAndMaxOfLogicalValues() throws IOException {
+        // Given — low-cardinality so dict encoding is a plausible pick; unordered so a broken
+        // scan (e.g. reading dict-code range instead of logical values) would surface a wrong
+        // min/max. Regression for #379: the primitive branch previously hardcoded null stats,
+        // silently disabling zone-map pruning for any numeric column that dict-encoded.
+        int[] data = {30, 10, 30, 20, 10};
+
+        // When
+        EncodeResult result = ENCODER.encode(DTypes.I32, data, EncodeTestHelper.testCtx());
+
+        // Then — min/max reflect the logical column values (10..30), not the dict codes (0..2)
+        assertThat(result.hasStats()).isTrue();
+        assertThat(scalar(result.statsMin()).int64_value()).isEqualTo(10L);
+        assertThat(scalar(result.statsMax()).int64_value()).isEqualTo(30L);
+    }
+
+    @Test
+    void encodeCascade_primitive_statsCarryMinAndMaxOfLogicalValues() throws IOException {
+        // Given — same #379 regression as encode_primitive_statsCarryMinAndMaxOfLogicalValues, but
+        // for the cascade entry point (CascadingCompressor), the one the "better compression"
+        // option actually drives.
+        long[] data = {30L, 10L, 30L, 20L, 10L};
+
+        // When
+        CascadeStep result = ENCODER.encodeCascade(DTypes.I64, data, EncodeTestHelper.testCtx());
+
+        // Then
+        assertThat(scalar(result.statsMin()).int64_value()).isEqualTo(10L);
+        assertThat(scalar(result.statsMax()).int64_value()).isEqualTo(30L);
+    }
+
+    @Test
+    void encode_primitive_unsignedU32_statsUseUnsignedField() throws IOException {
+        // Given
+        int[] data = {300, 100, 200};
+
+        // When
+        EncodeResult result = ENCODER.encode(new DType.Primitive(PType.U32, false), data, EncodeTestHelper.testCtx());
+
+        // Then
+        assertThat(scalar(result.statsMin()).uint64_value()).isEqualTo(100L);
+        assertThat(scalar(result.statsMax()).uint64_value()).isEqualTo(300L);
+    }
+
+    private static ProtoScalarValue scalar(byte[] bytes) throws IOException {
+        MemorySegment seg = MemorySegment.ofArray(bytes);
+        return ProtoScalarValue.decode(seg, 0, seg.byteSize());
+    }
 }
diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoderTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoderTest.java
index 08dca99ca..c5111c5be 100644
--- a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoderTest.java
+++ b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/FrameOfReferenceEncodingEncoderTest.java
@@ -23,6 +23,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
+import java.io.IOException;
 import java.lang.foreign.MemorySegment;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
@@ -233,5 +234,56 @@ void encodeDecode_i32_isLossless(int[] data) {
                 assertThat(arr.getInt(i)).as("index %d", i).isEqualTo(data[i]);
             }
         }
+
+        @Test
+        void encode_signedI64_statsCarryMinAndMax() throws IOException {
+            // Given — unordered so a broken scan (e.g. reusing the residual instead of the logical
+            // value) would surface a wrong min/max. Regression for #379: FOR previously hardcoded
+            // null stats, silently disabling zone-map pruning for any numeric column the "better
+            // compression" cascading path routed through FOR.
+            long[] data = {1030L, 990L, 1050L, 1020L, 1040L};
+
+            // When
+            EncodeResult result = ENCODER.encode(DTypes.I64, data, EncodeTestHelper.testCtx());
+
+            // Then
+            assertThat(result.hasStats()).isTrue();
+            assertThat(scalar(result.statsMin()).int64_value()).isEqualTo(990L);
+            assertThat(scalar(result.statsMax()).int64_value()).isEqualTo(1050L);
+        }
+
+        @Test
+        void encode_unsignedU32_statsUseUnsignedField() throws IOException {
+            // Given
+            int[] data = {200, 100, 300};
+
+            // When
+            EncodeResult result = ENCODER.encode(new DType.Primitive(PType.U32, false), data, EncodeTestHelper.testCtx());
+
+            // Then
+            assertThat(scalar(result.statsMin()).uint64_value()).isEqualTo(100L);
+            assertThat(scalar(result.statsMax()).uint64_value()).isEqualTo(300L);
+        }
+
+        @Test
+        void encodeCascade_signedI64_statsCarryMinAndMax() throws IOException {
+            // Given — same #379 regression as encode_signedI64_statsCarryMinAndMax, but for the
+            // cascade entry point (CascadingCompressor), the one the "better compression" option
+            // actually drives.
+            long[] data = {1030L, 990L, 1050L, 1020L, 1040L};
+
+            // When
+            CascadeStep result = ENCODER.encodeCascade(DTypes.I64, data, EncodeTestHelper.testCtx());
+
+            // Then
+            assertThat(result.applicable()).isTrue();
+            assertThat(scalar(result.statsMin()).int64_value()).isEqualTo(990L);
+            assertThat(scalar(result.statsMax()).int64_value()).isEqualTo(1050L);
+        }
+
+        private static ProtoScalarValue scalar(byte[] bytes) throws IOException {
+            MemorySegment seg = MemorySegment.ofArray(bytes);
+            return ProtoScalarValue.decode(seg, 0, seg.byteSize());
+        }
     }
 }
diff --git a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoderTest.java b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoderTest.java
index 1127db6dc..4ba5c4da5 100644
--- a/writer/src/test/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoderTest.java
+++ b/writer/src/test/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoderTest.java
@@ -244,6 +244,46 @@ void mixedValidity_stillEncodesAsRawBitmap() {
         assertThat(masked.isValid(2)).isTrue();
     }
 
+    @Test
+    void mixedValidity_statsIgnoreThePlaceholderAtInvalidSlots() throws java.io.IOException {
+        // Given — a `long[]` cannot represent "no value", so NullableData's invalid slots carry a
+        // placeholder (0 here); 0 is not even in the true value set (500..900), and the true min
+        // (500) sits behind a leading invalid slot a naive scan would still see as 0. Regression
+        // guard: MaskedEncodingEncoder must compute stats from (values, validity), excluding invalid
+        // slots — not trust whichever inner encoder ran blindly over the placeholder-filled array.
+        DType i64Nullable = new DType.Primitive(PType.I64, true);
+        NullableData data = new NullableData(
+                new long[]{0L, 500L, 600L, 0L, 700L, 800L, 0L, 900L},
+                new boolean[]{false, true, true, false, true, true, false, true});
+
+        // When
+        EncodeResult result = SUT.encode(i64Nullable, data, EncodeTestHelper.testCtx());
+
+        // Then
+        assertThat(result.hasStats()).isTrue();
+        assertThat(scalar(result.statsMin()).int64_value()).isEqualTo(500L);
+        assertThat(scalar(result.statsMax()).int64_value()).isEqualTo(900L);
+    }
+
+    @Test
+    void allInvalidColumn_hasNoStats() {
+        // Given — every slot invalid: no real value exists to report, so MIN/MAX must be null
+        // (not the 0 placeholder), matching Rust's per-zone-nullable zone-map semantics (#378).
+        DType i64Nullable = new DType.Primitive(PType.I64, true);
+        NullableData data = new NullableData(new long[]{0L, 0L, 0L}, new boolean[]{false, false, false});
+
+        // When
+        EncodeResult result = SUT.encode(i64Nullable, data, EncodeTestHelper.testCtx());
+
+        // Then
+        assertThat(result.hasStats()).isFalse();
+    }
+
+    private static io.github.dfa1.vortex.core.proto.ProtoScalarValue scalar(byte[] bytes) throws java.io.IOException {
+        MemorySegment seg = MemorySegment.ofArray(bytes);
+        return io.github.dfa1.vortex.core.proto.ProtoScalarValue.decode(seg, 0, seg.byteSize());
+    }
+
     @Test
     void withCascade_periodicNulls_prefersSparseOverRawBitmap() {
         // Given — 2000 rows, every 10th null: a clustered/regular pattern (mirrors the real