feat(writer): add WriteOptions.withZstd(boolean) for opt-in size/throughput tradeoff

Add `enableZstd` field to WriteOptions and `withZstd(boolean)` fluent method.
When enabled, ZstdEncoding is inserted into the cascade codec list and competes
with ALP/bitpack on every chunk — recovers the ~43 MB NYC taxi level (vs 49.7 MB
without Zstd) at the cost of ~6× slower Java read throughput.

Default remains false (good throughput). Users targeting archival or Rust-read
workloads can opt in via WriteOptions.cascading(3).withZstd(true).

Also marks PCO encode E0 gate as cleared: VortexWriter cascade is the consumer.
Next step: E1 (LeBitWriter).

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

Author: dfa1

TODO.md

@@ -9,21 +9,33 @@
   comparison via `?source=url1,url2`; then drop `.github/workflows/benchmark.yml`
 - [ ] Build something like hardwood.dev but for vortex files
 
-## Compression ratio gaps vs Rust (NYC taxi 2024-01: Java 43.1 MB, Rust 42.8 MB, Parquet 47.6 MB)
+## Compression ratio gaps vs Rust (NYC taxi 2024-01: Rust 42.8 MB, Parquet 47.6 MB)
 
 Progress: raw I64 → datetimeparts (76→53 MB), fix FOR/RLE/RunEnd accepts + sample size (53→51 MB),
-fix ConstantEncoding encodeCascade + add ZstdEncoding to CASCADE_CODECS (51→43.1 MB). Java now beats
-Parquet and is within 0.3 MB of Rust.
+fix ConstantEncoding encodeCascade (51→43.1 MB with Zstd; 49.7 MB without Zstd).
+Global dict encoding for low-cardinality integer columns (VendorID, PULocationID, etc.) — completed.
+`WriteOptions.withZstd(true)` — completed: adds ZstdEncoding to cascade, recovers the 43 MB level.
 
-Remaining 0.3 MB gap — biggest to smallest:
+Without Zstd (default, good read throughput): Java 49.7 MB — 6.5 MB above Rust.
+With Zstd (`WriteOptions.cascading(3).withZstd(true)`): Java ~43 MB — on par with Rust.
+Trade-off documented in `WriteOptions` Javadoc (6× slower Zstd decompression vs ALP+bitpack).
 
-- [ ] **Global dict encoding** — Rust applies `vortex.dict` across the ENTIRE column before chunking; produces one
-  tiny dict buffer + one globally-bitpacked codes array. Java applies dict per 131 072-row chunk. Low-cardinality
-  columns affected: `mta_tax` (8 unique F64), `Airport_fee` (4), `extra` (48), `PULocationID` (260), `DOLocationID` (
-  261),
-  `payment_type` (5), `store_and_fwd_flag` (3), `congestion_surcharge` (few), `tolls_amount` (1127).
-  Requires a two-pass write pipeline: first pass collects all values and builds the global dict; second pass emits
-  coded chunks. Estimated remaining gain ~0.3 MB (most gains already captured by ZstdEncoding on per-chunk dicts).
+Remaining gap (no-Zstd mode) — biggest to smallest:
+
+- [ ] **Nullable column handling** — `ParquetImporter` maps nulls to 0.0/0L (type defaults) for the 9 nullable F64
+  columns in the taxi dataset (`fare_amount`, `extra`, `mta_tax`, `tip_amount`, `tolls_amount`,
+  `improvement_surcharge`, `total_amount`, `congestion_surcharge`, `Airport_fee`). Rust uses `vortex.sparse`
+  or `vortex.masked` to store only valid values, then ALP on clean data. Java passes zero-polluted arrays to ALP.
+  Fix: add a `NullableData(double[] values, boolean[] validity)` wrapper; writer detects it, compacts valid values,
+  encodes validity as a Bool child. Requires API change to `VortexWriter.writeChunk` and corresponding
+  `ParquetImporter` changes.
+
+- [ ] **Global dict for F64 low-cardinality** — excluded from `isDictCandidate` because ALP/RLE were expected to
+  win; but for columns like `mta_tax` (8 unique F64 values) and `Airport_fee` (4 unique), dict codes are
+  ~same size as ALP+bitpack while Rust uses dict. Measure actual gain before implementing.
+
+- [ ] **FSST in CASCADE_CODECS** — `FsstEncoding` exists but not in the cascade; Rust uses FSST for
+  `store_and_fwd_flag`. Small gain on taxi (~0.1 MB).
 
 ## Performance
 
@@ -153,8 +165,8 @@ on any S3 fixture (all fixtures are Rust-produced; decode unblocks them).
 
 **Phases**:
 
-- [ ] **Phase E0 — gate**. Is there a consumer (CLI write path, vortex-arrow bridge)? If no,
-  stop. Decode is enough.
+- [x] **Phase E0 — gate**. Consumer = `VortexWriter` cascade (`WriteOptions.cascading(3)`) — closes
+  the NYC taxi compression gap vs Rust without the 6× Zstd read penalty. Gate cleared.
 - [ ] **Phase E1 — bit writer**. `LeBitWriter` over `Arena`-backed `MemorySegment`. Mirrors
   `pco/src/bit_writer.rs`. Property test: random bit sequences round-trip via `LeBitReader`.
 - [ ] **Phase E2 — Classic mode, no delta, fixed bins, no optimization**. Hardcoded bin layout
@@ -198,7 +210,8 @@ on any S3 fixture (all fixtures are Rust-produced; decode unblocks them).
 but suboptimal" encoder (Phase E1+E2+E5+E8 partial). Decode is the prerequisite —
 don't start before decode lands.
 
-**Decision**: keep `Encoder` stub until a real write consumer materializes. Reassess
-post-decode + post-`vortex-arrow` bridge.
+**Decision**: E0 gate cleared — start E1 (LeBitWriter). Suboptimal path E1+E2+E9 (~5 days)
+gives Rust-compatible wire format first; E3+E4 add ratio parity (~+5 days); E5+E6 add delta +
+mode selection (~+5 days). Tackle E1 next.
 
 

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

@@ -99,9 +99,18 @@ private static Path writeCsv(Path dir, List<OhlcGenerator.OhlcBatch> batches) th
     }
 
     private static Path writeJava(Path dir, List<OhlcGenerator.OhlcBatch> batches) throws IOException {
-        Path file = dir.resolve("ohlc-java.vtx");
+        return writeJava(dir, "ohlc-java.vtx", WriteOptions.cascading(3), batches);
+    }
+
+    private static Path writeJavaZstd(Path dir, List<OhlcGenerator.OhlcBatch> batches) throws IOException {
+        return writeJava(dir, "ohlc-java-zstd.vtx", WriteOptions.cascading(3).withZstd(true), batches);
+    }
+
+    private static Path writeJava(Path dir, String filename, WriteOptions opts,
+            List<OhlcGenerator.OhlcBatch> batches) throws IOException {
+        Path file = dir.resolve(filename);
         try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
-             VortexWriter writer = VortexWriter.create(ch, JAVA_SCHEMA, WriteOptions.cascading(3))) {
+             VortexWriter writer = VortexWriter.create(ch, JAVA_SCHEMA, opts)) {
             for (OhlcGenerator.OhlcBatch b : batches) {
                 writer.writeChunk(Map.of(
                         "symbol", b.symbols(), "date", b.dates(),
@@ -200,4 +209,36 @@ void fileSizeComparison(@TempDir Path tmp) throws IOException {
         }
         assertThat(totalRows).isEqualTo(TOTAL_ROWS);
     }
+
+    @Test
+    void withZstd_smallerFile_and_readable(@TempDir Path tmp) throws IOException {
+        // Given
+        List<OhlcGenerator.OhlcBatch> batches = OhlcGenerator.generate(TOTAL_ROWS, BATCH_SIZE);
+
+        // When
+        Path noZstd = writeJava(tmp, batches);
+        Path withZstd = writeJavaZstd(tmp, batches);
+
+        long noZstdSize = Files.size(noZstd);
+        long zstdSize = Files.size(withZstd);
+
+        System.out.printf(
+                "[ZstdComparison] %,d rows  noZstd=%,d bytes (%.1f MB)  withZstd=%,d bytes (%.1f MB)  ratio=%.2fx%n",
+                TOTAL_ROWS, noZstdSize, noZstdSize / 1_048_576.0,
+                zstdSize, zstdSize / 1_048_576.0,
+                (double) noZstdSize / zstdSize);
+
+        // Then — Zstd produces a smaller file for OHLC F64 data
+        assertThat(zstdSize).isLessThan(noZstdSize);
+
+        // Then — Zstd file is readable with correct row count
+        long totalRows = 0L;
+        try (VortexReader reader = VortexReader.open(withZstd, EncodingRegistry.loadAll())) {
+            var iter = reader.scan(io.github.dfa1.vortex.scan.ScanOptions.columns("volume"));
+            while (iter.hasNext()) {
+                totalRows += iter.next().<LongArray>column("volume").length();
+            }
+        }
+        assertThat(totalRows).isEqualTo(TOTAL_ROWS);
+    }
 }

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

@@ -1106,6 +1106,26 @@ void javaWriter_rustReader_listView_i64(@TempDir Path tmp) throws IOException {
         assertThat(flatElements).containsExactly(elements);
     }
 
+    @Test
+    void javaWriter_rustReader_cascading_withZstd_f64(@TempDir Path tmp) throws IOException {
+        // Given — Zstd wins the cascade for high-cardinality F64; Rust must decode vortex.zstd
+        Path file = tmp.resolve("java_zstd_cascade_f64.vtx");
+        int n = 2_000;
+        double[] data = new double[n];
+        for (int i = 0; i < n; i++) {
+            data[i] = i * 1.37 + 3.14;
+        }
+        try (var ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             var sut = VortexWriter.create(ch, F64_SCHEMA, WriteOptions.cascading(3).withZstd(true))) {
+            // When
+            sut.writeChunk(Map.of("v", data));
+        }
+
+        // Then — Rust reader must decode the file (whether via ALP or Zstd, value-equal)
+        double[] decoded = readDoubleColumn(file, "v");
+        assertBitwiseEqualsF64(decoded, data);
+    }
+
     @Test
     void javaWriter_rustReader_globalDict_i64(@TempDir Path tmp) throws IOException {
         // Given — low-cardinality I64 column triggers global dict across two chunks

performance/src/main/java/io/github/dfa1/vortex/performance/TaxiLayoutInspector.java

@@ -11,7 +11,9 @@
 import io.github.dfa1.vortex.encoding.EncodingRegistry;
 import io.github.dfa1.vortex.io.VortexInspector;
 import io.github.dfa1.vortex.io.VortexReader;
+import io.github.dfa1.vortex.parquet.ImportOptions;
 import io.github.dfa1.vortex.parquet.ParquetImporter;
+import io.github.dfa1.vortex.writer.WriteOptions;
 import org.apache.arrow.c.ArrowArray;
 import org.apache.arrow.c.ArrowSchema;
 import org.apache.arrow.c.Data;
@@ -99,6 +101,7 @@ public static void main(String[] args) throws Exception {
 
         Path jniVortex = Files.createTempFile("taxi-jni", ".vortex");
         Path javaVortex = Files.createTempFile("taxi-java", ".vortex");
+        Path javaZstdVortex = Files.createTempFile("taxi-java-zstd", ".vortex");
 
         try {
             System.out.print("Writing JNI Vortex (Rust encoder)...");
@@ -109,18 +112,30 @@ public static void main(String[] args) throws Exception {
             ParquetImporter.importParquet(parquetFile, javaVortex);
             System.out.printf(" %6.1f MB%n", mb(javaVortex));
 
+            System.out.print("Writing Java Vortex (cascading depth 3 + Zstd)...");
+            ImportOptions zstdOpts = ImportOptions.defaults()
+                    .withWriteOptions(WriteOptions.cascading(3).withZstd(true));
+            ParquetImporter.importParquet(parquetFile, javaZstdVortex, zstdOpts);
+            System.out.printf(" %6.1f MB%n", mb(javaZstdVortex));
+
             System.out.println("\n══════════════════════════════════════════════════════");
             System.out.println("  JNI / Rust writer");
             System.out.println("══════════════════════════════════════════════════════");
             inspect(jniVortex);
 
             System.out.println("\n══════════════════════════════════════════════════════");
-            System.out.println("  Java writer");
+            System.out.println("  Java writer (no Zstd)");
             System.out.println("══════════════════════════════════════════════════════");
             inspect(javaVortex);
+
+            System.out.println("\n══════════════════════════════════════════════════════");
+            System.out.println("  Java writer (with Zstd)");
+            System.out.println("══════════════════════════════════════════════════════");
+            inspect(javaZstdVortex);
         } finally {
             Files.deleteIfExists(jniVortex);
             Files.deleteIfExists(javaVortex);
+            Files.deleteIfExists(javaZstdVortex);
         }
     }
 

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

@@ -24,6 +24,7 @@
 import io.github.dfa1.vortex.encoding.RleEncoding;
 import io.github.dfa1.vortex.encoding.RunEndEncoding;
 import io.github.dfa1.vortex.encoding.VarBinEncoding;
+import io.github.dfa1.vortex.encoding.ZstdEncoding;
 import io.github.dfa1.vortex.fbs.ArraySpec;
 import io.github.dfa1.vortex.fbs.Extension;
 import io.github.dfa1.vortex.fbs.Footer;
@@ -82,25 +83,16 @@ public final class VortexWriter implements Closeable {
     private static final List<Encoding> DEFAULT_CODECS = List.of(
             new AlpEncoding(), new PrimitiveEncoding(), new BoolEncoding(), new DictEncoding(), new VarBinEncoding());
 
-    // ZstdEncoding is intentionally absent from this list.
-    // On NYC Taxi 2024-01: Zstd wins compression for F64 columns (50 MB → 43 MB) but beats
-    // ALP and kills decode throughput by 6× (240 → 40 ops/s). ZSTD decompression cannot
-    // compete with ALP reconstruction or bitpack unpack for numeric columns.
-    // Use ZstdEncoding only for Utf8/Binary where no faster structural encoding exists.
-    private static final List<Encoding> CASCADE_CODECS = List.of(
-            new DateTimePartsEncoding(),
-            new ConstantEncoding(),
-            new AlpEncoding(), new FrameOfReferenceEncoding(), new RunEndEncoding(), new RleEncoding(),
-            new DictEncoding(), new BitpackedEncoding(),
-            new VarBinEncoding(), new PrimitiveEncoding(), new BoolEncoding());
-
-    private static final EncodingRegistry CASCADE_REGISTRY = EncodingRegistry.of(CASCADE_CODECS);
+    // Base cascade codec list — no Zstd. Zstd is appended (before PrimitiveEncoding) when
+    // WriteOptions.enableZstd() is true. See WriteOptions.withZstd(boolean) for the tradeoff.
 
     private final WritableByteChannel channel;
     private final DType.Struct schema;
     private final WriteOptions options;
     private final List<Encoding> encodings;
     private final EncodingRegistry defaultRegistry;
+    private final List<Encoding> cascadeCodecs;
+    private final EncodingRegistry cascadeRegistry;
     private final List<SegRef> segs = new ArrayList<>();
     private final Map<String, List<ChunkRef>> colChunks = new LinkedHashMap<>();
     private final Map<EncodingId, Integer> encodingIdx = new LinkedHashMap<>();
@@ -121,11 +113,28 @@ private VortexWriter(
         this.options = options;
         this.encodings = encodings;
         this.defaultRegistry = EncodingRegistry.of(encodings);
+        this.cascadeCodecs = buildCascadeCodecs(options);
+        this.cascadeRegistry = EncodingRegistry.of(this.cascadeCodecs);
         for (String name : schema.fieldNames()) {
             colChunks.put(name, new ArrayList<>());
         }
     }
 
+    private static List<Encoding> buildCascadeCodecs(WriteOptions options) {
+        List<Encoding> codecs = new ArrayList<>(List.of(
+                new DateTimePartsEncoding(),
+                new ConstantEncoding(),
+                new AlpEncoding(), new FrameOfReferenceEncoding(), new RunEndEncoding(), new RleEncoding(),
+                new DictEncoding(), new BitpackedEncoding(),
+                new VarBinEncoding()));
+        if (options.enableZstd()) {
+            codecs.add(new ZstdEncoding());
+        }
+        codecs.add(new PrimitiveEncoding());
+        codecs.add(new BoolEncoding());
+        return List.copyOf(codecs);
+    }
+
     public static VortexWriter create(
             WritableByteChannel channel, DType.Struct schema, WriteOptions options
     ) {
@@ -307,8 +316,8 @@ private int writeSegment(DType dtype, Object data) throws IOException {
         try (Arena arena = Arena.ofConfined()) {
             EncodeResult result;
             if (options.allowedCascading() > 0) {
-                EncodeContext encodeCtx = EncodeContext.ofDepth(options.allowedCascading(), arena, CASCADE_REGISTRY);
-                CascadingCompressor compressor = new CascadingCompressor(CASCADE_CODECS);
+                EncodeContext encodeCtx = EncodeContext.ofDepth(options.allowedCascading(), arena, cascadeRegistry);
+                CascadingCompressor compressor = new CascadingCompressor(cascadeCodecs);
                 result = compressor.encode(dtype, data, encodeCtx);
             } else {
                 Encoding encoding = findEncoding(dtype);

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

@@ -1,29 +1,62 @@
 package io.github.dfa1.vortex.writer;
 
 /// Tuning knobs for the Vortex writer.
+///
+/// @param chunkSize                 target row count per chunk (default 65 536)
+/// @param enableZoneMaps            write per-chunk min/max statistics for zone-map pruning
+/// @param compressionRatioThreshold minimum compression ratio for an encoding to be accepted (0–1)
+/// @param allowedCascading          maximum recursive cascade depth; 0 = no cascading
+/// @param globalDict                build one shared dictionary across all chunks for low-cardinality columns
+/// @param enableZstd                add Zstandard to the cascade codec list
+///                                  <p>When {@code true}, Zstd competes with structural encodings (ALP, bitpack, etc.)
+///                                  on every chunk and wins when it produces smaller output — typically reducing
+///                                  file size by 10–15% on real-world datasets compared to ALP+bitpack alone.
+///                                  Trade-off: Zstd decompression is ~6× slower than ALP decode;
+///                                  prefer the default ({@code false}) for read-heavy workloads.
 public record WriteOptions(
         int chunkSize,
         boolean enableZoneMaps,
         double compressionRatioThreshold,
         int allowedCascading,
-        boolean globalDict
+        boolean globalDict,
+        boolean enableZstd
 ) {
-    /// Default options: global dictionary encoding enabled, no cascading compression.
+    /// Default options: global dictionary encoding enabled, no cascading compression, Zstd disabled.
+    ///
+    /// @return default {@code WriteOptions}
     public static WriteOptions defaults() {
-        return new WriteOptions(65_536, true, 0.90, 0, true);
+        return new WriteOptions(65_536, true, 0.90, 0, true, false);
     }
 
     /// Enable cascading compression with up to {@code depth} recursive levels.
     /// Depth 0 preserves current first-match behaviour.
+    ///
+    /// @param depth maximum cascade depth
+    /// @return {@code WriteOptions} with cascading enabled at the given depth
     public static WriteOptions cascading(int depth) {
-        return new WriteOptions(65_536, true, 0.90, depth, true);
+        return new WriteOptions(65_536, true, 0.90, depth, true, false);
     }
 
     /// Returns a copy of these options with global dictionary encoding set to {@code enabled}.
     ///
     /// @param enabled {@code true} to enable global dictionary encoding across chunks
     /// @return a new {@code WriteOptions} with the global dict flag updated
     public WriteOptions withGlobalDict(boolean enabled) {
-        return new WriteOptions(chunkSize, enableZoneMaps, compressionRatioThreshold, allowedCascading, enabled);
+        return new WriteOptions(chunkSize, enableZoneMaps, compressionRatioThreshold, allowedCascading, enabled, enableZstd);
+    }
+
+    /// Returns a copy of these options with Zstandard compression set to {@code enabled}.
+    ///
+    /// <p>When enabled, Zstd is added to the cascade codec list and competes with structural encodings
+    /// (ALP, bitpack, FOR, etc.) on every chunk. Zstd typically wins on high-cardinality numeric columns
+    /// (e.g. {@code fare_amount}, {@code total_amount}), reducing file size by 10–15%.
+    ///
+    /// <p>Trade-off: Zstd decompression is ~6× slower than ALP reconstruction or bitpack unpack.
+    /// Use {@code false} (the default) for read-heavy or latency-sensitive workloads.
+    ///
+    /// @param enabled {@code true} to enable Zstd in the compression cascade
+    /// @return a new {@code WriteOptions} with the Zstd flag updated
+    public WriteOptions withZstd(boolean enabled) {
+        return new WriteOptions(chunkSize, enableZoneMaps, compressionRatioThreshold, allowedCascading, globalDict, enabled);
     }
 }