diff --git a/CHANGELOG.md b/CHANGELOG.md
index f36569f8..b0364702 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - Per-zone stats from current Rust writers (`vortex.zoned`, vortex-jni 0.76.0) decode again. The 0.76 zoned layout replaced the legacy `vortex.stats` bit-set metadata with an aggregate-function spec list and dropped the per-stat truncation flags; the reader now reconstructs the stats table from that spec list (min/max/sum/null_count), so `columnZoneStats` and aggregate push-down work against those files instead of throwing `ClassCastException`. ([#197](https://github.com/dfa1/vortex-java/pull/197))
+- Scans of files whose columns use different chunk grids no longer fail with `mixed per-column chunking beyond 1-vs-N is not supported`. The scan planner now splits at the merged boundary grid — the sorted union of every column's chunk boundaries, matching the Rust reference — and decodes each column's covering chunk once, slicing it zero-copy to each window. This handles both nested grids (Raincloud `emotions-dataset-for-nlp`, where `label`'s coarse chunks nest inside `text`'s finer grid) and disjoint grids (`uci-beijing-multi-site-air-quality`, where numeric and `station` boundaries do not nest). Aligned N-vs-N and 1-vs-N scans keep their existing slice-free fast path. ([#221](https://github.com/dfa1/vortex-java/issues/221))
 - CSV export renders nested struct columns as JSON object cells (`{"field":value,...}`, strings JSON-escaped, null fields as JSON `null`, nested structs recursed) instead of throwing `unsupported array type: StructArray`. ([#217](https://github.com/dfa1/vortex-java/issues/217))
 - Scanning a struct whose columns include a nested struct no longer fails chunk planning. A nested `vortex.struct` layout column (e.g. Raincloud `countries-of-the-world`'s `data` column) is now treated as a single full-range chunk source and decoded through the layout registry's new struct decoder into a `StructArray`, matching the Rust reference (a nested struct spans its parent's row range, not an independent chunking). `schema`/`count`/`inspect` already worked; plain scans and `select` of sibling columns now work too. CSV export of the struct column itself remains unsupported (a separate rendering limitation). ([#207](https://github.com/dfa1/vortex-java/issues/207))
 - CSV export renders unsigned integer columns (U8–U64) with their unsigned values — high-half values previously printed as two's-complement negatives (uci-wine `magnesium` U8 132 exported as -124), silent corruption found by the Raincloud conformance suite. ([#208](https://github.com/dfa1/vortex-java/issues/208))
diff --git a/docs/compatibility.md b/docs/compatibility.md
index b88f27fa..f44a2a9a 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -62,9 +62,11 @@ Per-slug status lives in `integration/src/test/resources/raincloud/expected-stat
 (`ok` must pass; `gap:<issue>` must still fail, so a fix flips the entry in the same change;
 `untriaged` runs and reports without failing the build). A scheduled workflow
 (`raincloud-conformance.yml`) hydrates a size-capped subset weekly. Current triage —
-28 `ok`, 2 known gaps (both misaligned per-column chunk grids,
-[#221](https://github.com/dfa1/vortex-java/issues/221)); 217 slugs untriaged. Fixed so far
-by this suite: lazy dict U8/U16 values ([#206](https://github.com/dfa1/vortex-java/issues/206)),
+30 `ok`, 0 known gaps; 217 slugs untriaged. Fixed so far
+by this suite: misaligned per-column chunk grids
+([#221](https://github.com/dfa1/vortex-java/issues/221) — scan now splits at the merged boundary
+grid, slicing each column's covering chunk per window), lazy dict U8/U16 values
+([#206](https://github.com/dfa1/vortex-java/issues/206)),
 nested struct columns in scan ([#207](https://github.com/dfa1/vortex-java/issues/207)),
 unsigned integers rendered signed ([#208](https://github.com/dfa1/vortex-java/issues/208) /
 [#216](https://github.com/dfa1/vortex-java/issues/216) — silent corruption, incl. wrong filter
diff --git a/integration/src/test/resources/raincloud/expected-status.csv b/integration/src/test/resources/raincloud/expected-status.csv
index 0b64d294..6b284d3e 100644
--- a/integration/src/test/resources/raincloud/expected-status.csv
+++ b/integration/src/test/resources/raincloud/expected-status.csv
@@ -91,7 +91,7 @@ diabetes-health-indicators-dataset,untriaged
 disease-symptom-description-dataset,untriaged
 docmatix-zero-shot,untriaged
 electric-motor-temperature,untriaged
-emotions-dataset-for-nlp,gap:221
+emotions-dataset-for-nlp,ok
 fhv_tripdata_2025,untriaged
 fhvhv_tripdata_2025,untriaged
 finemath-4plus,untriaged
@@ -193,7 +193,7 @@ uci-air-quality,ok
 uci-auto-mpg,ok
 uci-automobile,untriaged
 uci-bank-marketing,ok
-uci-beijing-multi-site-air-quality,gap:221
+uci-beijing-multi-site-air-quality,ok
 uci-bike-sharing-dataset,ok
 uci-breast-cancer,untriaged
 uci-breast-cancer-wisconsin-diagnostic,untriaged
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 e3c95d87..019d72b7 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
@@ -37,12 +37,14 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.IdentityHashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.SequencedMap;
+import java.util.TreeSet;
 import java.util.function.Consumer;
 
 /// Iterates over decoded chunks from a [io.github.dfa1.vortex.reader.VortexReader].
@@ -79,15 +81,26 @@ public final class ScanIterator implements Iterator<Chunk>, AutoCloseable {
     private List<ChunkSpec> chunks;
     private List<ColumnName> projectedNames;
     private List<DType> projectedDtypes;
-    private Map<ColumnName, Layout> columnTopLayouts;
     private Map<ColumnName, DType> columnDtypes;
     private int chunkIndex;
     private int peekedChunkIdx = -1;
     private long rowsReturned;
     private Chunk openChunk;
     private boolean closed;
-    private Arena sharedArena;
-    private Map<ColumnName, Array> sharedFullArrays;
+    // Coarse chunks that span several split windows are decoded once and cached here (keyed by
+    // layout identity), each in its own confined arena so it can be released the moment the scan
+    // advances past its last covering window — see evictPassedFlats.
+    private Map<Layout, CachedFlat> sharedFlats;
+    // Per projected column, the covering flat of the most recently decoded window; drives eviction.
+    private Layout[] lastCoveringFlats;
+
+    /// A coarse covering flat decoded once and sliced across the windows it spans, together with
+    /// the confined [Arena] that owns its buffers (closed on eviction / iterator close).
+    ///
+    /// @param arena the confined arena owning this flat's decoded buffers
+    /// @param array the decoded flat, sliced per window
+    private record CachedFlat(Arena arena, Array array) {
+    }
 
     public ScanIterator(VortexHandle file, ScanOptions options) {
         this.file = file;
@@ -123,63 +136,89 @@ private static void collectFlats(Layout layout, List<Layout> out) {
         }
     }
 
-    private static List<ChunkSpec> buildChunks(Map<ColumnName, List<Layout>> columnFlats) {
+    /// Plans the scan by merging every column's chunk grid into one split grid, mirroring the Rust
+    /// reference: `StructReader::register_splits` unions each field's chunk boundaries into a single
+    /// sorted, deduplicated set (`vortex-layout/src/scan/split_by.rs`), and the scan then walks each
+    /// adjacent-boundary window slicing every column's covering chunk to it. The merged grid refines
+    /// every column's grid, so each window lies entirely within exactly one chunk of each column.
+    ///
+    /// This subsumes the previous special cases without regressing them:
+    /// - aligned N-vs-N (all columns share a grid) — every window equals a whole chunk, so decode is
+    ///   direct with no slicing;
+    /// - 1-vs-N (one full-column flat over a chunked column) — the single flat covers every window
+    ///   and is decoded once, then sliced;
+    /// - nested boundaries (emotions-dataset-for-nlp: `label` `[131072 ×3, 23593]` vs `text`'s
+    ///   26-chunk `16384` grid) and disjoint grids (uci-beijing-multi-site-air-quality: numeric
+    ///   `[131072 ×3, 27552]` vs `station`'s `40960/49152/...` grid) both fall out of the union.
+    static List<ChunkSpec> buildChunks(Map<ColumnName, List<Layout>> columnFlats) {
         if (columnFlats.isEmpty()) {
             return List.of();
         }
         ColumnName[] colNames = columnFlats.keySet().toArray(ColumnName[]::new);
         int numCols = colNames.length;
-        int maxChunks = 0;
-        int refCol = 0;
-        for (int j = 0; j < numCols; j++) {
-            int n = columnFlats.get(colNames[j]).size();
-            if (n > maxChunks) {
-                maxChunks = n;
-                refCol = j;
-            }
-        }
-        // Detect single-flat columns sharing the chunked range of a wider column.
-        // Other mismatched widths (e.g. 5 flats vs 23 flats) are not supported.
-        boolean[] shared = new boolean[numCols];
+
+        // Per-column cumulative chunk starts: colStarts[j][c] is the first row of chunk c and the
+        // final entry is the column's total row count. Chunk c spans [colStarts[j][c], colStarts[j][c+1]).
+        long[][] colStarts = new long[numCols][];
         for (int j = 0; j < numCols; j++) {
-            int n = columnFlats.get(colNames[j]).size();
-            if (n == maxChunks) {
-                continue;
+            List<Layout> flats = columnFlats.get(colNames[j]);
+            long[] starts = new long[flats.size() + 1];
+            long acc = 0;
+            for (int c = 0; c < flats.size(); c++) {
+                starts[c] = acc;
+                acc += flats.get(c).rowCount();
             }
-            if (n == 1) {
-                shared[j] = true;
-            } else {
-                throw new VortexException(
-                        "scan: column '" + colNames[j] + "' has " + n
-                                + " flats but the widest column has " + maxChunks
-                                + "; mixed per-column chunking beyond 1-vs-N is not supported");
+            starts[flats.size()] = acc;
+            colStarts[j] = starts;
+        }
+
+        long[] boundaries = mergedBoundaries(colStarts);
+        int numWindows = boundaries.length - 1;
+        var result = new ArrayList<ChunkSpec>(numWindows);
+        int[] cursor = new int[numCols]; // covering chunk index per column; advances monotonically
+        for (int w = 0; w < numWindows; w++) {
+            long windowStart = boundaries[w];
+            long windowRows = boundaries[w + 1] - windowStart;
+            Layout[] layouts = new Layout[numCols];
+            long[] sliceOffsets = new long[numCols];
+            for (int j = 0; j < numCols; j++) {
+                long[] starts = colStarts[j];
+                int c = cursor[j];
+                // Advance to the chunk whose range contains windowStart (starts[c+1] > windowStart).
+                while (c + 1 < starts.length && starts[c + 1] <= windowStart) {
+                    c++;
+                }
+                cursor[j] = c;
+                List<Layout> flats = columnFlats.get(colNames[j]);
+                if (c >= flats.size()) {
+                    throw new VortexException("scan: column '" + colNames[j]
+                            + "' has no chunk covering rows [" + windowStart + ", "
+                            + (windowStart + windowRows) + ")");
+                }
+                layouts[j] = flats.get(c);
+                sliceOffsets[j] = windowStart - starts[c];
             }
-        }
-        var result = new ArrayList<ChunkSpec>(maxChunks);
-        long sliceStart = 0;
-        for (int i = 0; i < maxChunks; i++) {
-            long chunkRowCount = columnFlats.get(colNames[refCol]).get(i).rowCount();
-            result.add(buildChunkSpec(colNames, columnFlats, shared, i, sliceStart, chunkRowCount));
-            sliceStart += chunkRowCount;
+            result.add(new ChunkSpec(windowRows, colNames, layouts, sliceOffsets));
         }
         return List.copyOf(result);
     }
 
-    private static ChunkSpec buildChunkSpec(ColumnName[] colNames, Map<ColumnName, List<Layout>> columnFlats,
-            boolean[] shared, int chunkIdx, long sliceStart, long chunkRowCount) {
-        int numCols = colNames.length;
-        Layout[] layouts = new Layout[numCols];
-        long[] sliceOffsets = new long[numCols];
-        for (int j = 0; j < numCols; j++) {
-            if (shared[j]) {
-                layouts[j] = null;
-                sliceOffsets[j] = sliceStart;
-            } else {
-                layouts[j] = columnFlats.get(colNames[j]).get(chunkIdx);
-                sliceOffsets[j] = 0L;
+    /// Returns the sorted, deduplicated union of every column's cumulative chunk starts (each ending
+    /// in the column's total row count). The result always contains `0` and the total; adjacent
+    /// entries are the scan's split windows.
+    private static long[] mergedBoundaries(long[][] colStarts) {
+        var set = new TreeSet<Long>();
+        for (long[] starts : colStarts) {
+            for (long s : starts) {
+                set.add(s);
             }
         }
-        return new ChunkSpec(chunkRowCount, colNames, layouts, sliceOffsets);
+        long[] out = new long[set.size()];
+        int i = 0;
+        for (long v : set) {
+            out[i++] = v;
+        }
+        return out;
     }
 
     // ── Layout tree traversal ─────────────────────────────────────────────────
@@ -268,6 +307,8 @@ public Chunk next() {
         chunkIndex = peekedChunkIdx + 1;
         peekedChunkIdx = -1;
 
+        evictPassedFlats(spec);
+
         long remaining = options.limit() - rowsReturned;
         long chunkRows = Math.min(spec.rowCount(), remaining);
 
@@ -363,11 +404,14 @@ public long[] chunkRowCounts() {
     /// `null`, since the flat writer does not retain it). Either way a column that is absent
     /// or carries no stats yields [ArrayStats#empty()] per zone.
     ///
-    /// Zone granularity is the layout's, not the scan's. The fallback path is one entry per
-    /// chunk, positionally aligned with [#chunkRowCounts()]. The zone-map path is one entry per
-    /// zone of the stats table: this writer emits one zone per chunk (so the same alignment
-    /// holds), but a file from another writer may use a fixed zone length independent of chunk
-    /// boundaries, in which case the zone count need not match [#chunkRowCounts()].
+    /// Zone granularity is the layout's, not the scan's. The fallback path is one entry per scan
+    /// window, positionally aligned with [#chunkRowCounts()]. Each entry carries the stats of the
+    /// covering chunk of that window, so when columns chunk on different grids a coarse chunk's
+    /// stats repeat across every sub-window it spans (still a conservative min/max/null-count bound
+    /// for each sub-window). The zone-map path is one entry per zone of the stats table: this writer
+    /// emits one zone per chunk (so the same alignment holds), but a file from another writer may
+    /// use a fixed zone length independent of chunk boundaries, in which case the zone count need
+    /// not match [#chunkRowCounts()].
     ///
     /// 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
@@ -537,10 +581,11 @@ public void close() {
             openChunk.close();
         }
         openChunk = null;
-        if (sharedArena != null) {
-            sharedArena.close();
-            sharedArena = null;
-            sharedFullArrays = null;
+        if (sharedFlats != null) {
+            for (CachedFlat cached : sharedFlats.values()) {
+                cached.arena().close();
+            }
+            sharedFlats = null;
         }
     }
 
@@ -555,7 +600,6 @@ private void initialize() {
         DType rootDtype = file.dtype();
 
         var columnFlats = new LinkedHashMap<ColumnName, List<Layout>>();
-        var columnTopLayouts = new LinkedHashMap<ColumnName, Layout>();
         Map<ColumnName, DType> columnDtypes = new LinkedHashMap<>();
 
         if (rootLayout.isStruct() && rootDtype instanceof DType.Struct structDtype) {
@@ -570,7 +614,6 @@ private void initialize() {
                 var flats = new ArrayList<Layout>();
                 collectFlats(colTop, flats);
                 columnFlats.put(colName, flats);
-                columnTopLayouts.put(colName, colTop);
                 columnDtypes.put(colName, colDtype);
             }
         } else {
@@ -578,43 +621,13 @@ private void initialize() {
             collectFlats(rootLayout, flats);
             ColumnName colName = ColumnName.of("_col");
             columnFlats.put(colName, flats);
-            columnTopLayouts.put(colName, rootLayout);
             columnDtypes.put(colName, rootDtype);
         }
 
         projectedNames = List.copyOf(columnDtypes.keySet());
         projectedDtypes = List.copyOf(columnDtypes.values());
-        this.columnTopLayouts = Map.copyOf(columnTopLayouts);
+        lastCoveringFlats = new Layout[projectedNames.size()];
         chunks = buildChunks(columnFlats);
-        decodeSharedColumns(columnFlats, columnTopLayouts, columnDtypes);
-    }
-
-    private void decodeSharedColumns(
-            Map<ColumnName, List<Layout>> columnFlats,
-            Map<ColumnName, Layout> columnTopLayouts,
-            Map<ColumnName, DType> columnDtypes) {
-        int maxFlats = 0;
-        for (List<Layout> flats : columnFlats.values()) {
-            if (flats.size() > maxFlats) {
-                maxFlats = flats.size();
-            }
-        }
-        if (maxFlats <= 1) {
-            return;
-        }
-        for (var entry : columnFlats.entrySet()) {
-            if (entry.getValue().size() != 1) {
-                continue;
-            }
-            if (sharedArena == null) {
-                sharedArena = Arena.ofConfined();
-                sharedFullArrays = new HashMap<>();
-            }
-            ColumnName name = entry.getKey();
-            Layout topLayout = columnTopLayouts.get(name);
-            DType dtype = columnDtypes.get(name);
-            sharedFullArrays.put(name, decodeLayout(topLayout, dtype, sharedArena));
-        }
     }
 
     // A LinkedHashMap preserves schema/projection order (the public columns() contract is a
@@ -640,8 +653,8 @@ private SequencedMap<ColumnName, Chunk.Column> buildColumnMap(ChunkSpec chunk, A
     }
 
     /// Builds the column map for [#decodeChunkAt(int)]. Identical decode to [#buildColumnMap]
-    /// except that a shared (single-flat) column is decoded into `arena` and sliced there,
-    /// so the resulting [Chunk] owns every buffer and survives this iterator's close.
+    /// except that a covering chunk spanning several windows is decoded into `arena` and sliced
+    /// there, so the resulting [Chunk] owns every buffer and survives this iterator's close.
     private SequencedMap<ColumnName, Chunk.Column> buildSelfContainedColumnMap(ChunkSpec chunk, Arena arena) {
         Layout[] layouts = chunk.columnLayouts();
         long[] sliceOffsets = chunk.sliceOffsets();
@@ -668,30 +681,67 @@ private SequencedMap<ColumnName, Chunk.Column> singleColumn(Array array) {
         return unmodifiable(map);
     }
 
-    private Array decodeOrSliceSelfContained(int colIdx, Layout layout, long sliceStart,
+    private Array decodeOrSliceSelfContained(int colIdx, Layout coveringFlat, long sliceOffset,
                                              long rowCount, Arena arena) {
-        if (layout != null) {
-            return decodeLayout(layout, projectedDtypes.get(colIdx), arena);
-        }
-        // Shared single-flat column: decode its full top layout into THIS chunk's arena and
-        // slice, so the returned Chunk does not reference the iterator's shared arena.
-        ColumnName name = projectedNames.get(colIdx);
         DType dtype = projectedDtypes.get(colIdx);
-        Array full = decodeLayout(columnTopLayouts.get(name), dtype, arena);
-        return sliceArray(full, sliceStart, rowCount, dtype);
+        Array full = decodeLayout(coveringFlat, dtype, arena);
+        if (isWholeFlat(coveringFlat, sliceOffset, rowCount)) {
+            return full;
+        }
+        // Covering chunk spans several windows: slice it into THIS chunk's arena so the returned
+        // Chunk owns every buffer and survives this iterator's close.
+        return sliceArray(full, sliceOffset, rowCount, dtype);
     }
 
-    private Array decodeOrSlice(int colIdx, Layout layout, long sliceStart, long rowCount,
+    private Array decodeOrSlice(int colIdx, Layout coveringFlat, long sliceOffset, long rowCount,
                                 Arena arena) {
-        if (layout != null) {
-            return decodeLayout(layout, projectedDtypes.get(colIdx), arena);
-        }
-        Array full = sharedFullArrays.get(projectedNames.get(colIdx));
-        if (full == null) {
-            throw new VortexException("scan: missing shared array for column "
-                    + projectedNames.get(colIdx));
+        DType dtype = projectedDtypes.get(colIdx);
+        if (isWholeFlat(coveringFlat, sliceOffset, rowCount)) {
+            // Fast path: the window is a whole chunk — decode straight into the chunk's arena with
+            // no slice wrapper (aligned N-vs-N and per-chunk decode keep their zero-copy behavior).
+            return decodeLayout(coveringFlat, dtype, arena);
+        }
+        // Covering chunk spans several windows: decode it once (cached) and slice zero-copy per
+        // window, matching how Rust decodes a chunk once and slices each split range.
+        Array full = sharedCoveringFlat(coveringFlat, dtype);
+        return sliceArray(full, sliceOffset, rowCount, dtype);
+    }
+
+    private static boolean isWholeFlat(Layout coveringFlat, long sliceOffset, long rowCount) {
+        return sliceOffset == 0 && rowCount == coveringFlat.rowCount();
+    }
+
+    /// Decodes `flat` once into its own confined arena and caches it by identity, so a coarse chunk
+    /// that covers several split windows is decoded a single time and then sliced per window. The
+    /// arena is released by [#evictPassedFlats] once the scan advances off the flat.
+    private Array sharedCoveringFlat(Layout flat, DType dtype) {
+        if (sharedFlats == null) {
+            sharedFlats = new IdentityHashMap<>();
+        }
+        return sharedFlats.computeIfAbsent(flat, f -> {
+            Arena flatArena = Arena.ofConfined();
+            return new CachedFlat(flatArena, decodeLayout(f, dtype, flatArena));
+        }).array();
+    }
+
+    /// Releases every cached coarse flat the scan has moved past. As `spec` is decoded, a column
+    /// whose covering flat differs from the one it used in the previously decoded window can never
+    /// revisit that earlier flat — the planner's per-column cursor advances monotonically — so its
+    /// arena is closed and its cache entry dropped. Whole-window (uncached) flats are ignored. The
+    /// previously open [Chunk] is already closed here (enforced by [#next()]), so no live slice
+    /// still references the freed arena.
+    private void evictPassedFlats(ChunkSpec spec) {
+        Layout[] current = spec.columnLayouts();
+        for (int j = 0; j < current.length; j++) {
+            Layout previous = lastCoveringFlats[j];
+            if (previous != null && previous != current[j]) {
+                CachedFlat cached = sharedFlats == null ? null : sharedFlats.remove(previous);
+                if (cached != null) {
+                    cached.arena().close();
+                }
+            }
+            lastCoveringFlats[j] = current[j];
         }
-        return sliceArray(full, sliceStart, rowCount, projectedDtypes.get(colIdx));
     }
 
     private static Array sliceArray(Array full, long offset, long length, DType dtype) {
@@ -873,7 +923,7 @@ public SegmentSpec segmentSpec(int index) {
     // ── Internal record ───────────────────────────────────────────────────────
 
     @SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared.
-    private record ChunkSpec(
+    record ChunkSpec(
             long rowCount, ColumnName[] columnNames, Layout[] columnLayouts, long[] sliceOffsets) {
         Layout layoutFor(ColumnName col) {
             for (int i = 0; i < columnNames.length; i++) {
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
new file mode 100644
index 00000000..91a7d35d
--- /dev/null
+++ b/reader/src/test/java/io/github/dfa1/vortex/reader/ScanIteratorChunkGridTest.java
@@ -0,0 +1,190 @@
+package io.github.dfa1.vortex.reader;
+
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.LayoutId;
+import io.github.dfa1.vortex.reader.ScanIterator.ChunkSpec;
+import io.github.dfa1.vortex.reader.layout.Layout;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/// Unit tests for [ScanIterator#buildChunks(Map)] — the scan planner that merges each column's
+/// chunk grid into one split grid (the Rust reference's `StructReader::register_splits` union of
+/// field boundaries). The scaled-down grids below mirror the two real corpus files from #221:
+/// `emotions-dataset-for-nlp` (nested boundaries) and `uci-beijing-multi-site-air-quality`
+/// (disjoint boundaries).
+class ScanIteratorChunkGridTest {
+
+    private static final ColumnName A = ColumnName.of("a");
+    private static final ColumnName B = ColumnName.of("b");
+
+    @Test
+    void alignedGridsDecodeEveryColumnAsAWholeChunk() {
+        // Given two columns with the *same* grid [4, 4] over 8 rows — the aligned N-vs-N fast path.
+        var columnFlats = flats(A, new long[]{4, 4}, B, new long[]{4, 4});
+
+        // When
+        List<ChunkSpec> result = ScanIterator.buildChunks(columnFlats);
+
+        // Then the merged grid equals the shared grid: two windows, and every column's window is a
+        // whole chunk (offset 0, window rows == covering-flat rows) so decode stays slice-free.
+        assertThat(rowCounts(result)).containsExactly(4L, 4L);
+        assertWholeChunk(result.get(0), A, 4);
+        assertWholeChunk(result.get(0), B, 4);
+        assertWholeChunk(result.get(1), A, 4);
+        assertWholeChunk(result.get(1), B, 4);
+    }
+
+    @Test
+    void singleFlatColumnSharesTheChunkedGrid() {
+        // Given one full-column flat [8] beside a chunked column [4, 4] — the 1-vs-N case.
+        var columnFlats = flats(A, new long[]{8}, B, new long[]{4, 4});
+
+        // When
+        List<ChunkSpec> result = ScanIterator.buildChunks(columnFlats);
+
+        // Then the single flat is the covering chunk of both windows, sliced at 0 then 4, while the
+        // chunked column contributes a whole chunk per window.
+        assertThat(rowCounts(result)).containsExactly(4L, 4L);
+        assertSlice(result.get(0), A, 8, 0);
+        assertWholeChunk(result.get(0), B, 4);
+        assertSlice(result.get(1), A, 8, 4);
+        assertWholeChunk(result.get(1), B, 4);
+    }
+
+    @Test
+    void nestedBoundariesSliceTheCoarseColumnAtTheFineGrid() {
+        // Given the emotions-dataset-for-nlp shape scaled down: a coarse column [8, 8, 8, 4] (like
+        // `label`'s [131072 ×3, 23593]) beside a fine column of 2-row chunks (like `text`'s 16384
+        // grid, where 8 = 4 × 2 so every coarse boundary is also a fine boundary).
+        var columnFlats = flats(
+                A, new long[]{8, 8, 8, 4},
+                B, new long[]{2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2});
+
+        // When
+        List<ChunkSpec> result = ScanIterator.buildChunks(columnFlats);
+
+        // Then the merged grid is the fine grid (it already contains every coarse boundary): 14
+        // windows of 2. The fine column is a whole chunk each window; the coarse column is sliced
+        // within whichever coarse chunk covers the window, so it is decoded once per coarse chunk.
+        assertThat(result).hasSize(14);
+        assertThat(rowCounts(result)).allMatch(n -> n == 2L);
+        // Rows [0, 8) fall inside coarse chunk 0 at offsets 0, 2, 4, 6.
+        assertSlice(result.get(0), A, 8, 0);
+        assertSlice(result.get(3), A, 8, 6);
+        // Rows [8, 16) fall inside coarse chunk 1 (rowCount 8) at offsets 0, 2, 4, 6.
+        assertSlice(result.get(4), A, 8, 0);
+        assertSlice(result.get(7), A, 8, 6);
+        // Rows [24, 28) fall inside coarse chunk 3 (rowCount 4) at offsets 0, 2.
+        assertSlice(result.get(12), A, 4, 0);
+        assertSlice(result.get(13), A, 4, 2);
+        // The fine column is always a whole 2-row chunk.
+        for (ChunkSpec spec : result) {
+            assertWholeChunk(spec, B, 2);
+        }
+    }
+
+    @Test
+    void disjointBoundariesSpliceAtTheMergedGrid() {
+        // Given the uci-beijing-multi-site-air-quality shape scaled down: [3, 3, 2] vs [4, 4] over 8
+        // rows. The boundaries do not nest (3 + 3 = 6 != 4, mirroring 40960 + 49152 != 131072), so
+        // neither grid refines the other — the scan must emit windows at the union of boundaries.
+        var columnFlats = flats(
+                A, new long[]{3, 3, 2},
+                B, new long[]{4, 4});
+
+        // When
+        List<ChunkSpec> result = ScanIterator.buildChunks(columnFlats);
+
+        // Then the merged grid is {0, 3, 4, 6, 8} -> windows [0,3) [3,4) [4,6) [6,8), and every
+        // column slices its covering chunk to each window (a chunk spanning several windows is
+        // named identically across them, so it is decoded once and sliced).
+        assertThat(rowCounts(result)).containsExactly(3L, 1L, 2L, 2L);
+        // Column A [3, 3, 2]: chunk0 covers [0,3) whole; chunk1 covers [3,4) and [4,6); chunk2 whole.
+        assertWholeChunk(result.get(0), A, 3);
+        assertSlice(result.get(1), A, 3, 0);
+        assertSlice(result.get(2), A, 3, 1);
+        assertThat(coveringFlat(result.get(1), A)).isSameAs(coveringFlat(result.get(2), A));
+        assertWholeChunk(result.get(3), A, 2);
+        // Column B [4, 4]: chunk0 covers [0,3) and [3,4); chunk1 covers [4,6) and [6,8).
+        assertSlice(result.get(0), B, 4, 0);
+        assertSlice(result.get(1), B, 4, 3);
+        assertThat(coveringFlat(result.get(0), B)).isSameAs(coveringFlat(result.get(1), B));
+        assertSlice(result.get(2), B, 4, 0);
+        assertSlice(result.get(3), B, 4, 2);
+        assertThat(coveringFlat(result.get(2), B)).isSameAs(coveringFlat(result.get(3), B));
+    }
+
+    @Test
+    void emptyProjectionYieldsNoChunks() {
+        // Given no columns
+        var columnFlats = new LinkedHashMap<ColumnName, List<Layout>>();
+
+        // When
+        List<ChunkSpec> result = ScanIterator.buildChunks(columnFlats);
+
+        // Then
+        assertThat(result).isEmpty();
+    }
+
+    // ── helpers ───────────────────────────────────────────────────────────────
+
+    /// Builds an insertion-ordered `column -> flats` map for two columns. A [LinkedHashMap] states
+    /// the intended column order explicitly (the planner's output is order-independent and read
+    /// back by name here, but an unordered [Map#of] would hide that intent).
+    private static Map<ColumnName, List<Layout>> flats(ColumnName n1, long[] g1, ColumnName n2, long[] g2) {
+        var out = new LinkedHashMap<ColumnName, List<Layout>>();
+        out.put(n1, toFlats(g1));
+        out.put(n2, toFlats(g2));
+        return out;
+    }
+
+    private static List<Layout> toFlats(long[] chunkRows) {
+        var list = new ArrayList<Layout>(chunkRows.length);
+        for (long rows : chunkRows) {
+            list.add(new Layout(LayoutId.FLAT, rows, null, List.of(), List.of()));
+        }
+        return list;
+    }
+
+    private static List<Long> rowCounts(List<ChunkSpec> chunks) {
+        var out = new ArrayList<Long>(chunks.size());
+        for (ChunkSpec spec : chunks) {
+            out.add(spec.rowCount());
+        }
+        return out;
+    }
+
+    private static Layout coveringFlat(ChunkSpec spec, ColumnName column) {
+        return spec.layoutFor(column);
+    }
+
+    private static long sliceOffset(ChunkSpec spec, ColumnName column) {
+        ColumnName[] names = spec.columnNames();
+        for (int i = 0; i < names.length; i++) {
+            if (names[i].equals(column)) {
+                return spec.sliceOffsets()[i];
+            }
+        }
+        throw new IllegalStateException("no such column " + column);
+    }
+
+    /// Asserts the window decodes `column` as a whole covering chunk (fast path: offset 0 and the
+    /// window's rows exactly equal the covering flat's rows, so no slice wrapper is created).
+    private static void assertWholeChunk(ChunkSpec spec, ColumnName column, long flatRows) {
+        assertThat(coveringFlat(spec, column).rowCount()).isEqualTo(flatRows);
+        assertThat(sliceOffset(spec, column)).isZero();
+        assertThat(spec.rowCount()).isEqualTo(flatRows);
+    }
+
+    /// Asserts the window slices `column`'s covering chunk (of `flatRows` rows) at `offset`.
+    private static void assertSlice(ChunkSpec spec, ColumnName column, long flatRows, long offset) {
+        assertThat(coveringFlat(spec, column).rowCount()).isEqualTo(flatRows);
+        assertThat(sliceOffset(spec, column)).isEqualTo(offset);
+    }
+}
diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/VortexReaderDecodeChunkTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/VortexReaderDecodeChunkTest.java
index da7ffc28..334c9009 100644
--- a/reader/src/test/java/io/github/dfa1/vortex/reader/VortexReaderDecodeChunkTest.java
+++ b/reader/src/test/java/io/github/dfa1/vortex/reader/VortexReaderDecodeChunkTest.java
@@ -262,6 +262,54 @@ void decodeChunk_sharedSingleFlatColumn_isValueEquivalentAndSurvivesIteratorClos
         }
     }
 
+    // A genuinely DISJOINT per-column chunk grid (issue #221, "beijing" tier): column "a" chunks
+    // [3, 3, 2] and column "b" chunks [4, 4] over the same 8 rows. The boundaries do not nest
+    // ({0,3,6,8} vs {0,4,8}), so neither grid refines the other; the scan must split at the union
+    // {0,3,4,6,8}. "b" is nullable so the coarse-chunk slice also crosses the MaskedArray branch.
+    private static final long[] MIXED_A_ROWS = {3, 3, 2};
+    private static final long[][] MIXED_A = {{10, 11, 12}, {13, 14, 15}, {16, 17}};
+    private static final long[] MIXED_B_ROWS = {4, 4};
+    private static final long[][] MIXED_B = {{20, 21, 22, 23}, {24, 25, 26, 27}};
+    private static final boolean[][] MIXED_B_VALID = {{true, false, true, true}, {true, true, false, true}};
+    private static final List<Long> MIXED_A_EXPECTED = List.of(10L, 11L, 12L, 13L, 14L, 15L, 16L, 17L);
+    private static final List<Long> MIXED_B_EXPECTED =
+            java.util.Arrays.asList(20L, null, 22L, 23L, 24L, 25L, null, 27L);
+
+    @Test
+    void scan_disjointMixedGrid_streamsExactValuesAcrossMergedWindows(@TempDir Path tmp) throws Exception {
+        // This is the value-level ground truth for #221's coarse-sub-chunk slice path. Neither the
+        // Java VortexWriter (its writeChunk requires all columns to agree on row count) nor the JNI
+        // writer (it writes uniform row batches) can EMIT a disjoint per-column grid, so the file is
+        // hand-assembled here; Rust-parity of the fix is covered separately by the size-gated
+        // RaincloudConformanceIntegrationTest (uci-beijing-multi-site-air-quality). Streaming the
+        // whole scan exercises the merged-grid planner end to end AND, by fully iterating, drives
+        // the coarse-flat eviction path (earlier chunks are released while later windows must still
+        // decode correctly). The ground truth is the exact values written into each segment.
+
+        // Given a file whose two columns chunk on disjoint grids [3,3,2] vs [4,4]
+        Path file = writeMixedGridFile(tmp);
+        var windowRowCounts = new ArrayList<Long>();
+        var streamedA = new ArrayList<Long>();
+        var streamedB = new ArrayList<Long>();
+
+        // When streaming the whole scan
+        try (var reader = VortexReader.open(file, registry());
+                var iter = reader.scan(ScanOptions.all())) {
+            iter.forEachRemaining(chunk -> {
+                windowRowCounts.add(chunk.rowCount());
+                streamedA.addAll(values(chunk.column("a")));
+                streamedB.addAll(values(chunk.column("b")));
+            });
+        }
+
+        // Then the scan splits at the merged grid {0,3,4,6,8} -> windows of 3,1,2,2 rows, and every
+        // value (and null) survives the coarse-chunk sub-window slice — including "a"'s middle chunk
+        // sliced at a nonzero offset for window [4,6) and "b"'s first chunk sliced for [0,3)+[3,4).
+        assertThat(windowRowCounts).containsExactly(3L, 1L, 2L, 2L);
+        assertThat(streamedA).isEqualTo(MIXED_A_EXPECTED);
+        assertThat(streamedB).isEqualTo(MIXED_B_EXPECTED);
+    }
+
     // ── Helpers ────────────────────────────────────────────────────────────────
 
     private static ReadRegistry registry() {
@@ -302,6 +350,76 @@ private static Path writeMultiChunkFile(Path dir) throws Exception {
         return writeFile(dir, "multi.vtx", CHUNK_ROWS, COL_A, COL_B, COL_B_VALID);
     }
 
+    /// Assembles a struct file whose columns chunk on independent grids: "a" over [MIXED_A_ROWS]
+    /// (non-nullable) and "b" over [MIXED_B_ROWS] (nullable). Segment order: a-chunk0..a-chunkN,
+    /// then b-chunk0..b-chunkM. Layout: flat=0, chunked=1, struct=2; array specs: primitive=0,
+    /// bool=1.
+    private static Path writeMixedGridFile(Path dir) throws Exception {
+        List<byte[]> segments = new ArrayList<>();
+        for (long[] chunk : MIXED_A) {
+            segments.add(primitiveSegment(chunk, null));
+        }
+        int bBase = segments.size();
+        for (int k = 0; k < MIXED_B.length; k++) {
+            segments.add(primitiveSegment(MIXED_B[k], MIXED_B_VALID[k]));
+        }
+
+        long[] segOffsets = new long[segments.size()];
+        long[] segLengths = new long[segments.size()];
+        long off = 0;
+        for (int i = 0; i < segments.size(); i++) {
+            segOffsets[i] = off;
+            segLengths[i] = segments.get(i).length;
+            off += segments.get(i).length;
+        }
+
+        int[] aSeg = new int[MIXED_A_ROWS.length];
+        for (int k = 0; k < aSeg.length; k++) {
+            aSeg[k] = k;
+        }
+        int[] bSeg = new int[MIXED_B_ROWS.length];
+        for (int k = 0; k < bSeg.length; k++) {
+            bSeg[k] = bBase + k;
+        }
+
+        ByteBuffer footerBuf = MalformedFiles.buildFooter(
+                new String[]{"vortex.primitive", "vortex.bool"},
+                new String[]{"vortex.flat", "vortex.chunked", "vortex.struct"},
+                segOffsets, segLengths);
+        ByteBuffer dtypeBuf = buildStructDtype();
+        ByteBuffer layoutBuf = buildMixedLayout(aSeg, MIXED_A_ROWS, bSeg, MIXED_B_ROWS);
+
+        return writeVtxFile(dir, "mixed.vtx", segments, footerBuf, dtypeBuf, layoutBuf);
+    }
+
+    /// Layout for the disjoint-grid file: two chunked columns whose flats carry independent
+    /// row counts, so the reader's planner must split at the union of both grids.
+    private static ByteBuffer buildMixedLayout(int[] aSeg, long[] aRows, int[] bSeg, long[] bRows) {
+        var fbb = new FbsBuilder(1024);
+        long total = 0;
+        for (long r : aRows) {
+            total += r;
+        }
+        int chunkedA = buildChunkedColumn(fbb, aSeg, aRows, total);
+        int chunkedB = buildChunkedColumn(fbb, bSeg, bRows, total);
+        int structChildV = FbsLayout.createChildrenVector(fbb, new int[]{chunkedA, chunkedB});
+        int structOff = FbsLayout.createFbsLayout(fbb, 2, total, 0, structChildV, 0);
+        FbsLayout.finishFbsLayoutBuffer(fbb, structOff);
+        return MalformedFiles.slice(fbb);
+    }
+
+    /// Builds one `vortex.chunked` layout node whose children are one `vortex.flat` per chunk,
+    /// each flat carrying its own `rows[k]` row count and single segment index `segIdx[k]`.
+    private static int buildChunkedColumn(FbsBuilder fbb, int[] segIdx, long[] rows, long total) {
+        int[] flatOffs = new int[rows.length];
+        for (int k = 0; k < rows.length; k++) {
+            int segV = FbsLayout.createSegmentsVector(fbb, new long[]{segIdx[k]});
+            flatOffs[k] = FbsLayout.createFbsLayout(fbb, 0, rows[k], 0, 0, segV);
+        }
+        int childV = FbsLayout.createChildrenVector(fbb, flatOffs);
+        return FbsLayout.createFbsLayout(fbb, 1, total, 0, childV, 0);
+    }
+
     /// Streams the whole scan of the shared-column file, collecting the per-chunk column "a" and
     /// the per-chunk slice of the single-flat shared column "c" so a random-access decode can be
     /// compared element-by-element.