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 a398e4e533 [iceberg] Exclude NaN from manifest partition bounds (#9188)
a398e4e533 is described below
commit a398e4e5332b6a41cab5743019e41ea4f556a3b0
Author: Arnav Balyan <[email protected]>
AuthorDate: Thu Aug 13 19:29:57 2026 +0530
[iceberg] Exclude NaN from manifest partition bounds (#9188)
---
.../iceberg/manifest/IcebergManifestFile.java | 55 ++++++++++++------
.../paimon/iceberg/IcebergCompatibilityTest.java | 67 +++++++++++++++-------
2 files changed, 81 insertions(+), 41 deletions(-)
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 201b389570..f2f11a697c 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
@@ -20,6 +20,7 @@ package org.apache.paimon.iceberg.manifest;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.serializer.Serializer;
import org.apache.paimon.format.FileFormat;
import org.apache.paimon.format.FormatReaderFactory;
import org.apache.paimon.format.FormatWriterFactory;
@@ -36,6 +37,8 @@ import org.apache.paimon.io.SingleFileWriter;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.statistics.FullSimpleColStatsCollector;
+import org.apache.paimon.statistics.SimpleColStatsCollector;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.RowType;
@@ -194,6 +197,7 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
extends SingleFileWriter<IcebergManifestEntry,
IcebergManifestFileMeta> {
private final SimpleStatsCollector partitionStatsCollector;
+ private final IcebergPartitionStatsCollector[]
partitionStatsCollectors;
private final long sequenceNumber;
private int addedFilesCount = 0;
@@ -219,7 +223,21 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
serializer::toRow,
fileCompression,
false);
- this.partitionStatsCollector = new
SimpleStatsCollector(partitionType);
+ this.partitionStatsCollectors =
+ new
IcebergPartitionStatsCollector[partitionType.getFieldCount()];
+ SimpleColStatsCollector.Factory[] statsFactories =
+ new
SimpleColStatsCollector.Factory[partitionType.getFieldCount()];
+ for (int i = 0; i < partitionType.getFieldCount(); i++) {
+ int position = i;
+ statsFactories[i] =
+ () -> {
+ IcebergPartitionStatsCollector collector =
+ new IcebergPartitionStatsCollector();
+ partitionStatsCollectors[position] = collector;
+ return collector;
+ };
+ }
+ this.partitionStatsCollector = new
SimpleStatsCollector(partitionType, statsFactories);
this.sequenceNumber = sequenceNumber;
this.content = content;
}
@@ -257,22 +275,12 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
for (int i = 0; i < stats.length; i++) {
SimpleColStats fieldStats = stats[i];
DataType type = partitionType.getTypeAt(i);
- boolean containsNan = false;
- switch (type.getTypeRoot()) {
- case FLOAT:
- case DOUBLE:
- containsNan = isNaN(fieldStats.min()) ||
isNaN(fieldStats.max());
- break;
- default:
- // contains_nan is only meaningful for FLOAT/DOUBLE
per the Iceberg spec
- }
- // an unknown bound must be omitted, not published as a value
Object min = fieldStats.min();
Object max = fieldStats.max();
partitionSummaries.add(
new IcebergPartitionSummary(
Objects.requireNonNull(fieldStats.nullCount())
> 0,
- containsNan,
+ partitionStatsCollectors[i].containsNan(),
min == null ? null : toByteBuffer(type,
min).array(),
max == null ? null : toByteBuffer(type,
max).array()));
}
@@ -292,15 +300,24 @@ public class IcebergManifestFile extends
ObjectsFile<IcebergManifestEntry> {
deletedRowsCount,
partitionSummaries);
}
+ }
- private boolean isNaN(@Nullable Object value) {
- if (value instanceof Float) {
- return Float.isNaN((Float) value);
- }
- if (value instanceof Double) {
- return Double.isNaN((Double) value);
+ private static class IcebergPartitionStatsCollector extends
FullSimpleColStatsCollector {
+
+ private boolean containsNan;
+
+ @Override
+ public void collect(Object field, Serializer<Object> fieldSerializer) {
+ if ((field instanceof Float && Float.isNaN((Float) field))
+ || (field instanceof Double && Double.isNaN((Double)
field))) {
+ containsNan = true;
+ return;
}
- return false;
+ super.collect(field, fieldSerializer);
+ }
+
+ private boolean containsNan() {
+ return containsNan;
}
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
index 6c6ceed190..6d1d9fc886 100644
---
a/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/iceberg/IcebergCompatibilityTest.java
@@ -82,6 +82,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalDateTime;
@@ -1344,31 +1345,27 @@ public class IcebergCompatibilityTest {
write.write(GenericRow.of(2.0, 200), 1);
write.write(GenericRow.of(Double.NaN, 300), 1);
commit.commit(1, write.prepareCommit(false, 1));
- write.close();
- commit.close();
- FileIO fileIO = table.fileIO();
- IcebergMetadata metadata =
- IcebergMetadata.fromPath(
- fileIO, new Path(table.location(),
"metadata/v1.metadata.json"));
-
- String currentSnapshotManifest =
metadata.currentSnapshot().manifestList();
- File snapShotAvroFile = new File(currentSnapshotManifest);
+ assertThat(readPartitionSummaries(table, 1))
+ .anySatisfy(
+ summary -> {
+
assertThat(summary.get("contains_nan")).isEqualTo(true);
+ assertThat(readDoubleBound(summary,
"lower_bound")).isEqualTo(1.0);
+ assertThat(readDoubleBound(summary,
"upper_bound")).isEqualTo(2.0);
+ });
- boolean sawNanPartitionSummary = false;
- try (DataFileReader<GenericRecord> dataFileReader =
- new DataFileReader<>(
- new SeekableFileInput(snapShotAvroFile), new
GenericDatumReader<>())) {
- while (dataFileReader.hasNext()) {
- GenericRecord record = dataFileReader.next();
- String partitionSummary = record.get("partitions").toString();
- if (partitionSummary.contains("contains_nan\": true")) {
- sawNanPartitionSummary = true;
- }
- }
- }
+ write.write(GenericRow.of(Double.NaN, 400), 1);
+ commit.commit(2, write.prepareCommit(false, 2));
+ write.close();
+ commit.close();
- assertThat(sawNanPartitionSummary).isTrue();
+ assertThat(readPartitionSummaries(table, 2))
+ .anySatisfy(
+ summary -> {
+
assertThat(summary.get("contains_nan")).isEqualTo(true);
+ assertThat(summary.get("lower_bound")).isNull();
+ assertThat(summary.get("upper_bound")).isNull();
+ });
}
@Test
@@ -2650,6 +2647,32 @@ public class IcebergCompatibilityTest {
icebergTable -> IcebergGenerics.read(icebergTable).build(),
Record::toString);
}
+ private List<GenericRecord> readPartitionSummaries(FileStoreTable table,
long version)
+ throws Exception {
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(table.location(), "metadata/v" + version +
".metadata.json"));
+ List<GenericRecord> partitionSummaries = new ArrayList<>();
+ try (DataFileReader<GenericRecord> dataFileReader =
+ new DataFileReader<>(
+ new SeekableFileInput(new
File(metadata.currentSnapshot().manifestList())),
+ new GenericDatumReader<>())) {
+ while (dataFileReader.hasNext()) {
+ GenericRecord record = dataFileReader.next();
+ partitionSummaries.add((GenericRecord) ((List<?>)
record.get("partitions")).get(0));
+ }
+ }
+ return partitionSummaries;
+ }
+
+ private double readDoubleBound(GenericRecord partitionSummary, String
field) {
+ return ((ByteBuffer) partitionSummary.get(field))
+ .duplicate()
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .getDouble();
+ }
+
private org.apache.iceberg.Table getIcebergTable() {
HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(),
tempDir.toString());
TableIdentifier icebergIdentifier = TableIdentifier.of("mydb.db", "t");
