From 1fbffd46d7d0538a7960230af7f0e09bcd0c7c95 Mon Sep 17 00:00:00 2001 From: Xiangyi Zhu <82511136+zhuxiangyi@users.noreply.github.com> Date: Fri, 18 Sep 2026 18:05:48 +0800 Subject: [PATCH] [format] Validate the avro block size before creating the writer AvroFileFormat passed file.block-size straight to Avro's setSyncInterval, which only accepts 32 bytes to 1 GiB. Other values failed at write time on the executor with Avro's own "Invalid syncInterval value: N" or, above 2 GiB, with "integer overflow" from Math.toIntExact, neither naming the option. Check the range up front and fail with the option name, value and bounds. The unchecked value also made DeletionVectorTest "select with format filter push down" flaky: it draws a random block size from [1, 10240] and a random format, so about one run in a thousand lands on avro with a block size below 32. It failed the Spark 3 / Scala 2.13 job of #9953 with block size 6 on a change unrelated to Avro. Draw from [32, 10271] instead. testFileBlockSizeOverflow, which pinned the raw ArithmeticException, is replaced by testFileBlockSizeOutOfAvroRange (1, 6, 31, 2^30+1, 2^31, 4 GiB+, Long.MAX_VALUE) and testFileBlockSizeAtAvroRangeBounds (32, 33, 2^30). --- .../paimon/format/avro/AvroFileFormat.java | 23 ++++++++++++++- .../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 ad243b613254..aa03f1f8628e 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 @@ private AvroBlockWriter createBlockWriter( } 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 void validateDataFields(RowType rowType) { } } + /** + * 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 bef9ef8873b4..0f73258b6637 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 @@ private void assertFileBlockSize(FileFormat format, int expectedBlockSize, Strin } @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 @@ void testFileBlockSizeOverflow(long blockSize) throws IOException { 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 05c706b307a2..4dc4e3d01715 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 (