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 4750a774feedd82ebcd897ae2971acf1387a0272 Author: umi <[email protected]> AuthorDate: Thu Sep 17 16:50:27 2026 +0800 [format] Support file.block-size for Avro data file (#9910) --- docs/generated/core_configuration.html | 2 +- .../main/java/org/apache/paimon/CoreOptions.java | 3 +- .../java/org/apache/paimon/format/FileFormat.java | 5 +- .../iceberg/manifest/IcebergManifestFile.java | 3 +- .../iceberg/manifest/IcebergManifestList.java | 3 +- .../manifest/IcebergManifestBlockSizeTest.java | 182 +++++++++++++++++++++ .../apache/paimon/format/avro/AvroFileFormat.java | 6 + .../paimon/format/avro/AvroFileFormatTest.java | 100 +++++++++++ 8 files changed, 299 insertions(+), 5 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index cf48d473c4..2fb5acc403 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -738,7 +738,7 @@ under the License. <td><h5>file.block-size</h5></td> <td style="word-wrap: break-word;">(none)</td> <td>MemorySize</td> - <td>File block size of format, default value of orc stripe is 64 MB, and parquet row group is 128 MB.</td> + <td>File block size of format, default value of orc stripe is 64 MB, parquet row group is 128 MB, and avro block is 64 KB.</td> </tr> <tr> <td><h5>file.compression</h5></td> 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 9bb324b780..637a8ba0e7 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 class CoreOptions implements Serializable { .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 64 KB."); public static final ConfigOption<MemorySize> FILE_INDEX_IN_MANIFEST_THRESHOLD = key("file-index.in-manifest-threshold") diff --git a/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java b/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java index d6bfe7a0d3..f17c23b255 100644 --- a/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java +++ b/paimon-common/src/main/java/org/apache/paimon/format/FileFormat.java @@ -117,6 +117,9 @@ public abstract class FileFormat { } public static FileFormat manifestFormat(CoreOptions options) { - return FileFormat.fromIdentifier(CoreOptions.FILE_FORMAT_AVRO, options.toConfiguration()); + // Manifest blocks retain the Avro default independently of data-file block sizing. + Options manifestOptions = new Options(options.toMap()); + manifestOptions.remove(CoreOptions.FILE_BLOCK_SIZE); + return FileFormat.fromIdentifier(CoreOptions.FILE_FORMAT_AVRO, manifestOptions); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java index 5e43a2e2fa..d21ead71f0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFile.java @@ -18,6 +18,7 @@ package org.apache.paimon.iceberg.manifest; +import org.apache.paimon.CoreOptions; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.data.InternalRow; import org.apache.paimon.data.serializer.Serializer; @@ -120,7 +121,7 @@ public class IcebergManifestFile extends ObjectsFile<IcebergManifestEntry> { + "kv_name_r2_upper_bounds:k129_v130," + "k_id_k129_v130:129," + "v_id_k129_v130:130"); - FileFormat manifestFileAvro = FileFormat.fromIdentifier("avro", avroOptions); + FileFormat manifestFileAvro = FileFormat.manifestFormat(new CoreOptions(avroOptions)); return new IcebergManifestFile( table.fileIO(), partitionType, diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java index 0439aba087..7ea98630c1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestList.java @@ -18,6 +18,7 @@ package org.apache.paimon.iceberg.manifest; +import org.apache.paimon.CoreOptions; import org.apache.paimon.annotation.VisibleForTesting; import org.apache.paimon.format.FileFormat; import org.apache.paimon.fs.FileIO; @@ -69,7 +70,7 @@ public class IcebergManifestList extends ObjectsFile<IcebergManifestFileMeta> { + "iceberg:true," + "manifest_file_partitions:r508," + "array_id_r508:508"); - FileFormat fileFormat = FileFormat.fromIdentifier("avro", avroOptions); + FileFormat fileFormat = FileFormat.manifestFormat(new CoreOptions(avroOptions)); boolean withFirstRowId = avroOptions.get(IcebergOptions.FORMAT_VERSION) >= IcebergMetadata.FORMAT_VERSION_V3; RowType manifestType = diff --git a/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergManifestBlockSizeTest.java b/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergManifestBlockSizeTest.java new file mode 100644 index 0000000000..554c29e06f --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergManifestBlockSizeTest.java @@ -0,0 +1,182 @@ +/* + * 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.iceberg.manifest; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.GenericMap; +import org.apache.paimon.format.avro.AvroBlockReader; +import org.apache.paimon.format.avro.AvroRawBlock; +import org.apache.paimon.fs.Path; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.iceberg.IcebergOptions; +import org.apache.paimon.iceberg.IcebergPathFactory; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.options.Options; +import org.apache.paimon.schema.FileSystemSchemaManager; +import org.apache.paimon.schema.Schema; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.FileStoreTableFactory; +import org.apache.paimon.types.DataTypes; + +import org.apache.avro.file.DataFileStream; +import org.apache.avro.generic.GenericDatumReader; +import org.apache.avro.generic.GenericRecord; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; + +import static org.apache.avro.file.DataFileConstants.DEFAULT_SYNC_INTERVAL; +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests that Iceberg metadata blocks are independent of data-file block sizing. */ +class IcebergManifestBlockSizeTest { + + private static final int NUM_RECORDS = 10_000; + + @TempDir java.nio.file.Path tempDir; + + @ParameterizedTest + @CsvSource({"1 kb, 2", "256 mb, 2", "1 kb, 3", "256 mb, 3"}) + void testManifestIgnoresDataFileBlockSize(String blockSize, int formatVersion) + throws Exception { + FileStoreTable table = createTable(blockSize, formatVersion); + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestFile manifest = IcebergManifestFile.create(table, pathFactory); + GenericMap emptyMap = new GenericMap(Collections.emptyMap()); + IcebergDataFileMeta dataFile = + new IcebergDataFileMeta( + IcebergDataFileMeta.Content.DATA, + "data.parquet", + "parquet", + BinaryRow.EMPTY_ROW, + 1, + 100, + emptyMap, + emptyMap, + emptyMap); + List<IcebergManifestEntry> entries = + Collections.nCopies( + NUM_RECORDS, + new IcebergManifestEntry( + IcebergManifestEntry.Status.ADDED, 1, 1, 1, dataFile)); + + List<IcebergManifestFileMeta> files = manifest.rollingWrite(entries.iterator(), 1); + + assertThat(files).hasSize(1); + assertDefaultBlocks(table, new Path(files.get(0).manifestPath()), "manifest_entry"); + assertThat(manifest.read(files.get(0))) + .hasSize(NUM_RECORDS) + .allSatisfy( + entry -> { + assertThat(entry.status()).isEqualTo(IcebergManifestEntry.Status.ADDED); + assertThat(entry.file().filePath()).isEqualTo("data.parquet"); + assertThat(entry.file().recordCount()).isEqualTo(1); + }); + assertThat(table.coreOptions().toConfiguration().get(CoreOptions.FILE_BLOCK_SIZE)) + .isEqualTo(MemorySize.parse(blockSize)); + } + + @ParameterizedTest + @CsvSource({"1 kb, 2", "256 mb, 2", "1 kb, 3", "256 mb, 3"}) + void testManifestListIgnoresDataFileBlockSize(String blockSize, int formatVersion) + throws Exception { + FileStoreTable table = createTable(blockSize, formatVersion); + IcebergPathFactory pathFactory = + new IcebergPathFactory(new Path(table.location(), "metadata")); + IcebergManifestList manifestList = IcebergManifestList.create(table, pathFactory); + List<IcebergManifestFileMeta> files = + Collections.nCopies( + NUM_RECORDS, + new IcebergManifestFileMeta( + "manifest.avro", + 100, + 0, + IcebergManifestFileMeta.Content.DATA, + 1, + 1, + 1, + 1, + 0, + 0, + 1, + 0, + 0, + Collections.emptyList(), + null)); + + String fileName = manifestList.writeWithoutRolling(files); + + assertDefaultBlocks(table, pathFactory.toManifestListPath(fileName), "manifest_file"); + assertThat(manifestList.read(fileName)).containsExactlyElementsOf(files); + assertThat(table.coreOptions().toConfiguration().get(CoreOptions.FILE_BLOCK_SIZE)) + .isEqualTo(MemorySize.parse(blockSize)); + } + + private FileStoreTable createTable(String blockSize, int formatVersion) throws Exception { + Options options = new Options(); + options.set(CoreOptions.FILE_FORMAT, "parquet"); + options.set(CoreOptions.FILE_BLOCK_SIZE, MemorySize.parse(blockSize)); + options.set(IcebergOptions.FORMAT_VERSION, formatVersion); + options.set(IcebergOptions.MANIFEST_COMPRESSION, "deflate"); + Schema schema = + new Schema( + DataTypes.ROW(DataTypes.INT()).getFields(), + Collections.emptyList(), + Collections.emptyList(), + options.toMap(), + ""); + LocalFileIO fileIO = LocalFileIO.create(); + Path path = new Path(tempDir.toUri()); + new FileSystemSchemaManager(fileIO, path).createTable(schema); + return FileStoreTableFactory.create(fileIO, path); + } + + private void assertDefaultBlocks(FileStoreTable table, Path path, String recordName) + throws IOException { + try (DataFileStream<GenericRecord> reader = + new DataFileStream<>( + table.fileIO().newInputStream(path), new GenericDatumReader<>())) { + assertThat(reader.getSchema().getName()).isEqualTo(recordName); + assertThat(reader.getMetaString("avro.codec")).isEqualTo("deflate"); + } + long records = 0; + int blocks = 0; + try (AvroBlockReader reader = new AvroBlockReader(table.fileIO().newInputStream(path))) { + while (reader.hasNextBlock()) { + AvroRawBlock block = reader.nextBorrowedRawBlock(); + int size = block.decompress(null).remaining(); + // The threshold is checked after each record, so a block can exceed it slightly. + assertThat(size).isLessThan(DEFAULT_SYNC_INTERVAL + 1024); + if (reader.hasNextBlock()) { + assertThat(size).isGreaterThanOrEqualTo(DEFAULT_SYNC_INTERVAL); + } + records += block.recordCount(); + blocks++; + } + } + assertThat(blocks).isGreaterThan(1); + assertThat(records).isEqualTo(NUM_RECORDS); + } +} 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 69512d5ee7..a23c66af21 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.CloseShieldOutputStream; 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 class AvroFileFormat extends FileFormat { AvroRowDatumWriter datumWriter = new AvroRowDatumWriter(rowType); DataFileWriter<InternalRow> 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 c7ca45b88f..70780f8d9b 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 @@ -18,6 +18,7 @@ package org.apache.paimon.format.avro; +import org.apache.paimon.CoreOptions; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; import org.apache.paimon.format.FileFormat; @@ -48,6 +49,9 @@ import org.apache.avro.io.EncoderFactory; 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 org.junit.jupiter.params.provider.ValueSource; import java.io.ByteArrayOutputStream; import java.io.File; @@ -76,6 +80,102 @@ public class AvroFileFormatTest { 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); + assertFileBlockSize(format, expectedBlockSize, compression); + } + + @ParameterizedTest + @ValueSource(strings = {"avro", "parquet", "orc"}) + void testManifestIgnoresDataFileBlockSize(String identifier) throws IOException { + Options options = new Options(); + options.set(CoreOptions.FILE_FORMAT, identifier); + options.setString("file.block-size", "1 kb"); + + FileFormat manifestFormat = FileFormat.manifestFormat(new CoreOptions(options)); + assertFileBlockSize(manifestFormat, 64000, "null"); + + assertThat(options.get(CoreOptions.FILE_BLOCK_SIZE).getBytes()).isEqualTo(1024); + assertFileBlockSize(FileFormat.fromIdentifier("avro", options), 1024, "null"); + } + + @Test + void testManifestIgnoresLargeDataFileBlockSize() throws IOException { + Options options = new Options(); + options.set(CoreOptions.FILE_FORMAT, "parquet"); + options.setString("file.block-size", "256 mb"); + + assertFileBlockSize(FileFormat.manifestFormat(new CoreOptions(options)), 64000, "null"); + } + + private void assertFileBlockSize(FileFormat format, int expectedBlockSize, String compression) + throws IOException { + 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); + } + + @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<DataField> dataFields = new ArrayList<>();
