feat(writer): add VarBin encode + Utf8 dtype serialization; add symbol column to CSV benchmark

- VarBinEncoding.encode() handles String[] → bytes buf + I64 offsets
- VortexWriter.DEFAULT_CODECS includes VarBinEncoding
- VortexWriter.serializeDType supports DType.Utf8
- VortexWriter.arrayLength handles String[]
- VortexVsCsvFileSizeTest: 30 real ticker symbols (AAPL, MSFT...) as a Utf8 column

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

Author: dfa1

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

@@ -9,8 +9,13 @@
 import io.github.dfa1.vortex.core.array.VarBinArray;
 import io.github.dfa1.vortex.core.VortexException;
 
+import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
 
 /// Decoder for {@code vortex.varbin} — variable-length binary / UTF-8 string arrays.
 ///
@@ -33,9 +38,45 @@ public EncodingId encodingId() {
 		return EncodingId.VORTEX_VARBIN;
 	}
 
+	@Override
+	public boolean accepts(DType dtype) {
+		return dtype instanceof DType.Utf8 || dtype instanceof DType.Binary;
+	}
+
 	@Override
 	public EncodeResult encode(DType dtype, Object data) {
-		throw new UnsupportedOperationException("encode not supported by " + encodingId());
+		String[] strings = (String[]) data;
+		int n = strings.length;
+
+		byte[][] byteArrays = new byte[n][];
+		int totalBytes = 0;
+		for (int i = 0; i < n; i++) {
+			byteArrays[i] = strings[i].getBytes(StandardCharsets.UTF_8);
+			totalBytes += byteArrays[i].length;
+		}
+
+		Arena arena = Arena.ofAuto();
+		MemorySegment bytesBuf = arena.allocate(totalBytes > 0 ? totalBytes : 1);
+		MemorySegment offsetsBuf = arena.allocate((long) (n + 1) * Long.BYTES, Long.BYTES);
+		ValueLayout.OfLong leI64 = ValueLayout.JAVA_LONG_UNALIGNED.withOrder(ByteOrder.LITTLE_ENDIAN);
+
+		long pos = 0;
+		offsetsBuf.setAtIndex(leI64, 0, 0L);
+		for (int i = 0; i < n; i++) {
+			MemorySegment.copy(MemorySegment.ofArray(byteArrays[i]), 0, bytesBuf, pos, byteArrays[i].length);
+			pos += byteArrays[i].length;
+			offsetsBuf.setAtIndex(leI64, i + 1, pos);
+		}
+
+		byte[] metaBytes = EncodingProtos.VarBinMetadata.newBuilder()
+				.setOffsetsPtype(dev.vortex.proto.DTypeProtos.PType.forNumber(PType.I64.ordinal()))
+				.build()
+				.toByteArray();
+
+		EncodeNode offsetsNode = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 1);
+		EncodeNode root = new EncodeNode(encodingId(), ByteBuffer.wrap(metaBytes),
+				new EncodeNode[]{offsetsNode}, new int[]{0});
+		return new EncodeResult(root, List.of(bytesBuf, offsetsBuf), null, null);
 	}
 
 	@Override

integration/src/test/java/io/github/dfa1/vortex/integration/VortexVsCsvFileSizeTest.java

@@ -29,9 +29,16 @@ class VortexVsCsvFileSizeTest {
 	private static final int TOTAL_ROWS = 100_000;
 	private static final int BATCH_SIZE = 50_000;
 
+	private static final String[] SYMBOLS = {
+			"AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "META", "TSLA", "BRK.B", "JPM", "V",
+			"UNH", "XOM", "LLY", "JNJ", "MA", "PG", "HD", "MRK", "AVGO", "CVX",
+			"PEP", "ABBV", "KO", "COST", "WMT", "MCD", "CSCO", "ACN", "BAC", "CRM"
+	};
+
 	private static final DType.Struct SCHEMA = new DType.Struct(
-			List.of("date", "open", "high", "low", "close", "volume"),
+			List.of("symbol", "date", "open", "high", "low", "close", "volume"),
 			List.of(
+					new DType.Utf8(false),
 					new DType.Primitive(PType.I32, false),
 					new DType.Primitive(PType.F64, false),
 					new DType.Primitive(PType.F64, false),
@@ -42,7 +49,7 @@ class VortexVsCsvFileSizeTest {
 			false
 	);
 
-	private record OhlcBatch(int[] dates, double[] open, double[] high,
+	private record OhlcBatch(String[] symbols, int[] dates, double[] open, double[] high,
 	                         double[] low, double[] close, long[] volume) {}
 
 	private static List<OhlcBatch> generateBatches() {
@@ -55,6 +62,7 @@ private static List<OhlcBatch> generateBatches() {
 
 		while (rowsLeft > 0) {
 			int n = Math.min(rowsLeft, BATCH_SIZE);
+			String[] symbols = new String[n];
 			int[] dates = new int[n];
 			double[] open = new double[n], high = new double[n], low = new double[n], close = new double[n];
 			long[] volume = new long[n];
@@ -65,6 +73,7 @@ private static List<OhlcBatch> generateBatches() {
 				double o = Math.round(px * (1 + ret * 0.3) * 100.0) / 100.0;
 				double c = Math.round(px * (1 + ret) * 100.0) / 100.0;
 				double spread = Math.abs(px * rng.nextDouble() * 0.03);
+				symbols[i] = SYMBOLS[ticker];
 				dates[i]  = startDay + (rowsDone + i) / 30;
 				open[i]   = o;
 				high[i]   = Math.round((Math.max(o, c) + spread) * 100.0) / 100.0;
@@ -73,7 +82,7 @@ private static List<OhlcBatch> generateBatches() {
 				volume[i] = Math.max(100_000L, Math.round(1_000_000 + rng.nextGaussian() * 200_000));
 				prices[ticker] = c;
 			}
-			batches.add(new OhlcBatch(dates, open, high, low, close, volume));
+			batches.add(new OhlcBatch(symbols, dates, open, high, low, close, volume));
 			rowsLeft -= n;
 			rowsDone += n;
 		}
@@ -92,18 +101,19 @@ void vortexSmallerThanCsv(@TempDir Path tmp) throws IOException {
 		     VortexWriter writer = VortexWriter.create(ch, SCHEMA, WriteOptions.defaults())) {
 			for (OhlcBatch b : batches) {
 				writer.writeChunk(Map.of(
-						"date", b.dates(), "open", b.open(), "high", b.high(),
+						"symbol", b.symbols(), "date", b.dates(), "open", b.open(), "high", b.high(),
 						"low", b.low(), "close", b.close(), "volume", b.volume()
 				));
 			}
 		}
 
 		// When — write CSV
 		try (BufferedWriter csv = Files.newBufferedWriter(csvFile)) {
-			csv.write("date,open,high,low,close,volume\n");
+			csv.write("symbol,date,open,high,low,close,volume\n");
 			for (OhlcBatch b : batches) {
 				for (int i = 0; i < b.dates().length; i++) {
-					csv.write(LocalDate.ofEpochDay(b.dates()[i]) + "," + b.open()[i] + "," + b.high()[i] + ","
+					csv.write(b.symbols()[i] + "," + LocalDate.ofEpochDay(b.dates()[i]) + ","
+							+ b.open()[i] + "," + b.high()[i] + ","
 							+ b.low()[i] + "," + b.close()[i] + "," + b.volume()[i] + "\n");
 				}
 			}

writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java

@@ -5,6 +5,7 @@
 import io.github.dfa1.vortex.encoding.AlpEncoding;
 import io.github.dfa1.vortex.encoding.BoolEncoding;
 import io.github.dfa1.vortex.encoding.Encoding;
+import io.github.dfa1.vortex.encoding.VarBinEncoding;
 import io.github.dfa1.vortex.encoding.EncodeNode;
 import io.github.dfa1.vortex.encoding.EncodeResult;
 import io.github.dfa1.vortex.encoding.EncodingId;
@@ -53,7 +54,8 @@ public final class VortexWriter implements Closeable {
 	private static final ByteBuffer MAGIC = ByteBuffer.wrap(new byte[]{'V', 'T', 'X', 'F'})
 			.asReadOnlyBuffer();
 
-	private static final List<Encoding> DEFAULT_CODECS = List.of(new AlpEncoding(), new PrimitiveEncoding(), new BoolEncoding());
+	private static final List<Encoding> DEFAULT_CODECS = List.of(
+			new AlpEncoding(), new VarBinEncoding(), new PrimitiveEncoding(), new BoolEncoding());
 
 	private final WritableByteChannel channel;
 	private final DType.Struct schema;
@@ -97,6 +99,7 @@ private static long arrayLength(Object data) {
 			case float[] a -> a.length;
 			case double[] a -> a.length;
 			case boolean[] a -> a.length;
+			case String[] a -> a.length;
 			default -> throw new UnsupportedOperationException(
 					"unsupported data type: " + data.getClass());
 		};
@@ -141,6 +144,10 @@ private static int serializeDType(FlatBufferBuilder fbb, DType dtype) {
 						fbb, namesVec, dtypesVec, s.nullable());
 				yield io.github.dfa1.vortex.fbs.DType.createDType(fbb, Type.Struct_, inner);
 			}
+			case DType.Utf8 u -> {
+				int inner = io.github.dfa1.vortex.fbs.Utf8.createUtf8(fbb, u.nullable());
+				yield io.github.dfa1.vortex.fbs.DType.createDType(fbb, Type.Utf8, inner);
+			}
 			default -> throw new UnsupportedOperationException("unsupported DType: " + dtype);
 		};
 	}