fix: write and read a chunk whose encoded bytes pass 2 GiB

AGENTS.md

@@ -118,6 +118,9 @@ Tag-excluded from a routine local build, even when named with `-Dit.test`/`-Dtes
 - **Integration fuzz targets** (`@Tag("fuzz")` in `integration`, failsafe): replaying them floods
   the log with Jazzer instrumentation output. Clear the exclusion with
   `-Dvortex.it.excludedGroups=`. Not run by the `ci` profile; the weekly `Fuzz` workflow runs them.
+- **Large integration tests** (`@Tag("large")`, failsafe): data past 2 GiB in one chunk, several GB
+  of memory, too big for CI runners. Clear the exclusion with `-Dvortex.it.excludedGroups=`. Not
+  run by the `ci` profile.
 
 Jazzer fuzz targets in the `fuzz` module (`@FuzzTest`, ADR 0020) are **not** excluded: without
 `JAZZER_FUZZ=1` each replays its saved inputs (`*Inputs/<method>/`: seeds plus every past finding)

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Fixed
+- A column whose encoded bytes pass 2 GiB in one chunk, such as large audio blobs under FSST, writes and reads back instead of failing with an integer overflow. ([#629](https://github.com/dfa1/vortex-java/issues/629))
 - Parquet import reads a `DECIMAL` column as a Vortex decimal with its precision and scale, in all four physical encodings; `INT32`/`INT64` ones lost their scale and the byte-array ones failed. ([#628](https://github.com/dfa1/vortex-java/issues/628))
 - Parquet import keeps the nulls of a nullable `TIMESTAMP` column instead of failing as if the column were `REQUIRED`. ([#627](https://github.com/dfa1/vortex-java/issues/627))
 

integration/pom.xml

@@ -21,9 +21,11 @@
              - `fuzz`: the Jazzer write->read targets (WriteReadFuzzIntegrationTest). Replaying
                their saved inputs floods the log with Jazzer instrumentation output, so they run
                only when asked for, and in the weekly Fuzz workflow.
+             - `large`: tests whose data passes 2 GiB in one chunk (several GB of memory), too big
+               for CI runners, so they run only when asked for.
              Override on the CLI, e.g. `-Dvortex.it.excludedGroups=` (empty) runs every tag;
              a literal pom value could not be cleared this way. -->
-        <vortex.it.excludedGroups>raincloud,slow,fuzz</vortex.it.excludedGroups>
+        <vortex.it.excludedGroups>raincloud,slow,fuzz,large</vortex.it.excludedGroups>
     </properties>
 
     <dependencies>
@@ -143,7 +145,7 @@
                 </property>
             </activation>
             <properties>
-                <vortex.it.excludedGroups>raincloud,fuzz</vortex.it.excludedGroups>
+                <vortex.it.excludedGroups>raincloud,fuzz,large</vortex.it.excludedGroups>
             </properties>
         </profile>
         <profile>

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

@@ -0,0 +1,137 @@
+package io.github.dfa1.vortex.integration;
+
+import dev.vortex.api.DataSource;
+import dev.vortex.api.Expression;
+import dev.vortex.api.Partition;
+import dev.vortex.api.Scan;
+import dev.vortex.api.ScanOptions;
+import dev.vortex.api.Session;
+import dev.vortex.arrow.ArrowAllocation;
+import dev.vortex.jni.NativeLoader;
+import io.github.dfa1.vortex.core.model.ColumnName;
+import io.github.dfa1.vortex.core.model.DType;
+import io.github.dfa1.vortex.core.model.EncodingId;
+import io.github.dfa1.vortex.reader.Chunk;
+import io.github.dfa1.vortex.reader.ScanIterator;
+import io.github.dfa1.vortex.reader.VortexReader;
+import io.github.dfa1.vortex.reader.array.VarBinArray;
+import io.github.dfa1.vortex.writer.ColumnEncoding;
+import io.github.dfa1.vortex.writer.VortexWriter;
+import io.github.dfa1.vortex.writer.WriteOptions;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.junit.jupiter.api.Tag;
+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.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/// An FSST column whose compressed codes pass 2 GiB in one chunk writes and reads back (#629): the
+/// code offsets go 64-bit, as Rust's `fsst_compress` switches to `i64` offsets past `i32::MAX`; the
+/// segment, whose length is a `u32` on the wire, is written past a `ByteBuffer`'s 2 GiB and read
+/// back as one. Found on Raincloud's `peoples-speech-clean-validation`, whose audio column is
+/// ~2.3 GB in one 8192-row block.
+///
+/// FSST only wins on bytes it can shrink, so the rows compress slightly (a repeated word in a tenth
+/// of their 8-byte slots, the rest bytes FSST codes one for one) and ~2.46 GB of input is needed;
+/// every row shares one array to keep the heap small. Tagged `large`: several GB of memory.
+@Tag("large")
+class FsstLargeChunkInteropIntegrationTest {
+
+    private static final Session SESSION = Session.create();
+    private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator();
+    private static final ColumnName BLOB = ColumnName.of("blob");
+    /// One 8192-row block: the writer's repartition never splits it.
+    private static final int ROWS = 8192;
+    private static final int ROW_BYTES = 300_000;
+
+    static {
+        NativeLoader.loadJni();
+    }
+
+    @Test
+    void codesPastTwoGibibytes_roundTripAndJniReads(@TempDir Path tmp) throws Exception {
+        // Given
+        byte[] row = slightlyCompressibleRow();
+        byte[][] rows = new byte[ROWS][];
+        Arrays.fill(rows, row);
+        Path file = tmp.resolve("fsst.vortex");
+        WriteOptions options = WriteOptions.defaults()
+                .withColumnEncoding(BLOB, ColumnEncoding.candidates(EncodingId.VORTEX_FSST));
+
+        // When
+        DType.Struct schema = new DType.Struct(List.of(BLOB), List.of(new DType.Binary(false)), false);
+        try (FileChannel channel = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE);
+             VortexWriter writer = VortexWriter.create(channel, schema, options)) {
+            writer.writeChunk(Map.of(BLOB, rows));
+        }
+
+        // Then — FSST was kept and its codes passed 2 GiB; the last rows, whose codes start past it,
+        // decode to the input in vortex-jni and vortex-java
+        assertThat(file.toFile().length()).isGreaterThan(Integer.MAX_VALUE).isLessThan((long) ROWS * ROW_BYTES);
+        assertThat(readTailWithJni(file, ROWS - 4, ROWS)).hasSize(4).allSatisfy(b -> assertThat(b).isEqualTo(row));
+        assertThat(readLastWithJava(file)).isEqualTo(row);
+    }
+
+    private static byte[] slightlyCompressibleRow() {
+        byte[] word = "AUDIOHDR".getBytes(StandardCharsets.US_ASCII);
+        Random random = new Random(629);
+        byte[] row = new byte[ROW_BYTES];
+        for (int at = 0; at < ROW_BYTES; at += word.length) {
+            if (random.nextInt(10) == 0) {
+                System.arraycopy(word, 0, row, at, word.length);
+            } else {
+                for (int i = at; i < at + word.length; i++) {
+                    row[i] = (byte) random.nextInt(200);
+                }
+            }
+        }
+        return row;
+    }
+
+    private static List<byte[]> readTailWithJni(Path file, long begin, long end) throws IOException {
+        ScanOptions opts = ScanOptions.builder()
+                .projection(Expression.select(new String[]{"blob"}, Expression.root()))
+                .rowRangeBegin(begin)
+                .rowRangeEnd(end)
+                .build();
+        List<byte[]> out = new ArrayList<>();
+        DataSource ds = DataSource.open(SESSION, file.toAbsolutePath().toUri().toString());
+        Scan scan = ds.scan(opts);
+        while (scan.hasNext()) {
+            Partition partition = scan.next();
+            try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) {
+                while (reader.loadNextBatch()) {
+                    VectorSchemaRoot root = reader.getVectorSchemaRoot();
+                    FieldVector vec = root.getVector("blob");
+                    for (int i = 0; i < root.getRowCount(); i++) {
+                        out.add((byte[]) vec.getObject(i));
+                    }
+                }
+            }
+        }
+        return out;
+    }
+
+    private static byte[] readLastWithJava(Path file) throws IOException {
+        try (VortexReader reader = VortexReader.open(file);
+             ScanIterator iter = reader.scan(io.github.dfa1.vortex.reader.ScanOptions.all());
+             Chunk chunk = iter.next()) {
+            VarBinArray blob = chunk.column(BLOB);
+            return blob.getBytes(chunk.rowCount() - 1);
+        }
+    }
+}

reader/src/main/java/io/github/dfa1/vortex/reader/SerializedArrayDecoder.java

@@ -49,7 +49,8 @@ public SerializedArrayDecoder(ReadRegistry registry) {
     /// @return the decoded [Array] for this segment
     public Array decode(MemorySegment seg, List<EncodingId> encodingSpecs,
             DType dtype, long rowCount, SegmentAllocator arena) {
-        int segLen = IoBounds.toIntSize(seg.byteSize());
+        // long: a segment's length is a u32 on the wire, so it may pass 2 GiB (#629)
+        long segLen = seg.byteSize();
 
         // The trailing u32 length field must itself be in range before we read it.
         IoBounds.checkRange(segLen - 4L, 4, segLen);

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

@@ -110,6 +110,9 @@ public final class VortexWriter implements Closeable {
     // values() clones its array on every call; the edition filter reads it once per writer.
     private static final List<EncodingId.WellKnown> WELL_KNOWN_IDS = List.of(EncodingId.WellKnown.values());
 
+    /// Largest slice of a segment written at once: a ByteBuffer holds at most 2 GiB.
+    private static final long MAX_WRITE_SLICE = 1L << 30;
+
     private static final int LAYOUT_FLAT = 0;
     private static final int LAYOUT_CHUNKED = 1;
     private static final int LAYOUT_STRUCT = 2;
@@ -1047,11 +1050,16 @@ private EncodingEncoder nullableCapableEncoder(DType dtype) {
     }
 
     private void write(MemorySegment seg) throws IOException {
-        ByteBuffer buf = seg.asByteBuffer();
-        while (buf.hasRemaining()) {
-            channel.write(buf);
+        // A segment may pass 2 GiB (a u32 length on the wire, #629) but a ByteBuffer cannot, so the
+        // segment is written in slices that each fit one
+        long size = seg.byteSize();
+        for (long at = 0; at < size; at += MAX_WRITE_SLICE) {
+            ByteBuffer buf = seg.asSlice(at, Math.min(MAX_WRITE_SLICE, size - at)).asByteBuffer();
+            while (buf.hasRemaining()) {
+                channel.write(buf);
+            }
         }
-        bytesWritten += seg.byteSize();
+        bytesWritten += size;
     }
 
     private void write(ByteBuffer buf) throws IOException {

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

@@ -128,7 +128,7 @@ public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext ctx) {
         Fsst c = compress(data, rows, ctx.arena());
         MinMax stats = VarBinEncodingEncoder.textMinMax(dtype, data, rows);
         Object uncompLens = PrimitiveArrays.fromIntsArray(c.uncompLens(), c.uncompLenPType(), EncodingId.VORTEX_FSST);
-        Object codesOffsets = PrimitiveArrays.fromIntsArray(c.codesOffsets(), c.codesOffPType(), EncodingId.VORTEX_FSST);
+        Object codesOffsets = PrimitiveArrays.fromLongsArray(c.codesOffsets(), c.codesOffPType(), EncodingId.VORTEX_FSST);
         EncodeNode partialRoot = new EncodeNode(
                 EncodingId.VORTEX_FSST,
                 MemorySegment.ofArray(c.metaBytes()),
@@ -143,12 +143,12 @@ public CascadeStep encodeCascade(DType dtype, Object data, EncodeContext ctx) {
     }
 
     /// The FSST-specific product of compression: the symbol-table buffers, the wire code stream, the
-    /// [ProtoFSSTMetadata] bytes, and the per-row uncompressed lengths / cumulative code offsets as
-    /// plain `int[]` (the two offset children, in narrowest-unsigned ptypes).
+    /// [ProtoFSSTMetadata] bytes, the per-row uncompressed lengths as `int[]` and the cumulative code
+    /// offsets as `long[]` (the two offset children, in narrowest-unsigned ptypes).
     @SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared.
     record Fsst(
             MemorySegment symBuf, MemorySegment symLenBuf, MemorySegment compBuf,
-            byte[] metaBytes, int[] uncompLens, int[] codesOffsets,
+            byte[] metaBytes, int[] uncompLens, long[] codesOffsets,
             PType uncompLenPType, PType codesOffPType, int n) {
     }
 
@@ -254,11 +254,13 @@ private static Fsst finishCompress(Compressor trained, int n, long totalInput, i
         // Written directly at wire position: codesOffsets[0] stays its default 0, and each row's
         // end lands at codesOffsets[i + 1] as it compresses, so no separate rowEnds array (and no
         // copy into a codesOffsets array afterward) is needed.
-        int[] codesOffsets = new int[n + 1];
+        // long, as Rust's fsst_compress takes i64 offsets once the codes may pass i32::MAX (#629):
+        // a chunk of large, barely compressible rows (audio, images) does.
+        long[] codesOffsets = new long[n + 1];
         long totalCompressed = 0;
         for (int i = 0; i < n; i++) {
             totalCompressed = rowCompressor.compress(compressor, i, scratch, totalCompressed);
-            codesOffsets[i + 1] = Math.toIntExact(totalCompressed);
+            codesOffsets[i + 1] = totalCompressed;
         }
 
         // A slice, not a copy: scratch is already arena-owned, so trimming the worst-case 2x