fix(reader): make constant Utf8/Binary decode lazy

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                                                   |

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<MemorySegment> 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.
     ///

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) {

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<Integer> 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");
+        }
+    }
 }

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});