diff --git a/paimon-common/src/main/java/org/apache/paimon/data/serializer/BinaryRowSerializer.java b/paimon-common/src/main/java/org/apache/paimon/data/serializer/BinaryRowSerializer.java index dbf37f520f4e..78dbc9f0d0c5 100644 --- a/paimon-common/src/main/java/org/apache/paimon/data/serializer/BinaryRowSerializer.java +++ b/paimon-common/src/main/java/org/apache/paimon/data/serializer/BinaryRowSerializer.java @@ -261,9 +261,26 @@ private void pointToMultiSegments( * binary row fixed part. See {@link BinaryRow}. */ private int checkSkipWriteForFixLengthPart(AbstractPagedOutputView out) throws IOException { + // The fixed-length part of a BinaryRow must reside within a single memory segment, + // because random field access (e.g. getLong/getString) always reads from segments[0] + // without bounds checking. If it is larger than the page size, advancing to the next + // segment cannot help and the row would silently span segments, leading to corrupted + // field offsets and eventually a NegativeArraySizeException. Fail fast instead. + int fixedPartLength = getSerializedRowFixedPartLength(); + int segmentSize = out.getSegmentSize(); + checkArgument( + fixedPartLength <= segmentSize, + "Cannot serialize BinaryRow to pages: the serialized fixed-length part (%d bytes, " + + "numFields=%d) is larger than the page size (%d bytes). The fixed-length " + + "part must fit within a single page. Please increase the 'page-size' option " + + "to at least %d bytes.", + fixedPartLength, + numFields, + segmentSize, + fixedPartLength); // skip if there is no enough size. - int available = out.getSegmentSize() - out.getCurrentPositionInSegment(); - if (available < getSerializedRowFixedPartLength()) { + int available = segmentSize - out.getCurrentPositionInSegment(); + if (available < fixedPartLength) { out.advance(); return available; } diff --git a/paimon-common/src/test/java/org/apache/paimon/data/serializer/BinaryRowSerializerPageSizeGuardTest.java b/paimon-common/src/test/java/org/apache/paimon/data/serializer/BinaryRowSerializerPageSizeGuardTest.java new file mode 100644 index 000000000000..44f15da20ff2 --- /dev/null +++ b/paimon-common/src/test/java/org/apache/paimon/data/serializer/BinaryRowSerializerPageSizeGuardTest.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.data.serializer; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.BinaryRowWriter; +import org.apache.paimon.data.RandomAccessInputView; +import org.apache.paimon.data.SimpleCollectingOutputView; +import org.apache.paimon.memory.CachelessSegmentPool; +import org.apache.paimon.memory.MemorySegment; + +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests for the page-size guard in {@link BinaryRowSerializer#serializeToPages}. When a row's + * fixed-length part is larger than the page size, serialization must fail fast with a clear error + * instead of silently spanning segments and corrupting field access (which previously surfaced as a + * cryptic {@link NegativeArraySizeException}). + */ +class BinaryRowSerializerPageSizeGuardTest { + + // 200 INT columns -> fixed-length part 1632 bytes + 4 length bytes = 1636 bytes. + private static final int NUM_FIELDS = 200; + + @Test + void testThrowsWhenFixedPartExceedsPageSize() { + BinaryRowSerializer serializer = new BinaryRowSerializer(NUM_FIELDS); + int fixedPartLength = serializer.getSerializedRowFixedPartLength(); + // Page size (power of two) smaller than the fixed-length part. + int pageSize = 1024; + assertThat(fixedPartLength).isGreaterThan(pageSize); + + SimpleCollectingOutputView out = + new SimpleCollectingOutputView( + new CachelessSegmentPool(pageSize, pageSize), pageSize); + + BinaryRow row = newIntRow(NUM_FIELDS); + assertThatThrownBy(() -> serializer.serializeToPages(row, out)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("page size") + .hasMessageContaining(Integer.toString(fixedPartLength)) + .hasMessageContaining(Integer.toString(pageSize)) + .hasMessageContaining(Integer.toString(NUM_FIELDS)); + } + + @Test + void testSerializeSucceedsWhenPageLargeEnough() throws Exception { + BinaryRowSerializer serializer = new BinaryRowSerializer(NUM_FIELDS); + int pageSize = 4096; + assertThat(serializer.getSerializedRowFixedPartLength()).isLessThanOrEqualTo(pageSize); + + SimpleCollectingOutputView out = + new SimpleCollectingOutputView( + new CachelessSegmentPool((long) pageSize * 4, pageSize), pageSize); + + BinaryRow row = newIntRow(NUM_FIELDS); + serializer.serializeToPages(row, out); + + // Round-trip the row back from the written pages. + ArrayList segments = out.fullSegments(); + RandomAccessInputView in = new RandomAccessInputView(segments, pageSize); + BinaryRow deserialized = serializer.deserializeFromPages(serializer.createInstance(), in); + for (int i = 0; i < NUM_FIELDS; i++) { + assertThat(deserialized.getInt(i)).isEqualTo(i); + } + } + + private static BinaryRow newIntRow(int numFields) { + BinaryRow row = new BinaryRow(numFields); + BinaryRowWriter writer = new BinaryRowWriter(row); + for (int i = 0; i < numFields; i++) { + writer.writeInt(i, i); + } + writer.complete(); + return row; + } +} diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterOperatorTest.java index c7cca8eceb1d..6c48b9379b0a 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterOperatorTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/WriterOperatorTest.java @@ -92,7 +92,7 @@ public void testPrimaryKeyTableMetrics() throws Exception { options.set("bucket", "1"); options.set("write-buffer-size", "256 b"); options.set("write-buffer-spillable", "false"); - options.set("page-size", "32 b"); + options.set("page-size", "64 b"); FileStoreTable table = createFileStoreTable( @@ -109,7 +109,7 @@ public void testAppendOnlyTableMetrics() throws Exception { Options options = new Options(); options.set("write-buffer-for-append", "true"); options.set("write-buffer-size", "256 b"); - options.set("page-size", "32 b"); + options.set("page-size", "64 b"); options.set("write-buffer-spillable", "false"); FileStoreTable table = @@ -446,7 +446,7 @@ public void testNumWritersMetric() throws Exception { options.set("bucket", "1"); options.set("write-buffer-size", "256 b"); options.set("write-buffer-spillable", "false"); - options.set("page-size", "32 b"); + options.set("page-size", "64 b"); FileStoreTable fileStoreTable = createFileStoreTable(