This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 239321dd829c77165f297fe3df9e485a52a09251 Author: Xiangyi Zhu <[email protected]> AuthorDate: Sat Sep 19 22:03:44 2026 +0800 [format] Validate the avro block size before creating the writer (#9961) --- .../apache/paimon/format/avro/AvroFileFormat.java | 23 ++++++++++++++++- .../paimon/format/avro/AvroFileFormatTest.java | 29 +++++++++++++++++++--- .../paimon/spark/sql/DeletionVectorTest.scala | 3 ++- 3 files changed, 49 insertions(+), 6 deletions(-) diff --git a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java index ad243b6132..aa03f1f862 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/avro/AvroFileFormat.java @@ -18,6 +18,7 @@ package org.apache.paimon.format.avro; +import org.apache.paimon.CoreOptions; import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.FileFormat; import org.apache.paimon.format.FileFormatFactory.FormatContext; @@ -67,6 +68,11 @@ public class AvroFileFormat extends FileFormat { private final Options options; private final int zstdLevel; + /** Bounds enforced by {@code DataFileWriter#setSyncInterval}. */ + private static final long MIN_SYNC_INTERVAL = 32; + + private static final long MAX_SYNC_INTERVAL = 1 << 30; + @Nullable private final MemorySize blockSize; public AvroFileFormat(FormatContext context) { @@ -114,7 +120,7 @@ public class AvroFileFormat extends FileFormat { } writer.setCodec(createCodecFactory(compression)); if (blockSize != null) { - writer.setSyncInterval(Math.toIntExact(blockSize.getBytes())); + writer.setSyncInterval(avroSyncInterval(blockSize)); } writer.setFlushOnEveryBlock(false); writer.create(schema, new CloseShieldOutputStream(out)); @@ -135,6 +141,21 @@ public class AvroFileFormat extends FileFormat { } } + /** + * Avro only accepts a sync interval between 32 bytes and 1 GiB; check it here so a bad {@code + * file.block-size} fails with the option name instead of inside the writer on an executor. + */ + static int avroSyncInterval(MemorySize blockSize) { + long bytes = blockSize.getBytes(); + if (bytes < MIN_SYNC_INTERVAL || bytes > MAX_SYNC_INTERVAL) { + throw new IllegalArgumentException( + String.format( + "%s for avro must be between 32 bytes and 1 gb, but was %s bytes.", + CoreOptions.FILE_BLOCK_SIZE.key(), bytes)); + } + return (int) bytes; + } + private CodecFactory createCodecFactory(String compression) { if (options.contains(AVRO_OUTPUT_CODEC)) { return CodecFactory.fromString(options.get(AVRO_OUTPUT_CODEC)); diff --git a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java index bef9ef8873..0f73258b66 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/avro/AvroFileFormatTest.java @@ -156,8 +156,10 @@ public class AvroFileFormatTest { } @ParameterizedTest - @ValueSource(longs = {2147483648L, 4294968320L, Long.MAX_VALUE}) - void testFileBlockSizeOverflow(long blockSize) throws IOException { + @ValueSource(longs = {1L, 6L, 31L, 1073741825L, 2147483648L, 4294968320L, Long.MAX_VALUE}) + void testFileBlockSizeOutOfAvroRange(long blockSize) throws IOException { + // Avro accepts a sync interval of 32 bytes to 1 GiB; anything else must be rejected up + // front with the option name, not deep inside the writer with Avro's own message. Options options = new Options(); options.setString("file.block-size", Long.toString(blockSize)); FileFormat format = FileFormat.fromIdentifier("avro", options); @@ -173,8 +175,27 @@ public class AvroFileFormatTest { writer.addElement(GenericRow.of(0)); } }) - .isInstanceOf(ArithmeticException.class) - .hasMessage("integer overflow"); + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("file.block-size") + .hasMessageContaining(Long.toString(blockSize)) + .hasMessageContaining("32 bytes") + .hasMessageContaining("1 gb"); + } + } + + @ParameterizedTest + @ValueSource(longs = {32L, 33L, 1073741824L}) + void testFileBlockSizeAtAvroRangeBounds(long blockSize) throws IOException { + Options options = new Options(); + options.setString("file.block-size", Long.toString(blockSize)); + FileFormat format = FileFormat.fromIdentifier("avro", options); + RowType rowType = DataTypes.ROW(DataTypes.INT().notNull()).notNull(); + LocalFileIO fileIO = LocalFileIO.create(); + Path file = new Path(new Path(tempPath.toUri()), UUID.randomUUID().toString()); + + try (PositionOutputStream out = fileIO.newOutputStream(file, false); + FormatWriter writer = format.createWriterFactory(rowType).create(out, "null")) { + writer.addElement(GenericRow.of(0)); } } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala index 05c706b307..4dc4e3d017 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/DeletionVectorTest.scala @@ -622,7 +622,8 @@ class DeletionVectorTest extends PaimonSparkTestBase with AdaptiveSparkPlanHelpe test("Paimon deletionVector: select with format filter push down") { val format = Random.shuffle(Seq("parquet", "orc", "avro")).head - val blockSize = Random.nextInt(10240) + 1 + // Avro rejects a block size below 32 bytes. + val blockSize = Random.nextInt(10240) + 32 spark.sql(s""" |CREATE TABLE T (id INT, name STRING) |TBLPROPERTIES (
