feat(writer): cascade the per-chunk dict Utf8 values pool (#299)

DictEncodingEncoder.encodeUtf8 hardcoded the distinct-values pool as a terminal
raw vortex.varbin node, so per-chunk dict-encoded Utf8 columns never FSST-compressed
their values. encodeCascade now exposes the values pool as an open Utf8 child slot
instead of CascadeStep.terminal, so the cascade competes FSST/VarBin/Zstd on it.
CascadingCompressor already excludes the winning vortex.dict from a child's
competition, so the pool is never wrapped in a second dict the reader cannot unwrap.

The non-cascade encode() path (cascade depth 0) still emits terminal varbin.

Verified: encodeCascade exposes a single open values child (unit test); per-chunk
dict Utf8 round-trips through the new path; 217 Java-writes-Rust-reads interop tests
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

@@ -123,6 +123,43 @@ void utf8_cardinalityAbovePriorCap_admittedToGlobalDict(@TempDir Path tmp) throw
         }
     }
 
+    @Test
+    void perChunkDictUtf8_valuesPool_isFsstCompressed(@TempDir Path tmp) throws IOException {
+        // Given — global dict disabled, so a redundant low-cardinality Utf8 column is dict-encoded
+        // per chunk (a vortex.dict *encoding*, not a layout). High row duplication makes per-chunk
+        // dict win the cascade; the shared-prefix distinct values make FSST win the values child.
+        Path file = tmp.resolve("perchunk_dict.vortex");
+        int rowsPerChunk = 10_000;
+        int chunkCount = 3;
+        int distinct = 500;
+
+        List<String> expected = new ArrayList<>();
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var sut = VortexWriter.create(ch, SCHEMA, WriteOptions.cascading(3).withGlobalDict(false))) {
+            // When
+            for (int c = 0; c < chunkCount; c++) {
+                String[] data = redundantChunk(rowsPerChunk, distinct, c * 53);
+                expected.addAll(List.of(data));
+                sut.writeChunk(Map.of(COL, data));
+            }
+        }
+
+        // Then — per-chunk dict still wins (the values pool is now a cascaded child rather than a
+        // terminal varbin node; its encoding is nested inside the dict segment, so it is not visible
+        // in usedEncodings — the mechanism is covered by DictEncodingEncoderTest, the round-trip
+        // below guards correctness through the new cascade path).
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll())) {
+            assertThat(InspectorTree.build(vf).usedEncodings())
+                    .as("per-chunk dict encoding")
+                    .contains("vortex.dict");
+        }
+
+        // And every value round-trips exactly.
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll())) {
+            assertThat(readAllStrings(vf)).isEqualTo(expected);
+        }
+    }
+
     private static boolean hasDictLayout(InspectorTree.Node node) {
         if (node.layout().isDict()) {
             return true;

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

@@ -84,7 +84,7 @@ public EncodeResult encode(DType dtype, Object data, EncodeContext ctx) {
     @Override
     public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext ctx) {
         if (dtype instanceof DType.Utf8) {
-            return CascadeStep.terminal(encodeUtf8((String[]) data, ctx));
+            return encodeUtf8Cascade((String[]) data, ctx);
         }
         DictData d = buildDictData(dtype, data, ctx);
         PType codePType = d.codePType();
@@ -101,6 +101,56 @@ public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext ctx) {
         return new CascadeStep(partialRoot, List.of(d.valuesBuf()), List.of(slot), null, null, true);
     }
 
+    /// Cascading Utf8 dict: emit the codes leaf but expose the distinct-values pool as an open Utf8
+    /// child slot so the cascade competes FSST/VarBin/Zstd on it (#299) instead of hardcoding raw
+    /// varbin. [CascadingCompressor] excludes the winning `vortex.dict` from that child's
+    /// competition, so the pool is never wrapped in a second dict the reader cannot unwrap.
+    ///
+    /// @param strings the column's string values
+    /// @param ctx     encoding context supplying the arena
+    /// @return a cascade step whose values child (index 1) is left open for compression
+    private static CascadeStep encodeUtf8Cascade(String[] strings, EncodeContext ctx) {
+        int n = strings.length;
+
+        var valueMap = new LinkedHashMap<String, Integer>();
+        for (String s : strings) {
+            valueMap.computeIfAbsent(s, _ -> valueMap.size());
+        }
+        int dictSize = valueMap.size();
+        PType codePType = codePType(dictSize);
+        int codeBytes = codePType.byteSize();
+
+        MemorySegment codesBuf = ctx.arena().allocate((long) n * codeBytes);
+        for (int i = 0; i < n; i++) {
+            writeCodeToSeg(codesBuf, codePType, i, valueMap.get(strings[i]));
+        }
+
+        byte[] metaBytes = new ProtoDictMetadata(
+                dictSize,
+                io.github.dfa1.vortex.core.proto.ProtoPType.fromValue(codePType.ordinal()),
+                null,
+                null
+        ).encode();
+
+        // Child order matches the terminal encodeUtf8: [codes, values]. Codes is a raw primitive
+        // leaf at buffer 0; values (index 1) is left null for the cascade to fill.
+        EncodeNode codesNode = EncodeNode.leaf(EncodingId.VORTEX_PRIMITIVE, 0);
+        EncodeNode partialRoot = new EncodeNode(
+                EncodingId.VORTEX_DICT, MemorySegment.ofArray(metaBytes),
+                new EncodeNode[]{codesNode, null},
+                new int[0]);
+
+        String[] distinct = valueMap.keySet().toArray(new String[0]);
+        ChildSlot valuesSlot = new ChildSlot(DType.UTF8, distinct, 1);
+
+        String minStr = valueMap.keySet().stream().min(String::compareTo).orElse(null);
+        String maxStr = valueMap.keySet().stream().max(String::compareTo).orElse(null);
+        byte[] statsMin = minStr != null ? ProtoScalarValue.ofStringValue(minStr).encode() : null;
+        byte[] statsMax = maxStr != null ? ProtoScalarValue.ofStringValue(maxStr).encode() : null;
+
+        return new CascadeStep(partialRoot, List.of(codesBuf), List.of(valuesSlot), statsMin, statsMax, true);
+    }
+
     private static EncodeResult encodeUtf8(String[] strings, EncodeContext ctx) {
         int n = strings.length;
 

writer/src/test/java/io/github/dfa1/vortex/writer/encode/DictEncodingEncoderTest.java

@@ -178,4 +178,22 @@ void encode_utf8_metadata_valuesLen_matchesUniqueCount() throws Exception {
         // Then
         assertThat(meta.values_len()).isEqualTo(2);
     }
+
+    @Test
+    void encodeCascade_utf8_exposesValuesPoolAsOpenChild() {
+        // Given — a low-cardinality Utf8 column.
+        String[] data = repeat(new String[]{"apple", "banana", "cherry"}, 100);
+
+        // When
+        CascadeStep result = ENCODER.encodeCascade(DTypes.UTF8, data, EncodeTestHelper.testCtx());
+
+        // Then — the distinct-values pool (child index 1) is left open for the cascade to compress
+        // (FSST/VarBin compete on it, #299), not hardcoded to a terminal raw-varbin node.
+        assertThat(result.isTerminal()).isFalse();
+        assertThat(result.openChildren()).singleElement().satisfies(slot -> {
+            assertThat(slot.parentChildIdx()).isEqualTo(1);
+            assertThat(slot.childDtype()).isEqualTo(DTypes.UTF8);
+            assertThat((String[]) slot.childData()).containsExactly("apple", "banana", "cherry");
+        });
+    }
 }