feat(chunked): implement vortex.chunked encode via ChunkedData

Encodes chunk offsets (U64) as first child, each chunk via fallback
encoding. ChunkedData carries chunk data + explicit row counts.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dfa1

TODO.md

@@ -93,7 +93,7 @@
 | `vortex.fsst`                | `FsstEncoding`             | ✅       | ❌ stub  | —         | Utf8, Binary |
 | `vortex.varbinview`          | `VarBinViewEncoding`       | ✅       | ❌ stub  | —         | Utf8, Binary |
 | `vortex.pco`                 | `PcoEncoding` (stub)       | ❌       | ❌       | very hard | ANS + bin tokenization not ported; unblocks `pco.vortex`, tpch/clickbench fixtures |
-| `vortex.chunked`             | `ChunkedEncoding`          | ✅       | ❌ stub  | medium    | decode: primitive + struct concat; encode stub |
+| `vortex.chunked`             | `ChunkedEncoding`          | ✅       | ✅       | medium    | decode: primitive + struct concat; encode via ChunkedData |
 | `fastlanes.rle`              | —                          | ❌       | ❌       | medium    | unblocks `rle.vortex` |
 | `vortex.alprd`               | —                          | ❌       | ❌       | medium    | unblocks `alprd.vortex` |
 | `vortex.decimal`             | `DecimalEncoding`          | ✅       | ✅       | —         | Decimal |

core/src/main/java/io/github/dfa1/vortex/encoding/ChunkedData.java

@@ -0,0 +1,18 @@
+package io.github.dfa1.vortex.encoding;
+
+import java.util.List;
+
+/// Input data for encoding a chunked array.
+/// Each element of {@code chunks} is the raw data for one chunk, in the same format
+/// the inner encoding expects (e.g. {@code long[]} for I64, {@link StructData} for Struct).
+/// {@code chunkLengths[i]} is the row count of {@code chunks.get(i)}.
+public record ChunkedData(List<Object> chunks, long[] chunkLengths) {
+    public ChunkedData {
+        if (chunks.size() != chunkLengths.length) {
+            throw new IllegalArgumentException(
+                    "chunks.size() %d != chunkLengths.length %d".formatted(chunks.size(), chunkLengths.length));
+        }
+        chunks = List.copyOf(chunks);
+        chunkLengths = chunkLengths.clone();
+    }
+}

core/src/main/java/io/github/dfa1/vortex/encoding/ChunkedEncoding.java

@@ -16,6 +16,7 @@
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.SegmentAllocator;
 import java.lang.foreign.ValueLayout;
+import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.ArrayList;
 import java.util.List;
@@ -43,14 +44,70 @@ public boolean accepts(DType dtype) {
 
 	@Override
 	public EncodeResult encode(DType dtype, Object data) {
-		throw new UnsupportedOperationException("encode not yet implemented for " + encodingId());
+		return Encoder.encode(dtype, (ChunkedData) data);
 	}
 
 	@Override
 	public Array decode(DecodeContext ctx) {
 		return Decoder.decode(ctx);
 	}
 
+	private static final class Encoder {
+
+		private static final List<Encoding> FALLBACK = List.of(
+				new PrimitiveEncoding(), new VarBinEncoding(), new BoolEncoding(),
+				new NullEncoding(), new ByteBoolEncoding(), new StructEncoding());
+
+		static EncodeResult encode(DType dtype, ChunkedData data) {
+			List<Object> chunks = data.chunks();
+			long[] chunkLengths = data.chunkLengths();
+			int nchunks = chunks.size();
+			if (nchunks == 0) {
+				throw new VortexException(EncodingId.VORTEX_CHUNKED, "at least one chunk required");
+			}
+
+			// Build cumulative offsets: [0, len0, len0+len1, ...]
+			long[] offsets = new long[nchunks + 1];
+			offsets[0] = 0;
+			for (int i = 0; i < nchunks; i++) {
+				offsets[i + 1] = offsets[i] + chunkLengths[i];
+			}
+
+			// Encode offsets child (U64 primitive)
+			DType u64 = new DType.Primitive(PType.U64, false);
+			EncodeResult offsetsResult = new PrimitiveEncoding().encode(u64, offsets);
+
+			List<MemorySegment> allBuffers = new ArrayList<>(offsetsResult.buffers());
+			EncodeNode[] children = new EncodeNode[nchunks + 1];
+			children[0] = offsetsResult.rootNode();
+
+			// Encode each chunk
+			Encoding inner = findEncoding(dtype);
+			for (int i = 0; i < nchunks; i++) {
+				EncodeResult chunkResult = inner.encode(dtype, chunks.get(i));
+				int bufOffset = allBuffers.size();
+				children[i + 1] = EncodeNode.remapBufferIndices(chunkResult.rootNode(), bufOffset);
+				allBuffers.addAll(chunkResult.buffers());
+			}
+
+			EncodeNode root = new EncodeNode(
+					EncodingId.VORTEX_CHUNKED,
+					ByteBuffer.wrap(new byte[0]),
+					children,
+					new int[]{});
+			return new EncodeResult(root, List.copyOf(allBuffers), null, null);
+		}
+
+		private static Encoding findEncoding(DType dtype) {
+			for (Encoding enc : FALLBACK) {
+				if (enc.accepts(dtype)) {
+					return enc;
+				}
+			}
+			throw new UnsupportedOperationException("no fallback encoding for dtype: " + dtype);
+		}
+	}
+
 	private static final class Decoder {
 
 		private static final ValueLayout.OfLong LE_LONG =

core/src/test/java/io/github/dfa1/vortex/encoding/ChunkedEncodingTest.java

@@ -11,6 +11,7 @@
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;
 import java.nio.ByteOrder;
+import java.util.List;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -20,6 +21,64 @@ class ChunkedEncodingTest {
 	private static final ValueLayout.OfLong LE_LONG =
 			ValueLayout.JAVA_LONG_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN);
 
+	@Nested
+	class Encode {
+
+		@Test
+		void roundTrip_twoChunks_i64_preservesValues() {
+			// Given
+			long[] chunk0 = {10L, 20L, 30L};
+			long[] chunk1 = {40L, 50L};
+			DType i64 = new DType.Primitive(PType.I64, false);
+			var sut = new ChunkedEncoding();
+			EncodingRegistry registry = EncodingRegistry.empty();
+			registry.register(sut);
+			registry.register(new PrimitiveEncoding());
+			ChunkedData data = new ChunkedData(List.of(chunk0, chunk1), new long[]{3, 2});
+
+			// When
+			EncodeResult encoded = sut.encode(i64, data);
+			DecodeContext ctx = EncodeTestHelper.toDecodeContext(encoded, 5L, i64, registry);
+			Array result = sut.decode(ctx);
+
+			// Then
+			assertThat(result.length()).isEqualTo(5);
+			assertThat(result.buffer(0).get(LE_LONG, 0L)).isEqualTo(10L);
+			assertThat(result.buffer(0).get(LE_LONG, 8L)).isEqualTo(20L);
+			assertThat(result.buffer(0).get(LE_LONG, 16L)).isEqualTo(30L);
+			assertThat(result.buffer(0).get(LE_LONG, 24L)).isEqualTo(40L);
+			assertThat(result.buffer(0).get(LE_LONG, 32L)).isEqualTo(50L);
+		}
+
+		@Test
+		void encodeNode_hasNoDirectBuffers_offsetsAsFirstChild() {
+			// Given
+			long[] chunk0 = {1L, 2L};
+			DType i64 = new DType.Primitive(PType.I64, false);
+			var sut = new ChunkedEncoding();
+			ChunkedData data = new ChunkedData(List.of(chunk0), new long[]{2});
+
+			// When
+			EncodeResult result = sut.encode(i64, data);
+
+			// Then
+			assertThat(result.rootNode().bufferIndices()).isEmpty();
+			assertThat(result.rootNode().children()).hasSize(2); // offsets + 1 chunk
+			assertThat(result.buffers()).hasSize(2); // offsets buf + chunk buf
+		}
+
+		@Test
+		void mismatchedLengths_throws() {
+			// Given
+			var sut = new ChunkedEncoding();
+			DType i64 = new DType.Primitive(PType.I64, false);
+
+			// When / Then
+			assertThatThrownBy(() -> new ChunkedData(List.of(new long[]{1L}), new long[]{1, 2}))
+					.isInstanceOf(IllegalArgumentException.class);
+		}
+	}
+
 	@Nested
 	class Decode {