feat: support vortex.dict layout and vortex.fsst codec

Adds full read support for two encodings Rust chooses for high-cardinality
string columns (e.g. 30 Nasdaq tickers over 10M rows):

- FsstCodec: decompresses Fast Static Symbol Table–encoded strings;
  3-buffer wire format (symbol table, lengths, compressed bytes) + 2 children
  (uncompressed lengths, codes offsets)
- DictLayoutMetadata proto: layout-level dict (codes_ptype at tag 1)
- decodeDictLayout: decodes values (FSST) + codes (bitpacked, possibly
  chunked), then expands via expandDictStrings
- decodeConcatPrimitive: handles chunked codes layout by decoding each flat
  and concatenating into one buffer before dict expansion
- RustVsJavaReadBenchmark: 30 real Nasdaq tickers with per-ticker price state

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

Author: dfa1

core/src/main/java/io/github/dfa1/vortex/core/Layout.java

@@ -37,4 +37,8 @@ public boolean isStruct() {
 	public boolean isZoned() {
 		return ZONED.equals(encodingId);
 	}
+
+	public boolean isDict() {
+		return DICT.equals(encodingId);
+	}
 }

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

@@ -17,6 +17,7 @@ public enum CodecId {
 	VORTEX_ALP("vortex.alp"),
 	VORTEX_BITPACKED("vortex.bitpacked"),
 	VORTEX_VARBIN("vortex.varbin"),
+	VORTEX_FSST("vortex.fsst"),
 	VORTEX_NULL("vortex.null"),
 
 	// Layout encoding IDs included so parser/registry can represent them safely

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

@@ -0,0 +1,142 @@
+package io.github.dfa1.vortex.encoding;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import dev.vortex.proto.EncodingProtos;
+import io.github.dfa1.vortex.core.Array;
+import io.github.dfa1.vortex.core.ArrayStats;
+import io.github.dfa1.vortex.core.DType;
+import io.github.dfa1.vortex.core.PType;
+
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+/// Decoder for {@code vortex.fsst} — Fast Static Symbol Table string compression.
+///
+/// <p>Wire format (3-buffer current format):
+/// <ul>
+///   <li>Buffer 0: symbol table — up to 256 entries, each 8 bytes LE (Symbol = u64)</li>
+///   <li>Buffer 1: symbol_lengths — 1 byte per symbol, actual byte count in the symbol</li>
+///   <li>Buffer 2: compressed code bytes for all strings concatenated</li>
+///   <li>Child 0: uncompressed_lengths — primitive array (I32/I64) of original string lengths</li>
+///   <li>Child 1: codes_offsets — primitive array (I32/I64) of offsets into buf[2], length = n+1</li>
+/// </ul>
+///
+/// <p>Output: varbin-compatible {@code Array} (buffer[0] = uncompressed bytes, child[0] = I32 offsets).
+public final class FsstCodec implements Codec {
+
+    private static final int ESCAPE = 0xFF;
+
+    private static final ValueLayout.OfLong   LE_LONG  = ValueLayout.JAVA_LONG_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN);
+    private static final ValueLayout.OfInt    LE_INT   = ValueLayout.JAVA_INT_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN);
+
+    @Override
+    public CodecId encodingId() {
+        return CodecId.VORTEX_FSST;
+    }
+
+    @Override
+    public EncodeResult encode(DType dtype, Object data) {
+        throw new UnsupportedOperationException("encode not supported by " + encodingId());
+    }
+
+    @Override
+    public Array decode(DecodeContext ctx) {
+        ByteBuffer rawMeta = ctx.metadata();
+        if (rawMeta == null) {
+            throw new IllegalStateException("vortex.fsst: missing metadata");
+        }
+        EncodingProtos.FSSTMetadata meta;
+        try {
+            meta = EncodingProtos.FSSTMetadata.parseFrom(rawMeta.duplicate());
+        } catch (InvalidProtocolBufferException e) {
+            throw new IllegalStateException("vortex.fsst: invalid metadata", e);
+        }
+
+        PType uncompLenPType = PType.values()[meta.getUncompressedLengthsPtype().getNumber()];
+        PType codesOffPType  = PType.values()[meta.getCodesOffsetsPtype().getNumber()];
+
+        long n = ctx.rowCount();
+
+        MemorySegment symbolsBuf      = ctx.buffer(0); // 8 bytes per symbol (LE u64)
+        MemorySegment symbolLensBuf   = ctx.buffer(1); // 1 byte per symbol
+        MemorySegment compressedBytes = ctx.buffer(2); // FSST-compressed heap
+
+        // Decode child arrays
+        Array uncompLens = decodeChild(ctx, 0, uncompLenPType, n);
+        Array codesOffsets = decodeChild(ctx, 1, codesOffPType, n + 1);
+        MemorySegment uncompLensSeg   = uncompLens.buffer(0);
+        MemorySegment codesOffsetsSeg = codesOffsets.buffer(0);
+
+        // Total uncompressed size (sum of uncompressed_lengths)
+        long totalUncompressed = 0L;
+        for (long i = 0; i < n; i++) {
+            totalUncompressed += readUnsigned(uncompLensSeg, i, uncompLenPType);
+        }
+
+        MemorySegment outBytes   = ctx.arena().allocate(totalUncompressed);
+        MemorySegment outOffsets = ctx.arena().allocate((n + 1) * 4L, 4);
+        outOffsets.setAtIndex(LE_INT, 0, 0);
+
+        long outPos = 0L;
+        for (long i = 0; i < n; i++) {
+            long cStart = readUnsigned(codesOffsetsSeg, i, codesOffPType);
+            long cEnd   = readUnsigned(codesOffsetsSeg, i + 1, codesOffPType);
+            outPos = decompressString(compressedBytes, symbolsBuf, symbolLensBuf,
+                    cStart, cEnd, outBytes, outPos);
+            outOffsets.setAtIndex(LE_INT, i + 1, (int) outPos);
+        }
+
+        Array offsets = new Array(new DType.Primitive(PType.I32, false), n + 1,
+                new MemorySegment[]{outOffsets.asReadOnly()}, Array.NO_CHILDREN, ArrayStats.empty());
+        return new Array(ctx.dtype(), n,
+                new MemorySegment[]{outBytes.asReadOnly()},
+                new Array[]{offsets},
+                ArrayStats.empty());
+    }
+
+    // ── Helpers ───────────────────────────────────────────────────────────────
+
+    private Array decodeChild(DecodeContext parent, int idx, PType ptype, long rowCount) {
+        ArrayNode childNode = parent.node().children()[idx];
+        DType dtype = new DType.Primitive(ptype, false);
+        DecodeContext childCtx = new DecodeContext(
+                childNode, dtype, rowCount,
+                parent.segmentBuffers(), parent.registry(), parent.arena());
+        return parent.registry().decode(childCtx);
+    }
+
+    private static long decompressString(
+            MemorySegment compressed, MemorySegment symbols, MemorySegment symLens,
+            long start, long end, MemorySegment out, long outPos
+    ) {
+        for (long j = start; j < end; j++) {
+            int b = Byte.toUnsignedInt(compressed.get(ValueLayout.JAVA_BYTE, j));
+            if (b == ESCAPE) {
+                // next byte is literal
+                out.set(ValueLayout.JAVA_BYTE, outPos++, compressed.get(ValueLayout.JAVA_BYTE, ++j));
+            } else {
+                int symLen = Byte.toUnsignedInt(symLens.get(ValueLayout.JAVA_BYTE, b));
+                // symbols[b] is 8 bytes LE; emit first symLen bytes
+                long sym = symbols.getAtIndex(LE_LONG, b);
+                for (int k = 0; k < symLen; k++) {
+                    out.set(ValueLayout.JAVA_BYTE, outPos++, (byte) (sym >>> (k * 8)));
+                }
+            }
+        }
+        return outPos;
+    }
+
+    private static long readUnsigned(MemorySegment seg, long idx, PType ptype) {
+        return switch (ptype) {
+            case U8  -> Byte.toUnsignedLong(seg.get(ValueLayout.JAVA_BYTE, idx));
+            case U16 -> Short.toUnsignedLong(seg.get(
+                    ValueLayout.JAVA_SHORT_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN), idx * 2));
+            case U32 -> Integer.toUnsignedLong(seg.getAtIndex(LE_INT, idx));
+            case I32 -> seg.getAtIndex(LE_INT, idx);
+            case I64, U64 -> seg.getAtIndex(LE_LONG, idx);
+            default  -> throw new IllegalArgumentException("vortex.fsst: unsupported ptype " + ptype);
+        };
+    }
+}

core/src/main/proto/encodings.proto

@@ -54,3 +54,15 @@ message SequenceMetadata {
 message VarBinMetadata {
   vortex.dtype.PType offsets_ptype = 1;
 }
+
+message FSSTMetadata {
+  vortex.dtype.PType uncompressed_lengths_ptype = 1;
+  vortex.dtype.PType codes_offsets_ptype = 2;
+}
+
+// Layout-level dict metadata (different field ordering from codec DictMetadata).
+message DictLayoutMetadata {
+  vortex.dtype.PType codes_ptype = 1;
+  optional bool is_nullable_codes = 2;
+  optional bool all_values_referenced = 3;
+}

core/src/main/resources/META-INF/services/io.github.dfa1.vortex.encoding.Codec

@@ -10,3 +10,4 @@ io.github.dfa1.vortex.encoding.AlpCodec
 io.github.dfa1.vortex.encoding.RunEndCodec
 io.github.dfa1.vortex.encoding.ConstantCodec
 io.github.dfa1.vortex.encoding.VarBinCodec
+io.github.dfa1.vortex.encoding.FsstCodec

performance/src/main/java/io/github/dfa1/vortex/performance/RustVsJavaReadBenchmark.java

@@ -42,12 +42,12 @@
 
 import java.io.IOException;
 import java.lang.foreign.MemorySegment;
+import java.util.Arrays;
 import java.lang.foreign.ValueLayout;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.time.LocalDate;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Random;
@@ -276,11 +276,25 @@ public long javaReadSymbol() throws IOException {
 		return sum;
 	}
 
+	// Real Nasdaq tickers — mix of lengths (1–5 chars) for realistic varbin distribution.
+	private static final String[] NASDAQ_TICKERS = {
+			"AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "TSLA", "AVGO", "COST", "NFLX",
+			"AMD", "ADBE", "QCOM", "PEP", "CSCO", "TXN", "INTC", "CMCSA", "INTU", "AMGN",
+			"HON", "AMAT", "MU", "LRCX", "KLAC", "MRVL", "PANW", "SNPS", "CDNS", "REGN"
+	};
+	private static final byte[][] TICKER_BYTES;
+
+	static {
+		TICKER_BYTES = new byte[NASDAQ_TICKERS.length][];
+		for (int i = 0; i < NASDAQ_TICKERS.length; i++) {
+			TICKER_BYTES[i] = NASDAQ_TICKERS[i].getBytes(StandardCharsets.UTF_8);
+		}
+	}
+
 	private void writeJni(Path path) throws IOException {
 		String uri = path.toAbsolutePath().toUri().toString();
 		try (dev.vortex.api.VortexWriter writer = dev.vortex.api.VortexWriter.create(
 				SESSION, uri, JNI_SCHEMA, new HashMap<>(), allocator)) {
-			var batch = new ArrayList<double[]>(BATCH_SIZE);
 			// reuse arrays — filled per-batch
 			int[] epochDays = new int[BATCH_SIZE];
 			byte[][] symbols = new byte[BATCH_SIZE][];
@@ -290,29 +304,35 @@ private void writeJni(Path path) throws IOException {
 			double[] close = new double[BATCH_SIZE];
 			long[] volume = new long[BATCH_SIZE];
 
+			// Per-ticker price state so each symbol has independent price evolution.
+			double[] prices = new double[NASDAQ_TICKERS.length];
+			Arrays.fill(prices, 100.0);
+
 			var rng = new Random(42L);
-			double px = 100.0;
 			int day = (int) LocalDate.of(2020, 1, 2).toEpochDay();
 			int rowsLeft = TOTAL_ROWS;
 
 			while (rowsLeft > 0) {
 				int n = Math.min(rowsLeft, BATCH_SIZE);
 				for (int i = 0; i < n; i++) {
+					int ticker = i % NASDAQ_TICKERS.length;
+					double px = prices[ticker];
 					double ret = rng.nextGaussian() * 0.02;
 					double o = round(px * (1 + ret * 0.3));
 					double c = round(px * (1 + ret));
 					double rng2 = Math.abs(px * rng.nextDouble() * 0.03);
 					double h = round(Math.max(o, c) + rng2);
 					double l = round(Math.min(o, c) - rng2);
-					epochDays[i] = day++;
-					symbols[i] = "ACME".getBytes(StandardCharsets.UTF_8);
+					epochDays[i] = day + i / NASDAQ_TICKERS.length;
+					symbols[i] = TICKER_BYTES[ticker];
 					open[i] = o;
 					high[i] = h;
 					low[i] = l;
 					close[i] = c;
 					volume[i] = Math.max(100_000L, Math.round(1_000_000 + rng.nextGaussian() * 200_000));
-					px = c;
+					prices[ticker] = c;
 				}
+				day += n / NASDAQ_TICKERS.length;
 				flushJni(writer, epochDays, symbols, open, high, low, close, volume, n);
 				rowsLeft -= n;
 			}