From 6bb6377fff3f32b7bdf9ba6e6d1f270f722a553c Mon Sep 17 00:00:00 2001 From: umi Date: Mon, 14 Sep 2026 15:35:17 +0800 Subject: [PATCH 1/3] [format] Support file.block-size for Avro --- docs/generated/core_configuration.html | 2 +- .../java/org/apache/paimon/CoreOptions.java | 3 +- .../paimon/format/avro/AvroFileFormat.java | 6 +++ .../format/avro/AvroFileFormatTest.java | 47 +++++++++++++++++++ 4 files changed, 56 insertions(+), 2 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index ac5730017523..09db240dfda1 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -738,7 +738,7 @@
file.block-size
(none) MemorySize - File block size of format, default value of orc stripe is 64 MB, and parquet row group is 128 MB. + File block size of format, default value of orc stripe is 64 MB, parquet row group is 128 MB, and avro block is 64000 bytes. For avro, this is the uncompressed block size threshold.
file.compression
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 7b0665c50296..66d917314d97 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -426,7 +426,8 @@ public InlineElement getDescription() { .memoryType() .noDefaultValue() .withDescription( - "File block size of format, default value of orc stripe is 64 MB, and parquet row group is 128 MB."); + "File block size of format, default value of orc stripe is 64 MB, parquet row group is 128 MB, " + + "and avro block is 64000 bytes. For avro, this is the uncompressed block size threshold."); public static final ConfigOption FILE_INDEX_IN_MANIFEST_THRESHOLD = key("file-index.in-manifest-threshold") 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 69512d5ee7cf..a23c66af2181 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 @@ -29,6 +29,7 @@ import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.options.ConfigOption; import org.apache.paimon.options.ConfigOptions; +import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.apache.paimon.predicate.Predicate; import org.apache.paimon.statistics.SimpleColStatsCollector; @@ -65,12 +66,14 @@ public class AvroFileFormat extends FileFormat { private final Options options; private final int zstdLevel; + @Nullable private final MemorySize blockSize; public AvroFileFormat(FormatContext context) { super(IDENTIFIER); this.options = getIdentifierPrefixOptions(context.options()); this.zstdLevel = context.zstdLevel(); + this.blockSize = context.blockSize(); } @Override @@ -93,6 +96,9 @@ public AvroBlockWriter createBlockWriter( AvroRowDatumWriter datumWriter = new AvroRowDatumWriter(rowType); DataFileWriter writer = new DataFileWriter<>(datumWriter); writer.setCodec(createCodecFactory(compression)); + if (blockSize != null) { + writer.setSyncInterval(Math.toIntExact(blockSize.getBytes())); + } writer.setFlushOnEveryBlock(false); writer.create(schema, new CloseShieldOutputStream(out)); return new AvroBlockWriter(writer, out, schema); 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 89605046b104..55a42ede198a 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 @@ -48,6 +48,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import java.io.ByteArrayOutputStream; import java.io.File; @@ -76,6 +78,51 @@ public static void before() { fileFormat = new AvroFileFormat(new FormatContext(new Options(), 1024, 1024)); } + @ParameterizedTest + @CsvSource({ + ", 64000, null", + ", 64000, deflate", + "1 kb, 1024, null", + "1 kb, 1024, deflate", + "4 kb, 4096, null", + "4 kb, 4096, deflate", + "128 kb, 131072, null", + "128 kb, 131072, deflate" + }) + void testFileBlockSize(String blockSize, int expectedBlockSize, String compression) + throws IOException { + Options options = new Options(); + if (blockSize != null) { + options.setString("file.block-size", 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()); + int numRecords = 300_000; + + try (PositionOutputStream out = fileIO.newOutputStream(file, false); + FormatWriter writer = + format.createWriterFactory(rowType).create(out, compression)) { + // Each record is encoded as one byte, so record counts also give uncompressed sizes. + for (int i = 0; i < numRecords; i++) { + writer.addElement(GenericRow.of(0)); + } + } + + long records = 0; + try (AvroBlockReader reader = new AvroBlockReader(fileIO.newInputStream(file))) { + while (reader.hasNextBlock()) { + AvroRawBlock block = reader.nextBorrowedRawBlock(); + long expectedRecords = Math.min(expectedBlockSize, numRecords - records); + assertThat(block.recordCount()).isEqualTo(expectedRecords); + assertThat(block.decompress(null).remaining()).isEqualTo((int) expectedRecords); + records += block.recordCount(); + } + } + assertThat(records).isEqualTo(numRecords); + } + @Test public void testSupportedDataTypes() { ArrayList dataFields = new ArrayList<>(); From 12f0d20196021e6f44f70de75555a0c3d8f8440c Mon Sep 17 00:00:00 2001 From: umi Date: Tue, 15 Sep 2026 14:10:21 +0800 Subject: [PATCH 2/3] fix --- .../main/java/org/apache/paimon/format/avro/AvroFileFormat.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 a23c66af2181..f01da00ee4b0 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 @@ -97,7 +97,7 @@ public AvroBlockWriter createBlockWriter( DataFileWriter writer = new DataFileWriter<>(datumWriter); writer.setCodec(createCodecFactory(compression)); if (blockSize != null) { - writer.setSyncInterval(Math.toIntExact(blockSize.getBytes())); + writer.setSyncInterval((int) blockSize.getBytes()); } writer.setFlushOnEveryBlock(false); writer.create(schema, new CloseShieldOutputStream(out)); From 55d8d1796dfe24156b5968e0ce93fbebd1473214 Mon Sep 17 00:00:00 2001 From: umi Date: Wed, 16 Sep 2026 21:31:01 +0800 Subject: [PATCH 3/3] [format] Reject overflowing Avro block sizes --- .../paimon/format/avro/AvroFileFormat.java | 2 +- .../format/avro/AvroFileFormatTest.java | 24 +++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) 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 f01da00ee4b0..a23c66af2181 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 @@ -97,7 +97,7 @@ public AvroBlockWriter createBlockWriter( DataFileWriter writer = new DataFileWriter<>(datumWriter); writer.setCodec(createCodecFactory(compression)); if (blockSize != null) { - writer.setSyncInterval((int) blockSize.getBytes()); + writer.setSyncInterval(Math.toIntExact(blockSize.getBytes())); } writer.setFlushOnEveryBlock(false); writer.create(schema, new CloseShieldOutputStream(out)); 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 55a42ede198a..a13bd2a0c258 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 @@ -50,6 +50,7 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.ValueSource; import java.io.ByteArrayOutputStream; import java.io.File; @@ -123,6 +124,29 @@ void testFileBlockSize(String blockSize, int expectedBlockSize, String compressi assertThat(records).isEqualTo(numRecords); } + @ParameterizedTest + @ValueSource(longs = {2147483648L, 4294968320L, Long.MAX_VALUE}) + void testFileBlockSizeOverflow(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)) { + assertThatThrownBy( + () -> { + try (FormatWriter writer = + format.createWriterFactory(rowType).create(out, "null")) { + writer.addElement(GenericRow.of(0)); + } + }) + .isInstanceOf(ArithmeticException.class) + .hasMessage("integer overflow"); + } + } + @Test public void testSupportedDataTypes() { ArrayList dataFields = new ArrayList<>();