[iceberg] Omit unknown bounds in Iceberg manifest partition summary

paimon-core/src/main/java/org/apache/paimon/iceberg/IcebergCommitCallback.java

@@ -1323,10 +1323,15 @@ private Pair<List<IcebergManifestFileMeta>, 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;

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(),

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<String> partitionSummaries = new ArrayList<>();
+        try (DataFileReader<GenericRecord> 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<String, String> 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 =