diff --git a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java index 4b6776d3396b..a9ec3a4cbea1 100644 --- a/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java +++ b/paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java @@ -1323,10 +1323,15 @@ private Pair, String> createWithDeleteManifestFile for (int i = 0; i < numFields; i++) { IcebergPartitionSummary summary = fileMeta.partitions().get(i); DataType fieldType = partitionType.getTypeAt(i); - minValues.setField( - i, IcebergConversions.toPaimonObject(fieldType, summary.lowerBound())); - maxValues.setField( - i, IcebergConversions.toPaimonObject(fieldType, summary.upperBound())); + // an omitted bound means the value is unknown; keep the slot null + byte[] lowerBound = summary.lowerBound(); + byte[] upperBound = summary.upperBound(); + if (lowerBound != null) { + minValues.setField(i, IcebergConversions.toPaimonObject(fieldType, lowerBound)); + } + if (upperBound != null) { + maxValues.setField(i, IcebergConversions.toPaimonObject(fieldType, upperBound)); + } // IcebergPartitionSummary only has `containsNull` field and does not have the // exact number of nulls. nullCounts[i] = summary.containsNull() ? 1 : 0; 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 d5abb480a587..201b38957031 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 @@ -266,12 +266,15 @@ public IcebergManifestFileMeta result() throws IOException { 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, - toByteBuffer(type, fieldStats.min()).array(), - toByteBuffer(type, fieldStats.max()).array())); + min == null ? null : toByteBuffer(type, min).array(), + max == null ? null : toByteBuffer(type, max).array())); } return new IcebergManifestFileMeta( path.toString(), 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 d8a22aaee494..6c6ceed190db 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 @@ -1371,6 +1371,73 @@ public void testDoublePartitionContainsNan() throws Exception { assertThat(sawNanPartitionSummary).isTrue(); } + @Test + public void testNullPartitionValue() throws Exception { + RowType rowType = + RowType.of( + new DataType[] {DataTypes.VARCHAR(10), DataTypes.INT()}, + new String[] {"pt", "v"}); + FileStoreTable table = + createPaimonTable( + rowType, Collections.singletonList("pt"), Collections.emptyList(), -1); + + String commitUser = UUID.randomUUID().toString(); + TableWriteImpl write = table.newWrite(commitUser); + TableCommitImpl commit = table.newCommit(commitUser); + + write.write(GenericRow.of(BinaryString.fromString("a"), 1), 1); + commit.commit(1, write.prepareCommit(false, 1)); + + // every entry of this manifest has a null partition value, so its partition + // statistics are unknown and the summary must omit both bounds + write.write(GenericRow.of(null, 2), 1); + commit.commit(2, write.prepareCommit(false, 2)); + + FileIO fileIO = table.fileIO(); + IcebergMetadata metadata = + IcebergMetadata.fromPath( + fileIO, new Path(table.location(), "metadata/v2.metadata.json")); + List partitionSummaries = new ArrayList<>(); + try (DataFileReader dataFileReader = + new DataFileReader<>( + new SeekableFileInput(new File(metadata.currentSnapshot().manifestList())), + new GenericDatumReader<>())) { + while (dataFileReader.hasNext()) { + partitionSummaries.add(dataFileReader.next().get("partitions").toString()); + } + } + assertThat(partitionSummaries) + .anySatisfy( + summary -> + assertThat(summary) + .contains("\"contains_null\": true") + .contains("\"lower_bound\": null") + .contains("\"upper_bound\": null")); + // known bounds are still recorded for non-null partition values + assertThat(partitionSummaries) + .anySatisfy( + summary -> + assertThat(summary) + .contains("\"lower_bound\": \"a\"") + .contains("\"upper_bound\": \"a\"")); + + write.write(GenericRow.of(BinaryString.fromString("b"), 3), 1); + commit.commit(3, write.prepareCommit(false, 3)); + + assertThat(getIcebergResult()) + .containsExactlyInAnyOrder("Record(a, 1)", "Record(null, 2)", "Record(b, 3)"); + + // a non add-only commit reads the omitted bounds back from the base manifests + Map partition = new HashMap<>(); + partition.put("pt", "a"); + commit.truncatePartitions(Collections.singletonList(partition)); + + assertThat(getIcebergResult()).containsExactlyInAnyOrder("Record(null, 2)", "Record(b, 3)"); + + write.close(); + commit.close(); + } + @Test public void testStringPartitionNullPadding() throws Exception { RowType rowType =