You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by pv...@apache.org on 2026/06/05 10:41:59 UTC
(iceberg) branch main updated: Data: Add TCK for Writer builder in FileFormat API (#16575)
This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 7197765959 Data: Add TCK for Writer builder in FileFormat API (#16575)
7197765959 is described below
commit 719776595988a0444a90a60a09fdb0d42ab5d7c1
Author: GuoYu <51...@qq.com>
AuthorDate: Fri Jun 5 18:41:52 2026 +0800
Data: Add TCK for Writer builder in FileFormat API (#16575)
---
.../apache/iceberg/data/BaseFormatModelTests.java | 369 +++++++++------------
.../apache/iceberg/data/FileFormatTestSupport.java | 61 ++++
.../org/apache/iceberg/data/avro/AvroFormat.java | 101 ++++++
.../org/apache/iceberg/data/orc/OrcFormat.java | 130 ++++++++
.../apache/iceberg/data/parquet/ParquetFormat.java | 129 +++++++
5 files changed, 586 insertions(+), 204 deletions(-)
diff --git a/data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java b/data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java
index 5373749406..78bdfe8bca 100644
--- a/data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java
+++ b/data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java
@@ -28,7 +28,6 @@ import static org.junit.jupiter.api.Assumptions.assumeFalse;
import java.io.File;
import java.io.IOException;
-import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Comparator;
@@ -39,14 +38,6 @@ import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.IntStream;
-import org.apache.avro.file.DataFileStream;
-import org.apache.avro.file.DataFileWriter;
-import org.apache.avro.generic.GenericData;
-import org.apache.avro.generic.GenericDatumReader;
-import org.apache.avro.generic.GenericDatumWriter;
-import org.apache.avro.io.DatumWriter;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
import org.apache.iceberg.ContentFile;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
@@ -61,14 +52,13 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.TestTables;
-import org.apache.iceberg.avro.AvroTestHelpers;
-import org.apache.iceberg.data.orc.GenericOrcWriter;
import org.apache.iceberg.deletes.EqualityDeleteWriter;
import org.apache.iceberg.deletes.PositionDelete;
import org.apache.iceberg.deletes.PositionDeleteWriter;
import org.apache.iceberg.encryption.EncryptedFiles;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.encryption.EncryptionKeyMetadata;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.NotFoundException;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expression;
@@ -81,15 +71,8 @@ import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.io.DeleteSchemaUtil;
import org.apache.iceberg.io.InputFile;
-import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.mapping.MappingUtil;
import org.apache.iceberg.mapping.NameMapping;
-import org.apache.iceberg.orc.ORCSchemaUtil;
-import org.apache.iceberg.orc.OrcRowWriter;
-import org.apache.iceberg.orc.OrcWritingTestUtils;
-import org.apache.iceberg.orc.TestORCSchemaUtil;
-import org.apache.iceberg.parquet.ParquetFileTestUtils;
-import org.apache.iceberg.parquet.ParquetSchemaUtil;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -97,14 +80,6 @@ import org.apache.iceberg.types.Comparators;
import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
-import org.apache.orc.OrcFile;
-import org.apache.orc.Reader;
-import org.apache.orc.TypeDescription;
-import org.apache.orc.Writer;
-import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
-import org.apache.parquet.avro.AvroParquetWriter;
-import org.apache.parquet.hadoop.ParquetFileReader;
-import org.apache.parquet.hadoop.ParquetWriter;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.io.TempDir;
@@ -130,8 +105,7 @@ public abstract class BaseFormatModelTests<T> {
@TempDir private File tableDir;
- private static final FileFormat[] FILE_FORMATS =
- new FileFormat[] {FileFormat.AVRO, FileFormat.PARQUET, FileFormat.ORC};
+ private static final FileFormat[] FILE_FORMATS = FileFormatTestSupport.formats();
private static final List<Arguments> FORMAT_AND_GENERATOR =
Arrays.stream(FILE_FORMATS)
@@ -197,24 +171,9 @@ public abstract class BaseFormatModelTests<T> {
void testDataWriterEngineWriteGenericRead(FileFormat fileFormat, DataGenerator dataGenerator)
throws IOException {
Schema schema = dataGenerator.schema();
- FileWriterBuilder<DataWriter<T>, Object> writerBuilder =
- FormatModelRegistry.dataWriteBuilder(fileFormat, engineType(), encryptedFile);
-
- DataWriter<T> writer = writerBuilder.schema(schema).spec(PartitionSpec.unpartitioned()).build();
-
List<Record> genericRecords = dataGenerator.generateRecords();
List<T> engineRecords = convertToEngineRecords(genericRecords, schema);
-
- try (writer) {
- engineRecords.forEach(writer::write);
- }
-
- DataFile dataFile = writer.toDataFile();
-
- assertThat(dataFile).isNotNull();
- assertThat(dataFile.recordCount()).isEqualTo(engineRecords.size());
- assertThat(dataFile.format()).isEqualTo(fileFormat);
-
+ writeEngineRecords(fileFormat, schema, engineRecords);
readAndAssertGenericRecords(fileFormat, schema, genericRecords);
}
@@ -224,24 +183,9 @@ public abstract class BaseFormatModelTests<T> {
void testDataWriterEngineWriteWithoutEngineSchema(
FileFormat fileFormat, DataGenerator dataGenerator) throws IOException {
Schema schema = dataGenerator.schema();
- FileWriterBuilder<DataWriter<T>, Object> writerBuilder =
- FormatModelRegistry.dataWriteBuilder(fileFormat, engineType(), encryptedFile);
-
- DataWriter<T> writer = writerBuilder.schema(schema).spec(PartitionSpec.unpartitioned()).build();
-
List<Record> genericRecords = dataGenerator.generateRecords();
List<T> engineRecords = convertToEngineRecords(genericRecords, schema);
-
- try (writer) {
- engineRecords.forEach(writer::write);
- }
-
- DataFile dataFile = writer.toDataFile();
-
- assertThat(dataFile).isNotNull();
- assertThat(dataFile.recordCount()).isEqualTo(engineRecords.size());
- assertThat(dataFile.format()).isEqualTo(fileFormat);
-
+ writeEngineRecords(fileFormat, schema, engineRecords);
readAndAssertGenericRecords(fileFormat, schema, genericRecords);
}
@@ -274,34 +218,10 @@ public abstract class BaseFormatModelTests<T> {
void testDataWriterEngineWriteEngineRead(FileFormat fileFormat, DataGenerator dataGenerator)
throws IOException {
Schema schema = dataGenerator.schema();
- FileWriterBuilder<DataWriter<T>, Object> writerBuilder =
- FormatModelRegistry.dataWriteBuilder(fileFormat, engineType(), encryptedFile);
-
- DataWriter<T> writer = writerBuilder.schema(schema).spec(PartitionSpec.unpartitioned()).build();
-
List<Record> genericRecords = dataGenerator.generateRecords();
List<T> engineRecords = convertToEngineRecords(genericRecords, schema);
-
- try (writer) {
- engineRecords.forEach(writer::write);
- }
-
- DataFile dataFile = writer.toDataFile();
-
- assertThat(dataFile).isNotNull();
- assertThat(dataFile.recordCount()).isEqualTo(engineRecords.size());
- assertThat(dataFile.format()).isEqualTo(fileFormat);
-
- InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
- List<T> readRecords;
- try (CloseableIterable<T> reader =
- FormatModelRegistry.readBuilder(fileFormat, engineType(), inputFile)
- .project(schema)
- .build()) {
- readRecords = ImmutableList.copyOf(reader);
- }
-
- assertEquals(schema, engineRecords, readRecords);
+ writeEngineRecords(fileFormat, schema, engineRecords);
+ readAndAssertEngineRecords(fileFormat, schema, genericRecords, Function.identity());
}
/** Write with engine type T, read with Generic Record */
@@ -1645,7 +1565,9 @@ public abstract class BaseFormatModelTests<T> {
List<Record> genericRecords = dataGenerator.generateRecords();
// Write the file WITHOUT Iceberg field IDs (as an external writer would).
- writeRecordsWithoutFieldIds(fileFormat, icebergSchema, genericRecords);
+ FileFormatTestSupport.forFormat(fileFormat)
+ .writeRecordsWithoutFieldIds(
+ encryptedFile.encryptingOutputFile(), icebergSchema, genericRecords);
NameMapping nameMapping = MappingUtil.create(icebergSchema);
@@ -1662,6 +1584,83 @@ public abstract class BaseFormatModelTests<T> {
assertEquals(icebergSchema, convertToEngineRecords(genericRecords, icebergSchema), readRecords);
}
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testDataWriterOverwrite(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema schema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ List<T> engineRecords = convertToEngineRecords(genericRecords, schema);
+
+ writeEngineRecords(fileFormat, schema, engineRecords);
+ readAndAssertGenericRecords(fileFormat, schema, genericRecords);
+
+ genericRecords = dataGenerator.generateRecords(20);
+ writeEngineRecords(
+ fileFormat, schema, convertToEngineRecords(genericRecords, schema), true /* overwrite */);
+ readAndAssertGenericRecords(fileFormat, schema, genericRecords);
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testDataWriterNoOverwriteFailsIfFileExists(FileFormat fileFormat) throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema schema = dataGenerator.schema();
+
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ List<T> engineRecords = convertToEngineRecords(genericRecords, schema);
+
+ writeEngineRecords(fileFormat, schema, engineRecords);
+ readAndAssertGenericRecords(fileFormat, schema, genericRecords);
+
+ assertThatThrownBy(() -> writeEngineRecords(fileFormat, schema, engineRecords))
+ .isInstanceOf(AlreadyExistsException.class)
+ .hasMessageContaining("Already exists");
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testDataWriterSet(FileFormat fileFormat) throws IOException {
+ writeAndAssertDataWriterWithConfig(
+ fileFormat,
+ (writerBuilder, format) -> testPropertiesToSet(format).forEach(writerBuilder::set),
+ format -> assertThat(checkTestProperties(format)).isTrue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testDataWriterSetAll(FileFormat fileFormat) throws IOException {
+ writeAndAssertDataWriterWithConfig(
+ fileFormat,
+ (writerBuilder, format) -> writerBuilder.setAll(testPropertiesToSet(format)),
+ format -> assertThat(checkTestProperties(format)).isTrue());
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testDataWriterMeta(FileFormat fileFormat) throws IOException {
+ writeAndAssertDataWriterWithConfig(
+ fileFormat,
+ (writerBuilder, format) -> writerBuilder.meta("tck.meta.key", "tck-meta-value"),
+ format ->
+ assertThat(fileMetadataValue(format, "tck.meta.key")).isEqualTo("tck-meta-value"));
+ }
+
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testDataWriterMetaMap(FileFormat fileFormat) throws IOException {
+ writeAndAssertDataWriterWithConfig(
+ fileFormat,
+ (writerBuilder, format) ->
+ writerBuilder.meta(
+ Map.of("tck.meta.key", "tck-meta-value", "tck.meta.key2", "tck-meta-value2")),
+ format -> {
+ assertThat(fileMetadataValue(format, "tck.meta.key")).isEqualTo("tck-meta-value");
+ assertThat(fileMetadataValue(format, "tck.meta.key2")).isEqualTo("tck-meta-value2");
+ });
+ }
+
private void readAndAssertGenericRecords(
FileFormat fileFormat, Schema schema, List<Record> expected) throws IOException {
InputFile inputFile = encryptedFile.encryptingOutputFile().toInputFile();
@@ -1737,7 +1736,7 @@ public abstract class BaseFormatModelTests<T> {
private DataFile writeRecordsForSplit(FileFormat fileFormat, Schema schema, List<Record> records)
throws IOException {
- String splitSizeProperty = splitSizeProperty(fileFormat);
+ String splitSizeProperty = FileFormatTestSupport.forFormat(fileFormat).splitSizeProperty();
DataWriter<Record> writer =
FormatModelRegistry.dataWriteBuilder(fileFormat, Record.class, encryptedFile)
.schema(schema)
@@ -1763,16 +1762,6 @@ public abstract class BaseFormatModelTests<T> {
return dataFile;
}
- private static String splitSizeProperty(FileFormat fileFormat) {
- return switch (fileFormat) {
- case PARQUET -> TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES;
- case ORC -> TableProperties.ORC_STRIPE_SIZE_BYTES;
- default ->
- throw new UnsupportedOperationException(
- "No split size property defined for format: " + fileFormat);
- };
- }
-
private static void assertCounts(
FileFormat fileFormat, Schema schema, List<Record> genericRecords, ContentFile<?> file) {
if (!supportsFeature(fileFormat, FEATURE_COLUMN_LEVEL_METRICS)) {
@@ -2110,112 +2099,6 @@ public abstract class BaseFormatModelTests<T> {
return result;
}
- private void writeRecordsWithoutFieldIds(
- FileFormat fileFormat, Schema schema, List<Record> records) throws IOException {
- switch (fileFormat) {
- case PARQUET -> writeParquetWithoutFieldIds(schema, records);
- case AVRO -> writeAvroWithoutFieldIds(schema, records);
- case ORC -> writeOrcWithoutFieldIds(schema, records);
- default -> throw new UnsupportedOperationException("Unsupported file format: " + fileFormat);
- }
- }
-
- private void writeAvroWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
- org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
-
- OutputFile outputFile = encryptedFile.encryptingOutputFile();
- DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
- try (OutputStream out = outputFile.create();
- DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
- writer.create(avroSchemaWithoutIds, out);
- for (Record record : records) {
- GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
- for (Types.NestedField field : schema.columns()) {
- avroRecord.put(field.name(), record.getField(field.name()));
- }
-
- writer.append(avroRecord);
- }
- }
-
- try (DataFileStream<GenericData.Record> reader =
- new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
- assertThat(AvroTestHelpers.hasIds(reader.getSchema())).isFalse();
- }
- }
-
- private void writeParquetWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
- org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
-
- OutputFile outputFile = encryptedFile.encryptingOutputFile();
-
- try (ParquetWriter<GenericData.Record> writer =
- AvroParquetWriter.<GenericData.Record>builder(ParquetFileTestUtils.file(outputFile))
- .withDataModel(GenericData.get())
- .withSchema(avroSchemaWithoutIds)
- .withConf(new Configuration())
- .build()) {
- for (Record record : records) {
- GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
- for (Types.NestedField field : schema.columns()) {
- avroRecord.put(field.name(), record.getField(field.name()));
- }
-
- writer.write(avroRecord);
- }
- }
-
- try (ParquetFileReader reader =
- ParquetFileReader.open(ParquetFileTestUtils.file(outputFile.toInputFile()))) {
- assertThat(ParquetSchemaUtil.hasIds(reader.getFooter().getFileMetaData().getSchema()))
- .isFalse();
- }
- }
-
- private void writeOrcWithoutFieldIds(Schema schema, List<Record> records) throws IOException {
- TypeDescription typeWithIds = ORCSchemaUtil.convert(schema);
- TypeDescription typeWithoutIds = TestORCSchemaUtil.removeIds(typeWithIds);
-
- OutputFile outputFile = encryptedFile.encryptingOutputFile();
- Path hadoopPath = new Path(outputFile.location());
-
- Configuration conf = new Configuration();
- OrcFile.WriterOptions options =
- OrcFile.writerOptions(conf)
- .useUTCTimestamp(true)
- .setSchema(typeWithoutIds)
- .fileSystem(OrcWritingTestUtils.outputFileSystem(outputFile));
-
- OrcRowWriter<Record> rowWriter = GenericOrcWriter.buildWriter(schema, typeWithIds);
-
- try (Writer orcWriter = OrcFile.createWriter(hadoopPath, options)) {
- VectorizedRowBatch batch = typeWithoutIds.createRowBatch();
- for (Record record : records) {
- rowWriter.write(record, batch);
- if (batch.size == batch.getMaxSize()) {
- orcWriter.addRowBatch(batch);
- batch.reset();
- }
- }
-
- if (batch.size > 0) {
- orcWriter.addRowBatch(batch);
- batch.reset();
- }
- }
-
- InputFile inputFile = outputFile.toInputFile();
- OrcFile.ReaderOptions readerOptions =
- OrcFile.readerOptions(conf)
- .useUTCTimestamp(true)
- .filesystem(OrcWritingTestUtils.inputFileSystem(inputFile))
- .maxLength(inputFile.getLength());
-
- try (Reader reader = OrcFile.createReader(hadoopPath, readerOptions)) {
- assertThat(TestORCSchemaUtil.hasIds(reader.getSchema())).isFalse();
- }
- }
-
private void runTypePromotionCheck(
FileFormat fileFormat, Type fromType, Type toType, Function<Object, Object> promoteValue)
throws IOException {
@@ -2257,4 +2140,82 @@ public abstract class BaseFormatModelTests<T> {
assertEquals(
readSchema, convertToEngineRecords(expectedGenericRecords, readSchema), readRecords);
}
+
+ private DataFile writeEngineRecords(FileFormat fileFormat, Schema schema, List<T> records)
+ throws IOException {
+ return writeEngineRecords(fileFormat, schema, records, false /* overwrite */);
+ }
+
+ private DataFile writeEngineRecords(
+ FileFormat fileFormat, Schema schema, List<T> records, boolean overwrite) throws IOException {
+ FileWriterBuilder<DataWriter<T>, Object> writerBuilder =
+ FormatModelRegistry.dataWriteBuilder(fileFormat, engineType(), encryptedFile);
+
+ writerBuilder.schema(schema).spec(PartitionSpec.unpartitioned());
+
+ if (overwrite) {
+ writerBuilder.overwrite();
+ }
+
+ DataWriter<T> writer = writerBuilder.build();
+
+ try (writer) {
+ records.forEach(writer::write);
+ }
+
+ DataFile dataFile = writer.toDataFile();
+ assertThat(dataFile).isNotNull();
+ assertThat(dataFile.recordCount()).isEqualTo(records.size());
+ assertThat(dataFile.format()).isEqualTo(fileFormat);
+
+ return dataFile;
+ }
+
+ private static Map<String, String> testPropertiesToSet(FileFormat fileFormat) {
+ return FileFormatTestSupport.forFormat(fileFormat).testPropertiesToSet();
+ }
+
+ private boolean checkTestProperties(FileFormat fileFormat) throws IOException {
+ return FileFormatTestSupport.forFormat(fileFormat)
+ .checkTestProperties(encryptedFile.encryptingOutputFile().toInputFile());
+ }
+
+ private String fileMetadataValue(FileFormat fileFormat, String key) throws IOException {
+ return FileFormatTestSupport.forFormat(fileFormat)
+ .metadataValue(encryptedFile.encryptingOutputFile().toInputFile(), key);
+ }
+
+ @FunctionalInterface
+ private interface DataWriterEffectAssertion {
+ void accept(FileFormat fileFormat) throws IOException;
+ }
+
+ private void writeAndAssertDataWriterWithConfig(
+ FileFormat fileFormat,
+ BiConsumer<FileWriterBuilder<DataWriter<T>, Object>, FileFormat> configureWriter,
+ DataWriterEffectAssertion assertWriterEffect)
+ throws IOException {
+ DataGenerator dataGenerator = new DataGenerators.DefaultSchema();
+ Schema schema = dataGenerator.schema();
+ List<Record> genericRecords = dataGenerator.generateRecords();
+ List<T> engineRecords = convertToEngineRecords(genericRecords, schema);
+
+ FileWriterBuilder<DataWriter<T>, Object> writerBuilder =
+ FormatModelRegistry.dataWriteBuilder(fileFormat, engineType(), encryptedFile);
+ writerBuilder.schema(schema).spec(PartitionSpec.unpartitioned());
+ configureWriter.accept(writerBuilder, fileFormat);
+
+ DataWriter<T> writer = writerBuilder.build();
+
+ try (writer) {
+ engineRecords.forEach(writer::write);
+ }
+
+ DataFile dataFile = writer.toDataFile();
+ assertThat(dataFile).isNotNull();
+ assertThat(dataFile.recordCount()).isEqualTo(genericRecords.size());
+ assertThat(dataFile.format()).isEqualTo(fileFormat);
+ assertWriterEffect.accept(fileFormat);
+ readAndAssertGenericRecords(fileFormat, schema, genericRecords);
+ }
}
diff --git a/data/src/test/java/org/apache/iceberg/data/FileFormatTestSupport.java b/data/src/test/java/org/apache/iceberg/data/FileFormatTestSupport.java
new file mode 100644
index 0000000000..c3bef8f69c
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/data/FileFormatTestSupport.java
@@ -0,0 +1,61 @@
+/*
+ * 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.iceberg.data;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.avro.AvroFormat;
+import org.apache.iceberg.data.orc.OrcFormat;
+import org.apache.iceberg.data.parquet.ParquetFormat;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+
+public interface FileFormatTestSupport {
+
+ FileFormatTestSupport[] ALL =
+ new FileFormatTestSupport[] {new AvroFormat(), new OrcFormat(), new ParquetFormat()};
+
+ static FileFormat[] formats() {
+ return Arrays.stream(ALL).map(FileFormatTestSupport::format).toArray(FileFormat[]::new);
+ }
+
+ static FileFormatTestSupport forFormat(FileFormat format) {
+ return Arrays.stream(ALL)
+ .filter(testSupportFormat -> testSupportFormat.format() == format)
+ .findFirst()
+ .orElseThrow(() -> new UnsupportedOperationException("Unsupported file format: " + format));
+ }
+
+ FileFormat format();
+
+ void writeRecordsWithoutFieldIds(OutputFile outputFile, Schema schema, List<Record> records)
+ throws IOException;
+
+ Map<String, String> testPropertiesToSet();
+
+ boolean checkTestProperties(InputFile inputFile) throws IOException;
+
+ String metadataValue(InputFile inputFile, String key) throws IOException;
+
+ String splitSizeProperty();
+}
diff --git a/data/src/test/java/org/apache/iceberg/data/avro/AvroFormat.java b/data/src/test/java/org/apache/iceberg/data/avro/AvroFormat.java
new file mode 100644
index 0000000000..e4b8cb3614
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/data/avro/AvroFormat.java
@@ -0,0 +1,101 @@
+/*
+ * 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.iceberg.data.avro;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.List;
+import java.util.Map;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.io.DatumWriter;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.AvroTestHelpers;
+import org.apache.iceberg.data.FileFormatTestSupport;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.types.Types;
+
+public class AvroFormat implements FileFormatTestSupport {
+ @Override
+ public FileFormat format() {
+ return FileFormat.AVRO;
+ }
+
+ @Override
+ public void writeRecordsWithoutFieldIds(
+ OutputFile outputFile, Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+ DatumWriter<GenericData.Record> datumWriter = new GenericDatumWriter<>(avroSchemaWithoutIds);
+ try (OutputStream out = outputFile.create();
+ DataFileWriter<GenericData.Record> writer = new DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchemaWithoutIds, out);
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.append(avroRecord);
+ }
+ }
+
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(outputFile.toInputFile().newStream(), new GenericDatumReader<>())) {
+ assertThat(AvroTestHelpers.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ @Override
+ public Map<String, String> testPropertiesToSet() {
+ // Avro currently has only two configurable writer properties: compression codec and
+ // compression level.
+ return Map.of(
+ TableProperties.AVRO_COMPRESSION_LEVEL, "1", TableProperties.AVRO_COMPRESSION, "snappy");
+ }
+
+ @Override
+ public boolean checkTestProperties(InputFile inputFile) throws IOException {
+ // Only the compression codec round-trips into the file metadata; the compression level is not
+ // persisted and cannot be read back.
+ return "snappy".equals(metadataValue(inputFile, "avro.codec"));
+ }
+
+ @Override
+ public String metadataValue(InputFile inputFile, String key) throws IOException {
+ try (DataFileStream<GenericData.Record> reader =
+ new DataFileStream<>(inputFile.newStream(), new GenericDatumReader<>())) {
+ return reader.getMetaString(key);
+ }
+ }
+
+ @Override
+ public String splitSizeProperty() {
+ throw new UnsupportedOperationException(
+ "No split size property defined for format: " + format());
+ }
+}
diff --git a/data/src/test/java/org/apache/iceberg/data/orc/OrcFormat.java b/data/src/test/java/org/apache/iceberg/data/orc/OrcFormat.java
new file mode 100644
index 0000000000..4da30d370f
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/data/orc/OrcFormat.java
@@ -0,0 +1,130 @@
+/*
+ * 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.iceberg.data.orc;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.data.FileFormatTestSupport;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORCSchemaUtil;
+import org.apache.iceberg.orc.OrcRowWriter;
+import org.apache.iceberg.orc.OrcWritingTestUtils;
+import org.apache.iceberg.orc.TestORCSchemaUtil;
+import org.apache.orc.OrcConf;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+
+public class OrcFormat implements FileFormatTestSupport {
+ @Override
+ public FileFormat format() {
+ return FileFormat.ORC;
+ }
+
+ @Override
+ public void writeRecordsWithoutFieldIds(
+ OutputFile outputFile, Schema schema, List<Record> records) throws IOException {
+ TypeDescription typeWithIds = ORCSchemaUtil.convert(schema);
+ TypeDescription typeWithoutIds = TestORCSchemaUtil.removeIds(typeWithIds);
+ Path hadoopPath = new Path(outputFile.location());
+
+ Configuration conf = new Configuration();
+ OrcFile.WriterOptions options =
+ OrcFile.writerOptions(conf)
+ .useUTCTimestamp(true)
+ .setSchema(typeWithoutIds)
+ .fileSystem(OrcWritingTestUtils.outputFileSystem(outputFile));
+
+ OrcRowWriter<Record> rowWriter = GenericOrcWriter.buildWriter(schema, typeWithIds);
+
+ try (Writer orcWriter = OrcFile.createWriter(hadoopPath, options)) {
+ VectorizedRowBatch batch = typeWithoutIds.createRowBatch();
+ for (Record record : records) {
+ rowWriter.write(record, batch);
+ if (batch.size == batch.getMaxSize()) {
+ orcWriter.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+
+ if (batch.size > 0) {
+ orcWriter.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+
+ try (Reader reader = newOrcReader(outputFile.toInputFile(), conf)) {
+ assertThat(TestORCSchemaUtil.hasIds(reader.getSchema())).isFalse();
+ }
+ }
+
+ @Override
+ public Map<String, String> testPropertiesToSet() {
+ return Map.of(
+ TableProperties.ORC_COMPRESSION, "snappy", OrcConf.ROW_INDEX_STRIDE.getAttribute(), "8192");
+ }
+
+ @Override
+ public boolean checkTestProperties(InputFile inputFile) throws IOException {
+ try (Reader reader = newOrcReader(inputFile, new Configuration())) {
+ return "SNAPPY".equals(reader.getCompressionKind().name())
+ && reader.getRowIndexStride() == 8192;
+ }
+ }
+
+ @Override
+ public String metadataValue(InputFile inputFile, String key) throws IOException {
+ try (Reader reader = newOrcReader(inputFile, new Configuration())) {
+ ByteBuffer metadataValue = reader.getMetadataValue(key).duplicate();
+ byte[] bytes = new byte[metadataValue.remaining()];
+ metadataValue.get(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+ }
+
+ @Override
+ public String splitSizeProperty() {
+ return TableProperties.ORC_STRIPE_SIZE_BYTES;
+ }
+
+ private static Reader newOrcReader(InputFile inputFile, Configuration conf) throws IOException {
+ Path hadoopPath = new Path(inputFile.location());
+ OrcFile.ReaderOptions readerOptions =
+ OrcFile.readerOptions(conf)
+ .useUTCTimestamp(true)
+ .filesystem(OrcWritingTestUtils.inputFileSystem(inputFile))
+ .maxLength(inputFile.getLength());
+
+ return OrcFile.createReader(hadoopPath, readerOptions);
+ }
+}
diff --git a/data/src/test/java/org/apache/iceberg/data/parquet/ParquetFormat.java b/data/src/test/java/org/apache/iceberg/data/parquet/ParquetFormat.java
new file mode 100644
index 0000000000..65edea8e23
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/data/parquet/ParquetFormat.java
@@ -0,0 +1,129 @@
+/*
+ * 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.iceberg.data.parquet;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.avro.generic.GenericData;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.avro.AvroTestHelpers;
+import org.apache.iceberg.data.FileFormatTestSupport;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.parquet.ParquetFileTestUtils;
+import org.apache.iceberg.parquet.ParquetSchemaUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.avro.AvroParquetWriter;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetWriter;
+
+public class ParquetFormat implements FileFormatTestSupport {
+ @Override
+ public FileFormat format() {
+ return FileFormat.PARQUET;
+ }
+
+ @Override
+ public void writeRecordsWithoutFieldIds(
+ OutputFile outputFile, Schema schema, List<Record> records) throws IOException {
+ org.apache.avro.Schema avroSchemaWithoutIds = AvroTestHelpers.removeIds(schema);
+
+ try (ParquetWriter<GenericData.Record> writer =
+ AvroParquetWriter.<GenericData.Record>builder(ParquetFileTestUtils.file(outputFile))
+ .withDataModel(GenericData.get())
+ .withSchema(avroSchemaWithoutIds)
+ .withConf(new Configuration())
+ .build()) {
+ for (Record record : records) {
+ GenericData.Record avroRecord = new GenericData.Record(avroSchemaWithoutIds);
+ for (Types.NestedField field : schema.columns()) {
+ avroRecord.put(field.name(), record.getField(field.name()));
+ }
+
+ writer.write(avroRecord);
+ }
+ }
+
+ try (ParquetFileReader reader =
+ ParquetFileReader.open(ParquetFileTestUtils.file(outputFile.toInputFile()))) {
+ assertThat(ParquetSchemaUtil.hasIds(reader.getFooter().getFileMetaData().getSchema()))
+ .isFalse();
+ }
+ }
+
+ @Override
+ public Map<String, String> testPropertiesToSet() {
+ return Map.of(
+ TableProperties.PARQUET_COMPRESSION,
+ "gzip",
+ TableProperties.PARQUET_PAGE_ROW_LIMIT,
+ "1",
+ TableProperties.PARQUET_ROW_GROUP_CHECK_MIN_RECORD_COUNT,
+ "1");
+ }
+
+ @Override
+ public boolean checkTestProperties(InputFile inputFile) throws IOException {
+ try (ParquetFileReader reader = ParquetFileReader.open(ParquetFileTestUtils.file(inputFile))) {
+ boolean compressionMatches =
+ "GZIP"
+ .equals(reader.getFooter().getBlocks().get(0).getColumns().get(0).getCodec().name());
+
+ return compressionMatches && hasExpectedRowsPerPage(reader, 1);
+ }
+ }
+
+ private boolean hasExpectedRowsPerPage(ParquetFileReader reader, int expectedRowsPerPage)
+ throws IOException {
+ PageReadStore rowGroup;
+
+ while ((rowGroup = reader.readNextRowGroup()) != null) {
+ PageReader pageReader =
+ rowGroup.getPageReader(reader.getFileMetaData().getSchema().getColumns().get(0));
+ DataPage page = pageReader.readPage();
+ if (page != null) {
+ return page.getValueCount() == expectedRowsPerPage;
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public String metadataValue(InputFile inputFile, String key) throws IOException {
+ try (ParquetFileReader reader = ParquetFileReader.open(ParquetFileTestUtils.file(inputFile))) {
+ return reader.getFooter().getFileMetaData().getKeyValueMetaData().get(key);
+ }
+ }
+
+ @Override
+ public String splitSizeProperty() {
+ return TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES;
+ }
+}