feat(writer): compress the global-dict Utf8 values pool (#299)

The global-dict flush wrote a low-cardinality Utf8 column's distinct-values pool
as raw vortex.varbin, so it never captured substring redundancy across dictionary
entries (shared "STREET"/"AVENUE"/borough tokens) — a large part of the nyc-311
file-size gap vs the Rust reference, which dict-encodes *and* FSST-compresses the
same columns.

writeGlobalDictUtf8Column now runs the values pool through the normal Utf8
competition (FSST/VarBin/Zstd) with vortex.dict excluded, so the cascade never
wraps the dictionary in a second dict the reader cannot unwrap. Threaded via a new
excludedFromCascade argument on writeSegment. At cascade depth 0 (no competition)
it keeps forcing flat VarBin as before.

Verified: values pool is FSST-encoded and round-trips (new integration test);
80 writer dict tests and 217 Java-writes-Rust-reads interop tests still pass.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dfa1

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

@@ -0,0 +1,108 @@
+package io.github.dfa1.vortex.integration;
+
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.inspect.InspectorTree;
+import io.github.dfa1.vortex.reader.ReadRegistry;
+import io.github.dfa1.vortex.reader.ScanOptions;
+import io.github.dfa1.vortex.reader.VortexReader;
+import io.github.dfa1.vortex.reader.array.Array;
+import io.github.dfa1.vortex.reader.array.VarBinArray;
+import io.github.dfa1.vortex.writer.VortexWriter;
+import io.github.dfa1.vortex.writer.WriteOptions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/// Global-dict Utf8 columns compress their distinct-values pool through the normal Utf8
+/// competition (FSST/VarBin/Zstd) instead of hardcoded raw varbin (#299). Rust dict-encodes
+/// *and* FSST-compresses the same columns; before this change Java left the pool uncompressed,
+/// which was a large part of the nyc-311 file-size gap.
+class DictValuesPoolCompressionIntegrationTest {
+
+    private static final ColumnName COL = ColumnName.of("desc");
+    private static final DType.Struct SCHEMA = new DType.Struct(List.of(COL), List.of(DType.UTF8), false);
+
+    /// Builds a low-cardinality column whose distinct values share a long common prefix, so FSST
+    /// beats raw varbin on the values pool decisively. `distinct * 2 < rows` keeps it inside the
+    /// global-dict gate; `distinct < 2048` keeps it inside the current cardinality cap so this test
+    /// exercises Finding 2 alone (independent of the cap raise).
+    private static String[] redundantChunk(int rows, int distinct, int chunkSeed) {
+        String prefix = "New York City Department of Housing Preservation and Development complaint at ";
+        String[] out = new String[rows];
+        for (int i = 0; i < rows; i++) {
+            int id = (chunkSeed + i) % distinct;
+            out[i] = prefix + "site #" + id + " STREET";
+        }
+        return out;
+    }
+
+    @Test
+    void globalDictUtf8_valuesPool_isFsstCompressed_andRoundTrips(@TempDir Path tmp) throws IOException {
+        // Given — 800 distinct long strings sharing a 78-char prefix, repeated across 5 chunks.
+        Path file = tmp.resolve("dict_fsst.vortex");
+        int rowsPerChunk = 4_000;
+        int chunkCount = 5;
+        int distinct = 800;
+
+        List<String> expected = new ArrayList<>();
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var sut = VortexWriter.create(ch, SCHEMA, WriteOptions.cascading(3))) {
+            // When
+            for (int c = 0; c < chunkCount; c++) {
+                String[] data = redundantChunk(rowsPerChunk, distinct, c * 37);
+                expected.addAll(List.of(data));
+                sut.writeChunk(Map.of(COL, data));
+            }
+        }
+
+        // Then — the values pool is FSST-compressed. The global dictionary itself is a *layout*
+        // (LAYOUT_DICT), so it is not listed among the segment encodings; FSST winning the
+        // values-pool competition is the observable proof the pool is no longer raw varbin.
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll())) {
+            InspectorTree tree = InspectorTree.build(vf);
+            assertThat(tree.usedEncodings())
+                    .as("FSST compresses the dictionary values pool")
+                    .contains("vortex.fsst");
+        }
+
+        // And global-dict dedup + FSST keeps the file far below the ~1.8 MB of raw repeated
+        // strings (20 000 rows × ~90 B) — dedup to 800 distinct, values pool FSST-compressed.
+        assertThat(Files.size(file))
+                .as("dict dedup + FSST values pool")
+                .isLessThan(300_000L);
+
+        // And every value round-trips exactly.
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll())) {
+            List<String> got = readAllStrings(vf);
+            assertThat(got).isEqualTo(expected);
+        }
+    }
+
+    private static List<String> readAllStrings(VortexReader vf) {
+        List<String> out = new ArrayList<>();
+        try (var iter = vf.scan(ScanOptions.columns("desc"))) {
+            while (iter.hasNext()) {
+                try (var chunk = iter.next()) {
+                    Array col = chunk.column("desc");
+                    VarBinArray strings = (VarBinArray) col;
+                    long n = strings.length();
+                    for (long i = 0; i < n; i++) {
+                        out.add(strings.getString(i));
+                    }
+                }
+            }
+        }
+        return out;
+    }
+}

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

@@ -588,12 +588,27 @@ private int writeSegment(DType dtype, Object data) throws IOException {
         return writeSegment(dtype, data, null);
     }
 
-    /// Writes a segment, optionally forcing a specific `encodingOverride` instead of
-    /// the configured cascade. Used by the global Utf8 dictionary path where the values
-    /// segment must be flat varbin — the cascade would otherwise re-pick
-    /// [DictEncodingEncoder] and wrap the dictionary in another dict (which the reader
-    /// cannot unwrap).
     private int writeSegment(DType dtype, Object data, EncodingEncoder encodingOverride) throws IOException {
+        return writeSegment(dtype, data, encodingOverride, Set.of());
+    }
+
+    /// Writes a segment, optionally forcing a specific `encodingOverride` instead of the
+    /// configured cascade, and optionally excluding encodings from the cascade competition.
+    ///
+    /// `excludedFromCascade` lets the global Utf8 dictionary path compress its values pool
+    /// through the normal Utf8 competition (FSST/VarBin/Zstd) while excluding
+    /// [DictEncodingEncoder] — the cascade would otherwise re-pick it and wrap the dictionary
+    /// in another dict the reader cannot unwrap. Only consulted on the cascade branch
+    /// (`encodingOverride == null && allowedCascading > 0`).
+    ///
+    /// @param dtype               logical type of the segment
+    /// @param data                the values to encode
+    /// @param encodingOverride    a specific encoder to force, or `null` to use the cascade
+    /// @param excludedFromCascade encoding ids to exclude from the cascade competition
+    /// @return the index of the written segment
+    /// @throws IOException if writing to the channel fails
+    private int writeSegment(DType dtype, Object data, EncodingEncoder encodingOverride,
+            Set<EncodingId> excludedFromCascade) throws IOException {
         // Non-extension nullable columns (Primitive, Utf8) wrap with MaskedEncodingEncoder here.
         // FbsExtension columns route through ExtEncodingEncoder.encode which itself delegates to
         // MaskedEncodingEncoder when its storage data is NullableData — handled inside ExtEncoding.
@@ -624,6 +639,9 @@ private int writeSegment(DType dtype, Object data, EncodingEncoder encodingOverr
                 result = encodingOverride.encode(dtype, data, encodeCtx);
             } else if (options.allowedCascading() > 0) {
                 EncodeContext encodeCtx = EncodeContext.ofDepth(options.allowedCascading(), arena, cascadeRegistry);
+                for (EncodingId excluded : excludedFromCascade) {
+                    encodeCtx = encodeCtx.withExcluded(excluded);
+                }
                 CascadingCompressor compressor = new CascadingCompressor(cascadeCodecs);
                 result = compressor.encode(dtype, data, encodeCtx);
             } else {
@@ -1510,11 +1528,15 @@ private void writeGlobalDictUtf8Column(ColumnName colName, DictColumnState state
         PType codePType = codePTypeForSize(dictSize);
 
         // Utf8 assigns codes in first-seen order with no frequency sort, so the incremental map's
-        // order already matches — no remap pass (ADR 0021). Write unique strings as a flat VarBin
-        // segment, bypassing cascade so DictEncoding doesn't wrap the (all-unique-by-construction)
-        // dictionary in another dict the reader cannot unwrap.
+        // order already matches — no remap pass (ADR 0021). Compress the distinct-values pool
+        // through the normal Utf8 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.
         String[] uniques = state.valueToCode.keySet().toArray(new String[0]);
-        int valuesSegIdx = writeSegment(state.dtype, uniques, new VarBinEncodingEncoder());
+        int valuesSegIdx = options.allowedCascading() > 0
+                ? writeSegment(state.dtype, uniques, null, Set.of(EncodingId.VORTEX_DICT))
+                : writeSegment(state.dtype, uniques, new VarBinEncodingEncoder());
 
         DType codesDtype = new DType.Primitive(codePType, state.nullable);
         List<Integer> codesSegIdxes = new ArrayList<>();