fix(writer): null as a global-dict entry for string and binary columns

## fix(writer): store null as a global-dict entry for string and binary columns

Same shape as the numeric path: a nullable Utf8/Binary global dict now
writes null as one more pool entry, invalid in the pool's validity, with
non-nullable U16 codes and is_nullable_codes = false, as Rust's dict
layout writer does. The pool (FSST/VarBin/Zstd competition) is now masked
when it holds the null entry, so writeSegment's masked override honors
the cascade exclusions too: the pool must never become a dict itself,
which the reader cannot unwrap.

Taxi 2024-01: 42,342,166 -> 42,340,950 bytes (vortex-jni 44,463,892)
  store_and_fwd_flag 66,460 -> 64,990 (jni 47,536: 1.40x -> 1.37x)
  pool and layout metadata now match vortex-jni's shape; the rest of
  the gap is 46 code chunks of 64k rows vs Rust's 6 of 512k
OHLC 10M rows: unchanged at 58,230,658 bytes.

JavaVsJniWriteBenchmark.javaWriteCascading, 5 forks:
  0.769 +- 0.026 -> 0.767 +- 0.017 ops/s (per fork 0.769 0.768 0.805 0.742 0.751)
  gc.alloc.rate.norm 2.50 -> 2.50 GB/op

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015R8cf8YM8XFQqrkyhSAtYr

Author: dfa1

CHANGELOG.md

@@ -8,7 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Changed
-- Nullable low-cardinality numeric columns written with a global dictionary store null as one more dictionary entry, invalid in the values' validity, with non-nullable codes, as Rust's dict layout does: no row mask on the codes, and their encoding is chosen on the real codes rather than placeholder ones. Global-dict codes are always U16, Rust's default, whatever the dictionary size ([#467](https://github.com/dfa1/vortex-java/pull/467)).
+- Nullable low-cardinality numeric, string and binary columns written with a global dictionary store null as one more dictionary entry, invalid in the values' validity, with non-nullable codes, as Rust's dict layout does: no row mask on the codes, and their encoding is chosen on the real codes rather than placeholder ones. Global-dict codes are always U16, Rust's default, whatever the dictionary size ([#467](https://github.com/dfa1/vortex-java/pull/467), [#468](https://github.com/dfa1/vortex-java/pull/468)).
 - Cascading writes are about 35% faster and allocate half as much (10M-row OHLC: 0.44 → 0.60 writes/s), to the same bytes: as in Rust, cascade children are no longer trial-encoded with candidates their stats rule out, and RLE/RunEnd skip arrays whose runs average under 4 values ([#465](https://github.com/dfa1/vortex-java/pull/465)).
 - **Breaking:** `EncodingEncoder#expectedRatio` takes `(DType, ArrayAndStats, EncodeContext)` instead of `(DType, Object, ArrayStats)`, as Rust's `expected_compression_ratio` takes `ArrayAndStats` and the compressor context: the cascade's stats scan runs on the first `stats()` call and is shared by every candidate, and `EncodeContext#sample()`/`finishedCascading()` tell a verdict where it is. `Estimate` is a sealed interface whose `Estimate.ratio(double)` competes with sampled candidates without encoding, as Rust's `EstimateVerdict::Ratio` does ([#464](https://github.com/dfa1/vortex-java/pull/464), [#466](https://github.com/dfa1/vortex-java/pull/466)).
 - **Breaking:** default writes emit Rust's `vortex.zoned` zone map instead of the legacy `vortex.stats`: one zone per 8192 rows regardless of `writeChunk` batches, with Rust's per-dtype stats (64-byte string bounds, `nan_count`) and no zone sum. Calcite and `ZoneReducer` aggregate push-down needs the legacy layout, so on these files it falls back to a scan; target `Editions.CORE_2025_10_0` to keep it. Read new files with this release or later ([#447](https://github.com/dfa1/vortex-java/issues/447)).

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

@@ -452,8 +452,7 @@ Object buildFrequencyRankedUniqueArray(int[] remap) {
     /// Emits one chunk's wire codes ([#CODES_PTYPE]) from its buffered `short[]` first-seen codes,
     /// optionally translated through a frequency remap (primitive path). A null slot (validity[i]
     /// false) emits `nullCode` unconditionally, never remapped: the primitive path points it at the
-    /// pool's invalid null entry, as Rust's dict builder does; the Utf8/Binary path passes `0`, which
-    /// the reader ignores because its codes child is masked by the same validity.
+    /// pool's invalid null entry, as Rust's dict builder does.
     ///
     /// @param buffered the buffered first-seen codes for one chunk
     /// @param remap    the first-seen -> frequency-rank remap, or `null` to emit codes unchanged
@@ -474,9 +473,10 @@ static short[] emitCodes(short[] buffered, int[] remap, boolean[] validity, int
     }
 
     /// The values pool with one more slot, for the invalid null entry Rust's dict builder adds
-    /// when it meets a null: the slot holds a zero placeholder and is masked off by the caller.
+    /// when it meets a null: the slot holds a zero (or `null`) placeholder and is masked off by the
+    /// caller.
     ///
-    /// @param values the frequency-ranked distinct values (a primitive array)
+    /// @param values the distinct values (a primitive array, `String[]` or `byte[][]`)
     /// @return a copy one element longer
     static Object withNullSlot(Object values) {
         return switch (values) {
@@ -486,7 +486,10 @@ static Object withNullSlot(Object values) {
             case byte[] a -> Arrays.copyOf(a, a.length + 1);
             case double[] a -> Arrays.copyOf(a, a.length + 1);
             case float[] a -> Arrays.copyOf(a, a.length + 1);
-            default -> throw new IllegalStateException("not a primitive values pool: " + values.getClass());
+            // Utf8/Binary: the null slot holds a real null, as ChunkImpl leaves it at invalid rows
+            case String[] a -> Arrays.copyOf(a, a.length + 1);
+            case byte[][] a -> Arrays.copyOf(a, a.length + 1);
+            default -> throw new IllegalStateException("not a values pool: " + values.getClass());
         };
     }
 

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

@@ -691,6 +691,9 @@ private int writeSegment(DType dtype, Object data, EncodingEncoder encodingOverr
                 EncodeContext encodeCtx = options.allowedCascading() > 0
                         ? EncodeContext.ofDepth(options.allowedCascading(), arena, cascadeRegistry, editionExcluded)
                         : EncodeContext.of(arena, defaultRegistry, editionExcluded);
+                // A wrapping override (masked) cascades its inner values, so the exclusions hold
+                // there too -- a nullable global-dict pool must never become a dict itself.
+                encodeCtx = encodeCtx.withExcluded(excludedFromCascade);
                 result = encodingOverride.encode(dtype, data, encodeCtx);
             } else if (options.allowedCascading() > 0) {
                 EncodeContext encodeCtx = EncodeContext.ofDepth(options.allowedCascading(), arena, cascadeRegistry, editionExcluded);
@@ -1366,31 +1369,45 @@ private void writeGlobalDictColumn(ColumnName colName, DictColumnState state) th
     }
 
     private void writeGlobalDictVarBinColumn(ColumnName colName, DictColumnState state) throws IOException {
-        int dictSize = state.cardinality();
+        // Null is one more pool entry, invalid, with non-nullable codes -- Rust's dict layout
+        // shape, as for numeric columns (writeGlobalDictColumn).
+        boolean hasNulls = state.chunkNullCounts().stream().anyMatch(c -> c > 0);
+        int distinct = state.cardinality();
+        int dictSize = hasNulls ? distinct + 1 : distinct;
 
         // Utf8/Binary assigns codes in first-seen order with no frequency sort, so the incremental map's
         // order already matches — no remap pass (ADR 0021). Compress the distinct-values pool
         // through the normal Utf8/Binary competition (FSST/VarBin/Zstd) so it captures substring
         // redundancy across dictionary entries (#299), but exclude Dict so the cascade never wraps
         // the (all-unique-by-construction) dictionary in another dict the reader cannot unwrap. At
-        // cascade depth 0 there is no competition to run, so force flat VarBin as before.
+        // cascade depth 0 there is no competition to run, so force flat VarBin as before; a pool
+        // with a null entry is masked there instead, its values still VarBin.
         Object uniques = state.varBinUniques();
-        int valuesSegIdx = options.allowedCascading() > 0
-                ? writeSegment(state.dtype(), uniques, null, Set.of(EncodingId.VORTEX_DICT))
-                : writeSegment(state.dtype(), uniques, new VarBinEncodingEncoder());
+        Object pool = uniques;
+        if (hasNulls) {
+            boolean[] poolValidity = new boolean[dictSize];
+            Arrays.fill(poolValidity, 0, distinct, true);
+            pool = new NullableData(DictColumnState.withNullSlot(uniques), poolValidity);
+        }
+        int valuesSegIdx;
+        if (options.allowedCascading() > 0) {
+            valuesSegIdx = writeSegment(state.dtype(), pool, null, Set.of(EncodingId.VORTEX_DICT));
+        } else if (hasNulls) {
+            valuesSegIdx = writeSegment(state.dtype(), pool, new MaskedEncodingEncoder(), Set.of(EncodingId.VORTEX_DICT));
+        } else {
+            valuesSegIdx = writeSegment(state.dtype(), pool, new VarBinEncodingEncoder());
+        }
 
-        DType codesDtype = new DType.Primitive(DictColumnState.CODES_PTYPE, state.nullable());
+        DType codesDtype = new DType.Primitive(DictColumnState.CODES_PTYPE, false);
         List<Integer> codesSegIdxes = new ArrayList<>();
         for (int c = 0; c < state.chunkCount(); c++) {
-            boolean[] validity = state.chunkValidity(c);
-            Object codesArr = DictColumnState.emitCodes(state.chunkCodes(c), null, validity, 0);
-            Object codesData = validity != null ? new NullableData(codesArr, validity) : codesArr;
-            codesSegIdxes.add(writeSegment(codesDtype, codesData));
+            Object codesArr = DictColumnState.emitCodes(state.chunkCodes(c), null, state.chunkValidity(c), distinct);
+            codesSegIdxes.add(writeSegment(codesDtype, codesArr));
         }
 
         dictColRefs.put(colName, new DictColRef(valuesSegIdx, dictSize, codesSegIdxes,
                 state.chunkRowCounts(), state.chunkNullCounts(),
-                state.chunkStatsMin(), state.chunkStatsMax(), state.chunkStatsSum(), state.nullable()));
+                state.chunkStatsMin(), state.chunkStatsMax(), state.chunkStatsSum(), false));
     }
 
     private record SegRef(long offset, long len) {