diff --git a/AGENTS.md b/AGENTS.md index bae6089ca..fe209be52 100644 --- a/AGENTS.md +++ b/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//`: seeds plus every past finding) diff --git a/CHANGELOG.md b/CHANGELOG.md index 221c37b29..bdef29f6a 100644 --- a/CHANGELOG.md +++ b/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)) diff --git a/integration/pom.xml b/integration/pom.xml index de82d9e91..fa559c3cb 100644 --- a/integration/pom.xml +++ b/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. --> - raincloud,slow,fuzz + raincloud,slow,fuzz,large @@ -143,7 +145,7 @@ - raincloud,fuzz + raincloud,fuzz,large diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/FsstLargeChunkInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/FsstLargeChunkInteropIntegrationTest.java new file mode 100644 index 000000000..c08a05bd6 --- /dev/null +++ b/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 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 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); + } + } +} diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/SerializedArrayDecoder.java b/reader/src/main/java/io/github/dfa1/vortex/reader/SerializedArrayDecoder.java index 25723b5a6..19b8f854e 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/SerializedArrayDecoder.java +++ b/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 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); diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java b/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java index 83a6fb0a2..0f6aec60b 100644 --- a/writer/src/main/java/io/github/dfa1/vortex/writer/VortexWriter.java +++ b/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 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 { diff --git a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/FsstEncodingEncoder.java b/writer/src/main/java/io/github/dfa1/vortex/writer/encode/FsstEncodingEncoder.java index 35d9b1654..c1da9ef9d 100644 --- a/writer/src/main/java/io/github/dfa1/vortex/writer/encode/FsstEncodingEncoder.java +++ b/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