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/05/18 12:16:16 UTC
(iceberg) branch main updated: Flink: Add decimal write/read roundtrip test for FlinkParquetReaders (#16346)
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 ae4dfea6e4 Flink: Add decimal write/read roundtrip test for FlinkParquetReaders (#16346)
ae4dfea6e4 is described below
commit ae4dfea6e40907459c611c99f0c22952177c6f3a
Author: Vova Kolmakov <wo...@gmail.com>
AuthorDate: Mon May 18 19:16:09 2026 +0700
Flink: Add decimal write/read roundtrip test for FlinkParquetReaders (#16346)
---
.../apache/iceberg/data/BaseFormatModelTests.java | 36 ++++++++++++++++++++++
.../org/apache/iceberg/data/DataGenerators.java | 15 ++++++++-
.../iceberg/flink/data/FlinkParquetReaders.java | 1 -
.../iceberg/flink/data/FlinkParquetReaders.java | 1 -
.../iceberg/flink/data/FlinkParquetReaders.java | 1 -
5 files changed, 50 insertions(+), 4 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 a38b025e0f..bdff9f27b2 100644
--- a/data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java
+++ b/data/src/test/java/org/apache/iceberg/data/BaseFormatModelTests.java
@@ -272,6 +272,42 @@ public abstract class BaseFormatModelTests<T> {
assertEquals(schema, convertToEngineRecords(genericRecords, schema), readRecords);
}
+ /** Write with engine type T, read with engine type T */
+ @ParameterizedTest
+ @FieldSource("FORMAT_AND_GENERATOR")
+ 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);
+ }
+
/** Write with engine type T, read with Generic Record */
@ParameterizedTest
@FieldSource("FORMAT_AND_GENERATOR")
diff --git a/data/src/test/java/org/apache/iceberg/data/DataGenerators.java b/data/src/test/java/org/apache/iceberg/data/DataGenerators.java
index 390c0949cb..41c540ca2f 100644
--- a/data/src/test/java/org/apache/iceberg/data/DataGenerators.java
+++ b/data/src/test/java/org/apache/iceberg/data/DataGenerators.java
@@ -29,7 +29,7 @@ import org.apache.iceberg.types.Types;
*/
class DataGenerators {
- static final DataGenerator[] ALL = new DataGenerator[] {new StructOfPrimitive()};
+ static final DataGenerator[] ALL = new DataGenerator[] {new StructOfPrimitive(), new Decimals()};
private DataGenerators() {}
@@ -50,6 +50,19 @@ class DataGenerators {
}
}
+ static class Decimals implements DataGenerator {
+ private final Schema schema =
+ new Schema(
+ required(1, "dec_9_2", Types.DecimalType.of(9, 2)),
+ required(2, "dec_15_3", Types.DecimalType.of(15, 3)),
+ required(3, "dec_38_10", Types.DecimalType.of(38, 10)));
+
+ @Override
+ public Schema schema() {
+ return schema;
+ }
+ }
+
static class DefaultSchema implements DataGenerator {
private final Schema schema =
new Schema(
diff --git a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
index 3de64aa998..9627258ea3 100644
--- a/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
+++ b/flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
@@ -362,7 +362,6 @@ public class FlinkParquetReaders {
public DecimalData read(DecimalData ignored) {
Binary binary = column.nextBinary();
BigDecimal bigDecimal = new BigDecimal(new BigInteger(binary.getBytes()), scale);
- // TODO: need a unit test to write-read-validate decimal via FlinkParquetWrite/Reader
return DecimalData.fromBigDecimal(bigDecimal, precision, scale);
}
}
diff --git a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
index 3de64aa998..9627258ea3 100644
--- a/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
+++ b/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
@@ -362,7 +362,6 @@ public class FlinkParquetReaders {
public DecimalData read(DecimalData ignored) {
Binary binary = column.nextBinary();
BigDecimal bigDecimal = new BigDecimal(new BigInteger(binary.getBytes()), scale);
- // TODO: need a unit test to write-read-validate decimal via FlinkParquetWrite/Reader
return DecimalData.fromBigDecimal(bigDecimal, precision, scale);
}
}
diff --git a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
index 0e6856daa6..49d96d1325 100644
--- a/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
+++ b/flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java
@@ -379,7 +379,6 @@ public class FlinkParquetReaders {
public DecimalData read(DecimalData ignored) {
Binary binary = column.nextBinary();
BigDecimal bigDecimal = new BigDecimal(new BigInteger(binary.getBytes()), scale);
- // TODO: need a unit test to write-read-validate decimal via FlinkParquetWrite/Reader
return DecimalData.fromBigDecimal(bigDecimal, precision, scale);
}
}