This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new ee7ac67561 [format] Support file.block-size for Avro data file (#9910)
ee7ac67561 is described below
commit ee7ac67561088a57b323a0344f19ec95b7dac60f
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<>();
