feat(reader): LayoutDecoder SPI — layout decode is pluggable

LayoutDecoder + LayoutRegistry in reader.layout mirror the
ReadRegistry idiom (builder final-freeze, string-keyed dispatch,
duplicate registration throws, no service file — programmatic
registration like ExtensionDecoder). The four built-ins move out
of ScanIterator verbatim: Flat, Chunked, Zoned (claims both the
canonical vortex.zoned and legacy vortex.stats ids via the
layoutIds() set), Dict. ScanIterator.decodeLayout is now one
registry call; zone-map pruning and chunk planning keep inspecting
built-ins only — the SPI covers full-column subtree decode.

Wired end-to-end per the no-decorative-flags rule: VortexHandle
gains layoutRegistry(), both readers take open(..., LayoutRegistry)
overloads, and a scan through a custom registry is proven by test.
Unknown layouts still fail loudly (Rust default). Reverses the
"Layout is a fixed set, no SPI" design decision — the reference
implementation treats layouts as runtime-pluggable.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: dfa1

CLAUDE.md

@@ -23,17 +23,18 @@ Benchmark classes follow this: `JavaVsJni{Read,Write,Filter}Benchmark`,
 
 ```
 core    — everything lives under `io.github.dfa1.vortex.core.*`:
-          core.model    DType, PType, TimeUnit, EncodingId, ExtensionId, TimeDtype, TimestampDtype
+          core.model    DType, PType, TimeUnit, EncodingId, LayoutId, ExtensionId, TimeDtype, TimestampDtype
           core.io       IoBounds, PTypeIO, VortexFormat
           core.error    VortexException
           core.compute  FastLanes, PrimitiveArrays
           core.fbs / core.proto — generated wire codecs + their runtimes
 reader  — VortexReader, VortexHttpReader, VortexHandle, ReadRegistry, Chunk, ArrayStats,
-          ScanOptions, RowFilter; file internals (Footer, Layout, Trailer,
-          PostscriptParser, …)
+          ScanOptions, RowFilter; file internals (Footer, Trailer, PostscriptParser, …)
           reader.array  — Array + all subtypes (decode outputs)
           reader.decode — EncodingDecoder, DecodeContext, ArrayNode + *EncodingDecoder impls
           reader.extension — ExtensionDecoder + Date/Time/Timestamp/Uuid impls
+          reader.layout — Layout, LayoutDecoder, LayoutDecodeContext, LayoutRegistry
+          + built-in *LayoutDecoder impls, ZonedStatsSchema
 writer  — VortexWriter, WriteRegistry, WriteOptions, ExtensionEncoder
           writer.encode — EncodingEncoder, EncodeContext, NullableData + *EncodingEncoder impls,
           extension encoders
@@ -194,9 +195,13 @@ in the Rust source for the exact schema, then implement from spec.
   not add variants. Use `new DType.Extension("ip.address", new DType.Primitive(PType.I32, false),
   null, false)` and register decoders/encoders on the registries (or `ServiceLoader<ExtensionEncoder>`).
   Mirrors Rust (`vortex.date`, `vortex.uuid`, …). No SPI for DType variants planned.
-- **Layout is a fixed set, no SPI.** `ScanIterator.decodeLayout()` dispatches the known IDs
-  (flat/chunked/zoned/struct/dict) and throws otherwise. Keep the fixed set; revisit only for a
-  concrete downstream case unaddressable by a different flat-segment encoding.
+- **Layout decode is pluggable via `LayoutDecoder` + `LayoutRegistry`** (`reader.layout`) — the
+  Rust reference registers layouts at runtime, so ours are open too. Builder-registered only
+  (`LayoutRegistry.builder().registerDefaults().register(custom).build()`, pass to
+  `VortexReader.open(path, readRegistry, layoutRegistry)`) — **no service file**. Unknown layouts
+  fail loudly (`VortexException`, Rust default; no allowUnknown for layouts). Scope: the SPI covers
+  full-column subtree decode; zone-map pruning, filtered scans, and chunk planning recognize the
+  built-in layouts only.
 - **Small public APIs.** Don't expose internals — when in doubt, leave it out or make it private.
 - **POM deps** grouped with comments: `<!-- production -->` then `<!-- testing -->`, each with
   project-internal (`io.github.dfa1.vortex:*`) deps first, then external. Omit empty sections.

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

@@ -3,6 +3,7 @@
 import io.github.dfa1.vortex.core.model.DType;
 import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.reader.layout.Layout;
+import io.github.dfa1.vortex.reader.layout.LayoutRegistry;
 
 import java.io.Closeable;
 import java.lang.foreign.MemorySegment;
@@ -57,6 +58,13 @@ public interface VortexHandle extends Closeable {
     /// @return the registry used to resolve encoding ids during scan
     ReadRegistry registry();
 
+    /// Returns the [LayoutRegistry] this handle was opened with, dispatching full-column layout
+    /// subtree decode. Defaults to [LayoutRegistry#defaults()] unless a custom registry was
+    /// supplied at open time.
+    ///
+    /// @return the registry used to decode layout nodes during scan
+    LayoutRegistry layoutRegistry();
+
     @Override
     void close();
 }

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

@@ -6,6 +6,7 @@
 import io.github.dfa1.vortex.core.io.VortexFormat;
 import io.github.dfa1.vortex.core.fbs.FbsPostscript;
 import io.github.dfa1.vortex.reader.layout.Layout;
+import io.github.dfa1.vortex.reader.layout.LayoutRegistry;
 
 import java.io.IOException;
 import java.lang.foreign.Arena;
@@ -47,11 +48,12 @@ public final class VortexHttpReader implements VortexHandle {
     private final DType dtype;
     private final Layout layout;
     private final ReadRegistry registry;
+    private final LayoutRegistry layoutRegistry;
 
     private VortexHttpReader(
         URI uri, HttpClient client, long fileSize,
         int version, Footer footer, DType dtype, Layout layout,
-        ReadRegistry registry
+        ReadRegistry registry, LayoutRegistry layoutRegistry
     ) {
         this.uri = uri;
         this.client = client;
@@ -62,6 +64,7 @@ private VortexHttpReader(
         this.dtype = dtype;
         this.layout = layout;
         this.registry = registry;
+        this.layoutRegistry = layoutRegistry;
     }
 
     public static VortexHttpReader open(URI uri) throws IOException {
@@ -83,6 +86,20 @@ public static VortexHttpReader open(URI uri, ReadRegistry registry) throws IOExc
     /// @return an open handle to the remote file
     /// @throws IOException if the file cannot be opened or parsed
     public static VortexHttpReader open(URI uri, ReadRegistry registry, HttpClient client) throws IOException {
+        return open(uri, registry, LayoutRegistry.defaults(), client);
+    }
+
+    /// Opens a remote Vortex file with explicit encoding and layout registries and a
+    /// caller-supplied [HttpClient].
+    ///
+    /// @param uri            HTTP(S) URL of the Vortex file
+    /// @param registry       the encoding decode registry
+    /// @param layoutRegistry the layout decode registry (custom layouts register here)
+    /// @param client         HTTP client to use for all Range requests
+    /// @return an open handle to the remote file
+    /// @throws IOException if the file cannot be opened or parsed
+    public static VortexHttpReader open(URI uri, ReadRegistry registry, LayoutRegistry layoutRegistry,
+            HttpClient client) throws IOException {
         // Single suffix Range request — Content-Range response header gives us fileSize.
         // Avoids a separate HEAD round trip.
         TailFetch tf = fetchTail(uri, client);
@@ -133,7 +150,7 @@ public static VortexHttpReader open(URI uri, ReadRegistry registry, HttpClient c
         return new VortexHttpReader(
             uri, client, fileSize, trailer.version(),
             parsed.footer(), parsed.dtype(), parsed.layout(),
-            registry
+            registry, layoutRegistry
         );
     }
 
@@ -312,6 +329,11 @@ public ReadRegistry registry() {
         return registry;
     }
 
+    @Override
+    public LayoutRegistry layoutRegistry() {
+        return layoutRegistry;
+    }
+
     @Override
     public void close() {
         arena.close();

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

@@ -6,6 +6,7 @@
 import io.github.dfa1.vortex.core.error.VortexException;
 import io.github.dfa1.vortex.core.io.VortexFormat;
 import io.github.dfa1.vortex.reader.layout.Layout;
+import io.github.dfa1.vortex.reader.layout.LayoutRegistry;
 
 import java.io.IOException;
 import java.lang.foreign.Arena;
@@ -34,11 +35,12 @@ public final class VortexReader implements VortexHandle {
     private final DType dtype;
     private final Layout layout;
     private final ReadRegistry registry;
+    private final LayoutRegistry layoutRegistry;
 
     private VortexReader(
             Arena arena, MemorySegment fileSegment, long fileSize,
             int version, Footer footer, DType dtype, Layout layout,
-            ReadRegistry registry
+            ReadRegistry registry, LayoutRegistry layoutRegistry
     ) {
         this.arena = arena;
         this.fileSegment = fileSegment;
@@ -48,6 +50,7 @@ private VortexReader(
         this.dtype = dtype;
         this.layout = layout;
         this.registry = registry;
+        this.layoutRegistry = layoutRegistry;
     }
 
     /// Open a Vortex file. Memory-maps the entire file; all subsequent reads
@@ -57,6 +60,19 @@ public static VortexReader open(Path path) throws IOException {
     }
 
     public static VortexReader open(Path path, ReadRegistry registry) throws IOException {
+        return open(path, registry, LayoutRegistry.defaults());
+    }
+
+    /// Opens a Vortex file with explicit encoding and layout registries. Memory-maps the entire
+    /// file; all subsequent reads are zero-copy slices. Call [#close()] when done.
+    ///
+    /// @param path           the file to open
+    /// @param registry       the encoding decode registry
+    /// @param layoutRegistry the layout decode registry (custom layouts register here)
+    /// @return an open handle to the file
+    /// @throws IOException if the file cannot be opened or parsed
+    public static VortexReader open(Path path, ReadRegistry registry, LayoutRegistry layoutRegistry)
+            throws IOException {
         Arena arena = Arena.ofConfined();
         try (var channel = FileChannel.open(path, StandardOpenOption.READ)) {
             long size = channel.size();
@@ -67,15 +83,15 @@ public static VortexReader open(Path path, ReadRegistry registry) throws IOExcep
             // lifetime. try-with-resources closes the file descriptor while all Array
             // buffers remain valid zero-copy slices until arena.close() is called.
             var segment = channel.map(FileChannel.MapMode.READ_ONLY, 0, size, arena);
-            return parse(segment, size, arena, registry);
+            return parse(segment, size, arena, registry, layoutRegistry);
         } catch (Exception e) {
             arena.close();
             throw e;
         }
     }
 
     private static VortexReader parse(
-            MemorySegment seg, long size, Arena arena, ReadRegistry registry
+            MemorySegment seg, long size, Arena arena, ReadRegistry registry, LayoutRegistry layoutRegistry
     ) {
         long bodyBytes = size - VortexFormat.TRAILER_SIZE;
         var trailerSeg = IoBounds.slice(seg, bodyBytes, VortexFormat.TRAILER_SIZE);
@@ -96,7 +112,7 @@ private static VortexReader parse(
         return new VortexReader(
                 arena, seg, size, trailer.version(),
                 parsed.footer(), parsed.dtype(), parsed.layout(),
-                registry
+                registry, layoutRegistry
         );
     }
 
@@ -160,6 +176,11 @@ public ReadRegistry registry() {
         return registry;
     }
 
+    @Override
+    public LayoutRegistry layoutRegistry() {
+        return layoutRegistry;
+    }
+
     /// Returns the number of chunks in this file.
     ///
     /// Equal to the length of [ScanIterator#chunkRowCounts()] from a full scan, and the number

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

@@ -0,0 +1,98 @@
+package io.github.dfa1.vortex.reader.layout;
+
+import io.github.dfa1.vortex.core.error.VortexException;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.EncodingId;
+import io.github.dfa1.vortex.core.model.LayoutId;
+import io.github.dfa1.vortex.core.model.PType;
+import io.github.dfa1.vortex.reader.array.Array;
+import io.github.dfa1.vortex.reader.array.ChunkedBoolArray;
+import io.github.dfa1.vortex.reader.array.ChunkedByteArray;
+import io.github.dfa1.vortex.reader.array.ChunkedDoubleArray;
+import io.github.dfa1.vortex.reader.array.ChunkedFloatArray;
+import io.github.dfa1.vortex.reader.array.ChunkedIntArray;
+import io.github.dfa1.vortex.reader.array.ChunkedLongArray;
+import io.github.dfa1.vortex.reader.array.ChunkedShortArray;
+import io.github.dfa1.vortex.reader.array.VarBinArray;
+
+import java.lang.foreign.ValueLayout;
+import java.util.ArrayList;
+import java.util.List;
+
+/// Built-in decoder for the `vortex.chunked` layout — a sequence of flat leaves decoded into the
+/// zero-copy `ChunkedXxxArray` shape (ADR 0012). Extracted verbatim from `ScanIterator`'s
+/// `collectFlats` + `decodeChunkedLayout`.
+final class ChunkedLayoutDecoder implements LayoutDecoder {
+
+    @Override
+    public LayoutId layoutId() {
+        return LayoutId.CHUNKED;
+    }
+
+    @Override
+    public Array decode(LayoutDecodeContext ctx, Layout layout, DType dtype) {
+        var flats = new ArrayList<Layout>();
+        collectFlats(layout, flats);
+        return decodeChunkedLayout(ctx, flats, dtype, layout.rowCount());
+    }
+
+    /// Flattens a layout subtree into its ordered flat (and dict) leaves. A private copy of
+    /// `ScanIterator.collectFlats`: the scan keeps its own for chunk-shape planning, while the
+    /// chunked decoder needs the same flattening to build the leaf array list.
+    ///
+    /// @param layout the layout subtree to flatten
+    /// @param out    accumulator for the flat leaves in scan order
+    private static void collectFlats(Layout layout, List<Layout> out) {
+        if (layout.isFlat()) {
+            out.add(layout);
+        } else if (layout.isDict()) {
+            // Dict layout is a leaf chunk — decoded as a unit (values + codes).
+            out.add(layout);
+        } else if (layout.isZoned()) {
+            // vortex.stats wraps one child (the data layout) — pass through for data
+            if (!layout.children().isEmpty()) {
+                collectFlats(layout.children().getFirst(), out);
+            }
+        } else if (layout.isChunked()) {
+            // metadata[0] == 1 means children[0] is the per-chunk stats layout; skip it
+            int start = (layout.metadata() != null
+                                 && layout.metadata().byteSize() > 0
+                                 && layout.metadata().get(ValueLayout.JAVA_BYTE, 0) == 1) ? 1 : 0;
+            for (int i = start; i < layout.children().size(); i++) {
+                collectFlats(layout.children().get(i), out);
+            }
+        }
+    }
+
+    private static Array decodeChunkedLayout(LayoutDecodeContext ctx, List<Layout> flats, DType dtype,
+            long totalRows) {
+        if (flats.isEmpty()) {
+            throw new VortexException(EncodingId.VORTEX_CHUNKED, "no flat children");
+        }
+        if (flats.size() == 1) {
+            return FlatLayoutDecoder.decodeFlat(ctx, flats.getFirst(), dtype);
+        }
+        // ADR 0012: every primitive ptype gets the zero-copy ChunkedXxxArray shape.
+        // The concat path is gone.
+        var chunkArrays = new ArrayList<Array>(flats.size());
+        for (Layout flat : flats) {
+            chunkArrays.add(FlatLayoutDecoder.decodeFlat(ctx, flat, dtype));
+        }
+        if (dtype instanceof DType.Bool) {
+            return ChunkedBoolArray.of(dtype, totalRows, chunkArrays);
+        }
+        if (dtype instanceof DType.Utf8 || dtype instanceof DType.Binary) {
+            return VarBinArray.ChunkedMode.of(dtype, totalRows, chunkArrays);
+        }
+        PType ptype = ((DType.Primitive) dtype).ptype();
+        return switch (ptype) {
+            case I64, U64 -> ChunkedLongArray.of(dtype, totalRows, chunkArrays);
+            case I32, U32 -> ChunkedIntArray.of(dtype, totalRows, chunkArrays);
+            case F64 -> ChunkedDoubleArray.of(dtype, totalRows, chunkArrays);
+            case F32 -> ChunkedFloatArray.of(dtype, totalRows, chunkArrays);
+            case I16, U16 -> ChunkedShortArray.of(dtype, totalRows, chunkArrays);
+            case I8, U8 -> ChunkedByteArray.of(dtype, totalRows, chunkArrays);
+            default -> throw new VortexException("unsupported ptype for chunked layout: " + ptype);
+        };
+    }
+}