From 83f9e2e09af0d24863e20873bed056e88c109797 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Thu, 6 Aug 2026 09:01:27 +0200 Subject: [PATCH] fix(reader): make constant Utf8/Binary decode lazy (#329) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ConstantEncodingDecoder.decodeString was the one vortex.constant value type without a LazyConstantXxxArray: every other type (primitive, bool, decimal) broadcasts a single stored value per row in O(1), but strings eagerly wrote n copies of the scalar into a real OffsetMode buffer (n * strLen allocation + copy), the only crash-adjacent risk that survived the RunEnd/Constant/Zoned/Pco hardening batch (a large n * strLen product could overflow negative or just OOM). Root cause: VarBinArray's only flat representation, OffsetMode, has no broadcast/modulo path the way AbstractMaterializedArray gives the primitive Materialized*Array types. Added VarBinArray.ConstantMode: a sealed-permitted record holding the scalar's bytes once, with every accessor returning it for any row — bytesSegment()/segmentIfPresent() follow the same "no single contiguous buffer" convention already used by ChunkedMode/ViewMode, so generic consumers still flatten it correctly via VarBinArray.toOffsetMode(). decodeString now builds this directly instead of an eager OffsetMode. docs/compatibility.md: vortex.constant's Notes column now credits VarBinArray.ConstantMode alongside LazyConstantXxxArray. --- docs/compatibility.md | 2 +- .../dfa1/vortex/reader/array/VarBinArray.java | 71 +++++++++-- .../decode/ConstantEncodingDecoder.java | 28 ++--- .../vortex/reader/array/VarBinArrayTest.java | 115 ++++++++++++++++++ .../decode/ConstantEncodingDecoderTest.java | 37 ++++++ 5 files changed, 228 insertions(+), 25 deletions(-) diff --git a/docs/compatibility.md b/docs/compatibility.md index d4a0c31c..0dc512d1 100644 --- a/docs/compatibility.md +++ b/docs/compatibility.md @@ -136,7 +136,7 @@ decoder falls into one of three shapes: | `vortex.null` | n/a | n/a | no per-row data | | `vortex.bytebool` | Zero-copy | Zero-copy | mmap slice | | `vortex.zigzag` | Lazy | Lazy | `LazyZigZagXxxArray` (I8/I16/I32/I64); broadcast → `LazyConstantXxxArray`, ADR 0010 + 0015 | -| `vortex.constant` | Lazy | Lazy | `LazyConstantXxxArray` (primitive + bool + decimal); per-row broadcast, no buffer, ADR 0015 | +| `vortex.constant` | Lazy | Lazy | `LazyConstantXxxArray` (primitive + bool + decimal) + `VarBinArray.ConstantMode` (Utf8/Binary); per-row broadcast, no buffer, ADR 0015 | | `vortex.ext` | Zero-copy | Zero-copy | wraps storage | | `vortex.runend` | Lazy | Lazy | `LazyRunEndXxxArray` (primitive + bool); Utf8/Binary stays Materialized (offset rebasing), ADR 0015 | | `vortex.varbin` | Zero-copy | Zero-copy | bytes + offsets slices | diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java index 95fc822a..fd1ff6e8 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java @@ -9,20 +9,21 @@ import java.lang.foreign.SegmentAllocator; import java.lang.foreign.ValueLayout; import java.nio.charset.StandardCharsets; +import java.util.Objects; import java.util.Optional; import java.util.function.IntConsumer; /// Sealed interface for variable-length binary / UTF-8 string columns. /// -/// Four implementations: [OffsetMode] for standard offset-based layout, -/// [DictMode] for dictionary-encoded strings, [ChunkedMode] for -/// multi-chunk columns, and [ViewMode] for Arrow StringView / BinaryView -/// layout (16-byte view per row + zero or more shared data buffers). All -/// accessors resolve transparently regardless of mode; only [OffsetMode] -/// exposes [OffsetMode#offsetsSegment()] and [OffsetMode#offsetsPtype()]. +/// Implementations: [OffsetMode] for standard offset-based layout, [DictMode] for +/// dictionary-encoded strings, [ChunkedMode] for multi-chunk columns, [ViewMode] for Arrow +/// StringView / BinaryView layout (16-byte view per row + zero or more shared data buffers), +/// [SlicedMode] for a zero-copy row-shifted view, and [ConstantMode] for a `vortex.constant` +/// value broadcast across every row. All accessors resolve transparently regardless of mode; +/// only [OffsetMode] exposes [OffsetMode#offsetsSegment()] and [OffsetMode#offsetsPtype()]. public sealed interface VarBinArray extends Array permits VarBinArray.OffsetMode, VarBinArray.DictMode, VarBinArray.ChunkedMode, - VarBinArray.ViewMode, VarBinArray.SlicedMode { + VarBinArray.ViewMode, VarBinArray.SlicedMode, VarBinArray.ConstantMode { /// Sliced view over a [VarBinArray]: every accessor delegates to `inner` /// with the row index shifted by `offset`. Used by the scan iterator to @@ -82,6 +83,62 @@ public VarBinArray limited(long rows) { } } + /// Metadata-only mode for `vortex.constant` Utf8/Binary columns: a single value broadcast + /// across `length` logical rows. No buffer is allocated and no per-row materialization + /// occurs — every accessor returns the same shared bytes, mirroring the + /// `LazyConstantXxxArray` family the primitive array types use for the same encoding. + /// + /// @param dtype logical element type (Utf8 or Binary) + /// @param length number of logical rows the constant is broadcast across + /// @param bytes the constant value's raw bytes, shared across every row; [#getBytes(long)] + /// clones it per that method's copy contract + @SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared. + record ConstantMode(DType dtype, long length, byte[] bytes) implements VarBinArray { + + @Override + public MemorySegment bytesSegment() { + return MemorySegment.NULL; + } + + /// No single contiguous segment backs a broadcast constant. + /// + /// @return always empty + @Override + public Optional segmentIfPresent() { + return Optional.empty(); + } + + @Override + public byte[] getBytes(long i) { + Objects.checkIndex(i, length); + return bytes.clone(); + } + + @Override + public String getString(long i) { + Objects.checkIndex(i, length); + return new String(bytes, StandardCharsets.UTF_8); + } + + @Override + public int getByteLength(long i) { + Objects.checkIndex(i, length); + return bytes.length; + } + + @Override + public void forEachByteLength(IntConsumer c) { + int len = bytes.length; + for (long i = 0; i < length; i++) { + c.accept(len); + } + } + + @Override + public VarBinArray limited(long rows) { + return rows >= length ? this : new ConstantMode(dtype, rows, bytes); + } + } /// Returns the concatenated raw bytes segment backing all elements. /// diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoder.java b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoder.java index d4e82d99..a0856977 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoder.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoder.java @@ -4,7 +4,6 @@ import io.github.dfa1.vortex.core.model.PType; import io.github.dfa1.vortex.core.error.VortexException; import io.github.dfa1.vortex.core.model.EncodingId; -import io.github.dfa1.vortex.core.io.VortexFormat; import io.github.dfa1.vortex.core.proto.ProtoScalarValue; import io.github.dfa1.vortex.reader.array.Array; import io.github.dfa1.vortex.reader.array.LazyConstantBoolArray; @@ -66,7 +65,7 @@ private static Array arrayFromScalar(DecodeContext ctx, ProtoScalarValue scalar, return arrayFromScalar(ctx, inner.value(), innerDtype, n); } if (dtype instanceof DType.Utf8 || dtype instanceof DType.Binary) { - return decodeString(ctx, scalar, dtype, n); + return decodeString(scalar, dtype, n); } if (dtype instanceof DType.Bool) { return decodeBool(dtype, scalar, n); @@ -133,24 +132,19 @@ private static Array decodeBool(DType dtype, ProtoScalarValue scalar, long n) { return new LazyConstantBoolArray(dtype, n, value); } - private static Array decodeString(DecodeContext ctx, ProtoScalarValue scalar, DType dtype, long n) { + /// Builds a metadata-only constant Utf8/Binary array — [VarBinArray.ConstantMode] returns + /// the same shared bytes for every row, so this stays O(1) regardless of `n` like every + /// other constant type, instead of eagerly writing `n` copies into a real buffer. + /// + /// @param scalar the constant scalar value + /// @param dtype logical dtype (Utf8 or Binary) + /// @param n row count + /// @return a lazy constant array of length `n` + private static Array decodeString(ProtoScalarValue scalar, DType dtype, long n) { byte[] strBytes = scalar.string_value() != null ? scalar.string_value().getBytes(StandardCharsets.UTF_8) : (scalar.bytes_value() != null ? scalar.bytes_value() : new byte[0]); - - int strLen = strBytes.length; - - MemorySegment bytesSeg = ctx.arena().allocate(n * strLen); - for (long i = 0; i < n; i++) { - MemorySegment.copy(MemorySegment.ofArray(strBytes), 0L, bytesSeg, i * strLen, strLen); - } - - MemorySegment offsetsSeg = ctx.arena().allocate((n + 1) * 4L, 4); - for (long i = 0; i <= n; i++) { - offsetsSeg.setAtIndex(VortexFormat.LE_INT, i, (int) (i * strLen)); - } - - return new VarBinArray.OffsetMode(dtype, n, bytesSeg.asReadOnly(), offsetsSeg.asReadOnly(), PType.I32); + return new VarBinArray.ConstantMode(dtype, n, strBytes); } private static long scalarToRawBits(ProtoScalarValue scalar) { diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java index fd7003c7..561d44ee 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java @@ -8,6 +8,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import java.lang.foreign.Arena; import java.lang.foreign.MemorySegment; import java.nio.ByteBuffer; import java.nio.ByteOrder; @@ -322,4 +323,118 @@ private static MemorySegment packOffsets(int[] offs, PType ptype) { } } + + @Nested + class Constant { + + @Test + void getString_returnsSameValueForEveryRow() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 5, "hi".getBytes(StandardCharsets.UTF_8)); + + // When / Then + assertThat(sut.getString(0)).isEqualTo("hi"); + assertThat(sut.getString(4)).isEqualTo("hi"); + } + + @Test + void getByteLength_returnsConstantLength() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 3, "abc".getBytes(StandardCharsets.UTF_8)); + + // When / Then + assertThat(sut.getByteLength(0)).isEqualTo(3); + assertThat(sut.getByteLength(2)).isEqualTo(3); + } + + @Test + void forEachByteLength_visitsLengthOncePerRow() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 4, "xy".getBytes(StandardCharsets.UTF_8)); + List lengths = new ArrayList<>(); + + // When + sut.forEachByteLength(lengths::add); + + // Then + assertThat(lengths).containsExactly(2, 2, 2, 2); + } + + /// [VarBinArray#getBytes(long)]'s contract is a copy per call; a shared backing array + /// broadcast across every row must not let one row's mutation leak into another's. + @Test + void getBytes_returnsIndependentCopyEachCall() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 2, "z".getBytes(StandardCharsets.UTF_8)); + + // When + byte[] first = sut.getBytes(0); + first[0] = (byte) 'Q'; + + // Then — mutating the returned copy must not corrupt subsequent reads + assertThat(sut.getBytes(1)).isEqualTo("z".getBytes(StandardCharsets.UTF_8)); + } + + @Test + void getString_outOfBoundsIndex_throws() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 2, "v".getBytes(StandardCharsets.UTF_8)); + + // When / Then + assertThatThrownBy(() -> sut.getString(2)).isInstanceOf(IndexOutOfBoundsException.class); + } + + @Test + void limited_returnsShorterConstant() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 10, "k".getBytes(StandardCharsets.UTF_8)); + + // When + VarBinArray result = sut.limited(3); + + // Then + assertThat(result.length()).isEqualTo(3); + assertThat(result.getString(2)).isEqualTo("k"); + } + + @Test + void limited_rowsAtOrAboveLength_returnsSameInstance() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 5, "m".getBytes(StandardCharsets.UTF_8)); + + // When + VarBinArray result = sut.limited(5); + + // Then + assertThat(result).isSameAs(sut); + } + + @Test + void bytesSegment_isNull_noContiguousBufferBacksABroadcast() { + // Given + VarBinArray.ConstantMode sut = new VarBinArray.ConstantMode(UTF8, 2, "n".getBytes(StandardCharsets.UTF_8)); + + // When / Then + assertThat(sut.bytesSegment()).isSameAs(MemorySegment.NULL); + assertThat(sut.segmentIfPresent()).isEmpty(); + } + + /// The general "any VarBinArray -> OffsetMode" path other decoders (RunEnd's string + /// expansion, dict/sparse child normalization) rely on must still work for a broadcast + /// constant, walking it via the typed accessors rather than `bytesSegment()`. + @Test + void toOffsetMode_materializesBroadcastIntoRealOffsets() { + // Given + VarBinArray sut = new VarBinArray.ConstantMode(UTF8, 3, "hey".getBytes(StandardCharsets.UTF_8)); + + // When + VarBinArray.OffsetMode result = VarBinArray.toOffsetMode(sut, Arena.ofAuto()); + + // Then + assertThat(result.length()).isEqualTo(3); + assertThat(result.getString(0)).isEqualTo("hey"); + assertThat(result.getString(1)).isEqualTo("hey"); + assertThat(result.getString(2)).isEqualTo("hey"); + } + } } diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoderTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoderTest.java index 578c46bf..29df2631 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoderTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/ConstantEncodingDecoderTest.java @@ -7,6 +7,7 @@ import io.github.dfa1.vortex.reader.ReadRegistry; import io.github.dfa1.vortex.reader.array.Array; import io.github.dfa1.vortex.reader.array.LongArray; +import io.github.dfa1.vortex.reader.array.VarBinArray; import org.junit.jupiter.api.Test; import java.lang.foreign.Arena; @@ -70,6 +71,42 @@ void decimalScalar_withBytesValue_decodes() { assertThat(result.length()).isEqualTo(2); } + @Test + void stringScalar_decodesToLazyConstantVarBin() { + // Given + ProtoScalarValue scalar = new ProtoScalarValue( + null, null, null, null, null, null, "hi", null, null, null, null); + + // When + Array result = decode(scalar, DType.UTF8, 3); + + // Then — VarBinArray.ConstantMode, not an eagerly materialized OffsetMode + assertThat(result).isInstanceOf(VarBinArray.ConstantMode.class); + VarBinArray strings = (VarBinArray) result; + assertThat(strings.getString(0)).isEqualTo("hi"); + assertThat(strings.getString(2)).isEqualTo("hi"); + } + + /// Row count no longer bounds the work `decodeString` does: it used to eagerly allocate + /// and copy `n` string repetitions into a real buffer (`n * strLen`), which for a large + /// `n` was both wasted work and an integer-overflow/OOM risk (#329). `ConstantMode` is + /// O(1) regardless of `n`, so a row count too large to ever materialize still decodes + /// instantly. + @Test + void stringScalar_hugeRowCount_decodesWithoutAllocating() { + // Given — a row count that would demand petabytes if eagerly materialized + ProtoScalarValue scalar = new ProtoScalarValue( + null, null, null, null, null, null, "x", null, null, null, null); + long hugeRowCount = Long.MAX_VALUE / 2; + + // When + Array result = decode(scalar, DType.UTF8, hugeRowCount); + + // Then + assertThat(result.length()).isEqualTo(hugeRowCount); + assertThat(((VarBinArray) result).getString(0)).isEqualTo("x"); + } + private static Array decode(ProtoScalarValue scalar, DType dtype, long n) { MemorySegment scalarBuf = MemorySegment.ofArray(scalar.encode()); ArrayNode node = new ArrayNode(EncodingId.VORTEX_CONSTANT, null, new ArrayNode[0], new int[]{0});