fix(writer): give Fsst/VarBinView/Zstd a byte-safe Binary path

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- `FsstEncodingEncoder`, `VarBinViewEncodingEncoder`, and `ZstdEncodingEncoder` now handle `DType.Binary` byte-for-byte instead of casting straight to `String[]` (a `ClassCastException` if the cascade competition picked one of them for an embedded blob column, e.g. Raincloud's `waxal-dagbani-asr-test` `audio.bytes`); `DType.Binary` also joins `DType.Utf8` in the cascade's real sample-and-measure competition instead of a first-match dispatch, so it gets the same Dict/FSST/VarBinView/Zstd contest Utf8 already had. A non-nullable `DType.Binary` column could not be written at all before this fix (`VortexWriter`'s row-count validation had no `byte[][]` case). ([#352](https://github.com/dfa1/vortex-java/issues/352))
+
 ### Changed
 
 - `dev.vortex:vortex-jni` 0.84.0 → 0.85.0; vortex-jni's writer no longer emits a per-zone `SUM` in the `vortex.zoned` stats table (upstream: a zone sum prunes nothing and its null-on-empty semantics were unsettled), so `ZoneReducer#sum` now falls back to a full scan for Rust-written files instead of pushing the reduction down. ([#360](https://github.com/dfa1/vortex-java/pull/360))

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

@@ -352,6 +352,7 @@ private static long arrayLength(Object data) {
             case double[] a -> a.length;
             case boolean[] a -> a.length;
             case String[] a -> a.length;
+            case byte[][] a -> a.length;
             // A struct column's row count is its fields' row count (all fields share length,
             // enforced by StructEncodingEncoder); an empty struct carries no rows.
             case StructData d -> d.fieldArrays().isEmpty() ? 0L : arrayLength(d.fieldArrays().getFirst());

writer/src/main/java/io/github/dfa1/vortex/writer/encode/CascadingCompressor.java

@@ -38,6 +38,7 @@ private static int dataLength(Object data) {
             case float[] a -> a.length;
             case double[] a -> a.length;
             case String[] a -> a.length;
+            case byte[][] a -> a.length;
             default -> throw new IllegalArgumentException("unsupported data type: " + data.getClass());
         };
     }
@@ -97,6 +98,12 @@ case StructData(var fieldArrays) -> {
                         System.arraycopy(a, srcOff, out, dstOff, len));
                 yield out;
             }
+            case byte[][] a -> {
+                byte[][] out = new byte[sampleSize][];
+                forEachStride(a.length, sampleSize, seed, (srcOff, dstOff, len) ->
+                        System.arraycopy(a, srcOff, out, dstOff, len));
+                yield out;
+            }
             default -> throw new IllegalArgumentException("unsupported data type: " + data.getClass());
         };
     }
@@ -160,24 +167,25 @@ private EncodeResult encodeWithCtx(DType dtype, Object data, EncodeContext ctx)
             return encodeStruct(structDtype, (StructData) data, ctx);
         }
 
-        // Utf8: same sample-and-measure competition as Primitive below (Dict, FSST, VarBin,
-        // Zstd all genuinely compete on measured size) rather than the extension-type
-        // first-match dispatch. No stats are computed — DictEncodingEncoder.expectedRatio()
-        // already defers to this path for Utf8 rather than consuming them — and there is no
-        // cheap analytic "no compression" baseline the way primitiveBytes is for fixed-width
-        // types, so the competition simply keeps whichever accepting encoder measures
-        // smallest (VarBinEncodingEncoder unconditionally accepts Utf8, so a winner always
-        // exists in practice).
-        if (dtype instanceof DType.Utf8) {
+        // Utf8/Binary: same sample-and-measure competition as Primitive below (Dict/VarBin
+        // compete on Utf8 too; FSST, VarBinView, and Zstd compete on both — all genuinely measured)
+        // rather than the extension-type first-match dispatch. No stats are computed —
+        // DictEncodingEncoder.expectedRatio() already defers to this path for Utf8 rather than
+        // consuming them, Binary isn't a Dict candidate at all — and there is no cheap analytic
+        // "no compression" baseline the way primitiveBytes is for fixed-width types, so the
+        // competition simply keeps whichever accepting encoder measures smallest
+        // (VarBinEncodingEncoder unconditionally accepts both, so a winner always exists in
+        // practice).
+        if (dtype instanceof DType.Utf8 || dtype instanceof DType.Binary) {
             return competeAndEncode(dtype, data, ctx, ArrayStats.EMPTY, sampleSize -> Long.MAX_VALUE);
         }
 
-        // Remaining non-primitives (extension types, Binary, List, ...): find the accepting
-        // encoding and splice through it so its cascaded children (e.g. datetimeparts →
-        // days/seconds/subseconds) are recursively compressed rather than stored as raw
-        // primitives. Honor the excluded set so spliceResult's notApplicable retry can rotate
-        // to the next accepting encoding (e.g. DateTimePartsEncoding → ExtEncoding when the
-        // input is raw storage rather than DateTimePartsData).
+        // Remaining non-primitives (extension types, List, ...): find the accepting encoding and
+        // splice through it so its cascaded children (e.g. datetimeparts → days/seconds/
+        // subseconds) are recursively compressed rather than stored as raw primitives. Honor the
+        // excluded set so spliceResult's notApplicable retry can rotate to the next accepting
+        // encoding (e.g. DateTimePartsEncoding → ExtEncoding when the input is raw storage rather
+        // than DateTimePartsData).
         if (!(dtype instanceof DType.Primitive p)) {
             return spliceResult(findPrimitiveEncoding(dtype, ctx.excluded()), dtype, data, ctx);
         }
@@ -340,8 +348,8 @@ private EncodeResult encodeStruct(DType.Struct dtype, StructData data, EncodeCon
             DType fieldDtype = fieldTypes.get(i);
             Object fieldData = fields.get(i);
             // Mirrors StructEncodingEncoder's own field loop: a nullable field arrives as
-            // NullableData(values, validity), not the dense array encodeWithCtx's per-dtype
-            // dispatch expects (e.g. VarBinEncodingEncoder casts data straight to String[]).
+            // NullableData(values, validity), not the dense array (String[], byte[][], ...)
+            // encodeWithCtx's per-dtype dispatch expects.
             EncodeResult fieldResult = (fieldData instanceof NullableData && !(fieldDtype instanceof DType.Extension))
                     ? new MaskedEncodingEncoder().encode(fieldDtype, fieldData, ctx)
                     : encodeWithCtx(fieldDtype, fieldData, ctx);

writer/src/main/java/io/github/dfa1/vortex/writer/encode/FsstEncodingEncoder.java

@@ -12,17 +12,17 @@
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;
-import java.nio.charset.StandardCharsets;
 import java.util.List;
 
 /// Write-only encoder for `vortex.fsst`.
 ///
 /// This class is a thin wire adapter over the standalone `vortex-fsst` module (issue #287): the
 /// FSST compression algorithm — symbol-table training and greedy longest-match compression — lives
-/// entirely in [CompressorBuilder]/[Compressor]. This adapter converts the input strings to UTF-8
-/// bytes, drives training, compresses each row, and lays the result out in the `vortex.fsst` wire
-/// format (symbol table buffers, remapped code stream, per-row uncompressed lengths and code
-/// offsets, plus the [ProtoFSSTMetadata] describing the two offset ptypes).
+/// entirely in [CompressorBuilder]/[Compressor]. This adapter normalizes the input (Utf8 `String[]`
+/// UTF-8 encoded, Binary `byte[][]` passed through — [VarBinBytes]) to raw row bytes, drives
+/// training, compresses each row, and lays the result out in the `vortex.fsst` wire format (symbol
+/// table buffers, remapped code stream, per-row uncompressed lengths and code offsets, plus the
+/// [ProtoFSSTMetadata] describing the two offset ptypes).
 ///
 /// The wire format packs each symbol's bytes LSB-first into a `long` (first byte in the low byte)
 /// alongside a per-symbol length byte, and reserves code `0xFF` as the single-literal-byte escape.
@@ -49,15 +49,13 @@ public EncodingId encodingId() {
 
     @Override
     public boolean accepts(DType dtype) {
-        // Binary excluded: encode()/encodeCascade() cast data straight to String[], not byte-safe
-        // for arbitrary bytes yet (#352).
-        return dtype instanceof DType.Utf8;
+        return dtype instanceof DType.Utf8 || dtype instanceof DType.Binary;
     }
 
     @Override
     public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
         Arena arena = ctx.arena();
-        Fsst c = compress((String[]) data, arena);
+        Fsst c = compress(data, arena);
 
         // Terminal layout: the per-row length and cumulative-offset children are raw primitive
         // segments (buffers 3 and 4). The cascading path (encodeCascade) instead exposes them as
@@ -95,15 +93,15 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
     /// to the terminal raw-primitive layout.
     ///
     /// @param dtype the Utf8/Binary type being encoded
-    /// @param data  the string values
+    /// @param data  the Utf8 (`String[]`) or Binary (`byte[][]`) values
     /// @param ctx   encoding context supplying the arena and cascade depth
     /// @return a cascade step with the two offset children left open, or a terminal step at depth 0
     @Override
     public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext ctx) {
         if (ctx.allowedCascading() == 0) {
             return CascadeStep.terminal(encode(dtype, data, ctx));
         }
-        Fsst c = compress((String[]) data, ctx.arena());
+        Fsst c = compress(data, ctx.arena());
         Object uncompLens = typedUnsigned(c.uncompLenPType(), c.uncompLens());
         Object codesOffsets = typedUnsigned(c.codesOffPType(), c.codesOffsets());
         EncodeNode partialRoot = new EncodeNode(
@@ -128,14 +126,13 @@ private record Fsst(
             PType uncompLenPType, PType codesOffPType, int n) {
     }
 
-    private static Fsst compress(String[] strings, Arena arena) {
-        int n = strings.length;
+    private static Fsst compress(Object data, Arena arena) {
+        byte[][] byteArrays = VarBinBytes.toByteArrays(data);
+        int n = byteArrays.length;
 
-        byte[][] byteArrays = new byte[n][];
         long totalInput = 0;
         int maxUncompLen = 0;
         for (int i = 0; i < n; i++) {
-            byteArrays[i] = strings[i].getBytes(StandardCharsets.UTF_8);
             totalInput += byteArrays[i].length;
             maxUncompLen = Math.max(maxUncompLen, byteArrays[i].length);
         }

writer/src/main/java/io/github/dfa1/vortex/writer/encode/MaskedEncodingEncoder.java

@@ -19,6 +19,8 @@
 /// (primitive / varbin / fixed-size-list).
 public final class MaskedEncodingEncoder implements EncodingEncoder {
 
+    private static final byte[] EMPTY_BYTES = new byte[0];
+
     private static final List<EncodingEncoder> INNER_FALLBACK = List.of(
             new PrimitiveEncodingEncoder(),
             new VarBinEncodingEncoder(),
@@ -69,11 +71,11 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
     /// never selects itself: its [#accepts] returns `false`, so the compressor cannot recurse into it.
     private static EncodeResult encodeValues(DType nonNullable, Object values, EncodeContext ctx) {
         if (ctx.allowedCascading() > 0) {
-            // The NullableData Utf8/Binary carrier keeps null elements in the String[] at null
-            // positions (validity masks them). Dict/FSST call getBytes() on every element, so
-            // substitute the empty string for nulls first — matching VarBinEncodingEncoder, which
-            // encodes a null as a zero-length slot. Those slots are never read: the enclosing
-            // vortex.masked validity bitmap marks the rows null.
+            // The NullableData Utf8/Binary carrier keeps null elements at null positions in the
+            // String[]/byte[][] (validity masks them). Dict/FSST/Zstd call getBytes()/read the row
+            // bytes of every element, so substitute an empty placeholder for nulls first —
+            // matching VarBinEncodingEncoder, which encodes a null as a zero-length slot. Those
+            // slots are never read: the enclosing vortex.masked validity bitmap marks the rows null.
             Object dense = denseValues(values);
             List<EncodingEncoder> candidates =
                     List.copyOf(ctx.registry().encoderMap().values());
@@ -142,15 +144,23 @@ private static boolean isConstantValidity(boolean[] validity) {
         return true;
     }
 
-    /// Returns `values` unchanged, except a `String[]` with null elements is copied with each null
-    /// replaced by the empty string so cascade encoders (Dict, FSST) can call `getBytes()` safely.
+    /// Returns `values` unchanged, except a `String[]`/`byte[][]` with null elements is copied
+    /// with each null replaced by an empty placeholder so cascade encoders (Dict, FSST, Zstd,
+    /// VarBinView) can read every element's bytes safely.
     ///
     /// @param values the non-nullable values carrier extracted from the [NullableData]
-    /// @return the same array, or a null-free copy when it is a `String[]` containing nulls
+    /// @return the same array, or a null-free copy when it is a `String[]`/`byte[][]` containing nulls
     private static Object denseValues(Object values) {
-        if (!(values instanceof String[] strings)) {
-            return values;
+        if (values instanceof String[] strings) {
+            return densifyStrings(strings);
+        }
+        if (values instanceof byte[][] raw) {
+            return densifyBytes(raw);
         }
+        return values;
+    }
+
+    private static String[] densifyStrings(String[] strings) {
         String[] out = null;
         for (int i = 0; i < strings.length; i++) {
             if (strings[i] == null) {
@@ -163,6 +173,19 @@ private static Object denseValues(Object values) {
         return out != null ? out : strings;
     }
 
+    private static byte[][] densifyBytes(byte[][] raw) {
+        byte[][] out = null;
+        for (int i = 0; i < raw.length; i++) {
+            if (raw[i] == null) {
+                if (out == null) {
+                    out = raw.clone();
+                }
+                out[i] = EMPTY_BYTES;
+            }
+        }
+        return out != null ? out : raw;
+    }
+
     private static EncodingEncoder pickInner(DType nonNullable) {
         for (EncodingEncoder e : INNER_FALLBACK) {
             if (e.accepts(nonNullable)) {

writer/src/main/java/io/github/dfa1/vortex/writer/encode/VarBinBytes.java

@@ -0,0 +1,48 @@
+package io.github.dfa1.vortex.writer.encode;
+
+import java.nio.charset.StandardCharsets;
+
+/// Normalizes Utf8 (`String[]`) or Binary (`byte[][]`) encoder input to a common `byte[][]`
+/// shape — Utf8 elements are UTF-8 encoded, Binary elements pass through unchanged. Shared by
+/// every varbin-family encoder ([VarBinEncodingEncoder], [VarBinViewEncodingEncoder],
+/// [FsstEncodingEncoder], [ZstdEncodingEncoder]) so a `DType.Binary` column gets the same
+/// byte-safe treatment `DType.Utf8` already had (issue #352).
+final class VarBinBytes {
+
+    private static final byte[] EMPTY = new byte[0];
+
+    private VarBinBytes() {
+    }
+
+    /// Converts `data` to `byte[][]`, preserving `null` entries as Java `null` rather than
+    /// substituting a placeholder.
+    ///
+    /// @param data a `String[]` (UTF-8 encoded) or `byte[][]` (returned row-for-row unchanged)
+    /// @return the row bytes, with any `null` entries preserved
+    static byte[][] toRawByteArrays(Object data) {
+        if (data instanceof byte[][] raw) {
+            return raw;
+        }
+        String[] strings = (String[]) data;
+        byte[][] out = new byte[strings.length][];
+        for (int i = 0; i < strings.length; i++) {
+            out[i] = strings[i] == null ? null : strings[i].getBytes(StandardCharsets.UTF_8);
+        }
+        return out;
+    }
+
+    /// Like [#toRawByteArrays(Object)], but substitutes a zero-length array for every `null`
+    /// entry — the values child of a masked/nullable layout, where validity (not this array)
+    /// carries nullity, so a null entry's bytes are never read back.
+    ///
+    /// @param data a `String[]` (UTF-8 encoded) or `byte[][]` (returned row-for-row unchanged)
+    /// @return the row bytes, with `null` entries replaced by a zero-length array
+    static byte[][] toByteArrays(Object data) {
+        byte[][] raw = toRawByteArrays(data);
+        byte[][] out = new byte[raw.length][];
+        for (int i = 0; i < raw.length; i++) {
+            out[i] = raw[i] == null ? EMPTY : raw[i];
+        }
+        return out;
+    }
+}

writer/src/main/java/io/github/dfa1/vortex/writer/encode/VarBinEncodingEncoder.java

@@ -9,14 +9,11 @@
 
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
-import java.nio.charset.StandardCharsets;
 import java.util.List;
 
 /// Write-only encoder for `vortex.varbin`.
 public final class VarBinEncodingEncoder implements EncodingEncoder {
 
-    private static final byte[] EMPTY_BYTES = new byte[0];
-
     @Override
     public EncodingId encodingId() {
         return EncodingId.VORTEX_VARBIN;
@@ -30,24 +27,12 @@ public boolean accepts(DType dtype) {
     @Override
     public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
         // Binary (DType.Binary) arrives as raw byte[][] and must round-trip byte-for-byte —
-        // routing it through the Utf8 String[] path below would corrupt any byte sequence that
-        // isn't valid UTF-8 (e.g. an embedded audio/image blob). Utf8 arrives as String[] and is
-        // UTF-8 encoded. Either way a null entry (this encoder is the values child of a
-        // masked/nullable layout, where validity carries nullity) contributes a zero-length slot.
-        byte[][] byteArrays;
-        String[] strings = null;
-        if (data instanceof byte[][] raw) {
-            byteArrays = new byte[raw.length][];
-            for (int i = 0; i < raw.length; i++) {
-                byteArrays[i] = raw[i] == null ? EMPTY_BYTES : raw[i];
-            }
-        } else {
-            strings = (String[]) data;
-            byteArrays = new byte[strings.length][];
-            for (int i = 0; i < strings.length; i++) {
-                byteArrays[i] = strings[i] == null ? EMPTY_BYTES : strings[i].getBytes(StandardCharsets.UTF_8);
-            }
-        }
+        // routing it through the Utf8 String[] path would corrupt any byte sequence that isn't
+        // valid UTF-8 (e.g. an embedded audio/image blob). Utf8 arrives as String[] and is UTF-8
+        // encoded. Either way a null entry (this encoder is the values child of a masked/nullable
+        // layout, where validity carries nullity) contributes a zero-length slot.
+        String[] strings = data instanceof String[] s ? s : null;
+        byte[][] byteArrays = VarBinBytes.toByteArrays(data);
         int n = byteArrays.length;
         int totalBytes = 0;
         for (byte[] b : byteArrays) {

writer/src/main/java/io/github/dfa1/vortex/writer/encode/VarBinViewEncodingEncoder.java

@@ -6,7 +6,6 @@
 
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
-import java.nio.charset.StandardCharsets;
 import java.util.List;
 
 /// Write-only encoder for `vortex.varbinview`.
@@ -22,20 +21,16 @@ public EncodingId encodingId() {
 
     @Override
     public boolean accepts(DType dtype) {
-        // Binary excluded: encode() casts data straight to String[], not byte-safe for
-        // arbitrary bytes yet (#352).
-        return dtype instanceof DType.Utf8;
+        return dtype instanceof DType.Utf8 || dtype instanceof DType.Binary;
     }
 
     @Override
     public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
-        String[] strings = (String[]) data;
-        int n = strings.length;
+        byte[][] bytes = VarBinBytes.toByteArrays(data);
+        int n = bytes.length;
 
-        byte[][] bytes = new byte[n][];
         int totalDataBytes = 0;
         for (int i = 0; i < n; i++) {
-            bytes[i] = strings[i].getBytes(StandardCharsets.UTF_8);
             if (bytes[i].length > MAX_INLINED_SIZE) {
                 totalDataBytes += bytes[i].length;
             }