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 761df49c20 [iceberg] Fix partition pruning that skips manifests
holding removed files (#10649)
761df49c20 is described below
commit 761df49c2021622a183322e3c6c63ae353e7304d
Author: Jiajia Li <[email protected]>
AuthorDate: Tue Oct 20 14:49:55 2026 +0800
[iceberg] Fix partition pruning that skips manifests holding removed files
(#10649)
---
.../paimon/iceberg/IcebergCommitCallback.java | 37 ++++-
.../paimon/iceberg/IcebergCompatibilityTest.java | 173 ++++++++++++++++++++-
2 files changed, 202 insertions(+), 8 deletions(-)
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 abfc1dd31e..cf89a26b7a 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
@@ -1405,28 +1405,37 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
RowType partitionType = table.schema().logicalPartitionType();
PartitionPredicate predicate =
PartitionPredicate.fromMultiple(partitionType,
modifiedPartitions);
+ // the fields where a modified partition is NaN, which manifest bounds
never cover
+ boolean[] modifiedNan = new boolean[partitionType.getFieldCount()];
+ for (BinaryRow partition : modifiedPartitions) {
+ for (int i = 0; i < modifiedNan.length; i++) {
+ modifiedNan[i] |= isNaN(partition, i,
partitionType.getTypeAt(i));
+ }
+ }
for (IcebergManifestFileMeta fileMeta : baseManifestFileMetas) {
// use partition predicate to only check modified partitions
int numFields = partitionType.getFieldCount();
GenericRow minValues = new GenericRow(numFields);
GenericRow maxValues = new GenericRow(numFields);
- long[] nullCounts = new long[numFields];
+ Long[] nullCounts = new Long[numFields];
for (int i = 0; i < numFields; i++) {
IcebergPartitionSummary summary = fileMeta.partitions().get(i);
DataType fieldType = partitionType.getTypeAt(i);
- // an omitted bound means the value is unknown; keep the slot
null
- byte[] lowerBound = summary.lowerBound();
- byte[] upperBound = summary.upperBound();
+ // an omitted bound means the value is unknown; keep the slot
null. NaN is never
+ // in the bounds, so they cannot rule out a NaN partition of a
summary that saw NaN
+ boolean boundsMissNan = modifiedNan[i] &&
summary.containsNan();
+ byte[] lowerBound = boundsMissNan ? null :
summary.lowerBound();
+ byte[] upperBound = boundsMissNan ? null :
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;
+ // `contains_null` does not give the number of nulls, so the
count is unknown;
+ // a count of 1 would make a manifest with one live row look
all null
+ nullCounts[i] = summary.containsNull() ? null : 0L;
}
if (predicate == null
@@ -1498,6 +1507,20 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
return Pair.of(newManifestFileMetas, operation);
}
+ private static boolean isNaN(BinaryRow row, int pos, DataType type) {
+ if (row.isNullAt(pos)) {
+ return false;
+ }
+ switch (type.getTypeRoot()) {
+ case FLOAT:
+ return Float.isNaN(row.getFloat(pos));
+ case DOUBLE:
+ return Double.isNaN(row.getDouble(pos));
+ default:
+ return false;
+ }
+ }
+
//
-------------------------------------------------------------------------------------
// Compact
//
-------------------------------------------------------------------------------------
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 07ab14c501..8364c4f6a5 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
@@ -34,6 +34,7 @@ import org.apache.paimon.data.Timestamp;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.iceberg.manifest.IcebergManifestEntry;
import org.apache.paimon.iceberg.manifest.IcebergManifestFile;
@@ -93,6 +94,7 @@ import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
+import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
@@ -110,7 +112,9 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -1575,6 +1579,110 @@ public class IcebergCompatibilityTest {
});
}
+ @Test
+ public void testDropNanPartitionInManifestWithOtherPartitions() throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.DOUBLE(), 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);
+
+ // one manifest holds all partitions, and its bounds [1.0, 2.0] leave
out NaN
+ write.write(GenericRow.of(1.0, 1));
+ write.write(GenericRow.of(Double.NaN, 2));
+ write.write(GenericRow.of(2.0, 3));
+ commit.commit(1, write.prepareCommit(false, 1));
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1.0, 1)", "Record(NaN, 2)",
"Record(2.0, 3)");
+
+ // the bounds still cover a finite partition
+
commit.truncatePartitions(Collections.singletonList(Collections.singletonMap("pt",
"2.0")));
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(1.0, 1)", "Record(NaN, 2)");
+
+
commit.truncatePartitions(Collections.singletonList(Collections.singletonMap("pt",
"NaN")));
+ assertThat(getIcebergResult()).containsExactly("Record(1.0, 1)");
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testPartitionDropSkipsManifestsWhoseBoundsExcludeIt() throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.DOUBLE(), DataTypes.INT()},
+ new String[] {"pt", "v"});
+ ReadCountingFileIO fileIO = new ReadCountingFileIO();
+ FileStoreTable table =
+ createPaimonTable(
+ fileIO,
+ rowType,
+ Collections.singletonList("pt"),
+ Collections.emptyList(),
+ -1,
+ new HashMap<>());
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ // this manifest saw NaN, and its bounds [5.0, 5.0] rule out pt=1.0
+ write.write(GenericRow.of(5.0, 1));
+ write.write(GenericRow.of(Double.NaN, 2));
+ commit.commit(1, write.prepareCommit(false, 1));
+ Set<String> nanManifests = currentManifestNames(table);
+ assertThat(nanManifests).hasSize(1);
+
+ write.write(GenericRow.of(1.0, 3));
+ commit.commit(2, write.prepareCommit(false, 2));
+ Set<String> finiteManifests = currentManifestNames(table);
+
+ // this manifest has no NaN, and its bounds [2.0, 2.0] rule out pt=NaN
+ write.write(GenericRow.of(2.0, 4));
+ commit.commit(3, write.prepareCommit(false, 3));
+ Set<String> finiteOnlyManifests = currentManifestNames(table);
+ finiteOnlyManifests.removeAll(finiteManifests);
+ assertThat(finiteOnlyManifests).hasSize(1);
+
+ fileIO.readCounts.clear();
+
commit.truncatePartitions(Collections.singletonList(Collections.singletonMap("pt",
"1.0")));
+
assertThat(fileIO.readCounts.keySet()).doesNotContainAnyElementsOf(nanManifests);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(5.0, 1)", "Record(NaN, 2)",
"Record(2.0, 4)");
+
+ fileIO.readCounts.clear();
+
commit.truncatePartitions(Collections.singletonList(Collections.singletonMap("pt",
"NaN")));
+
assertThat(fileIO.readCounts.keySet()).doesNotContainAnyElementsOf(finiteOnlyManifests);
+ assertThat(getIcebergResult())
+ .containsExactlyInAnyOrder("Record(5.0, 1)", "Record(2.0, 4)");
+
+ write.close();
+ commit.close();
+ }
+
+ private Set<String> currentManifestNames(FileStoreTable table) {
+ long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(
+ table.location(),
+ "metadata/v" + latestSnapshotId +
".metadata.json"));
+ IcebergPathFactory pathFactory =
+ new IcebergPathFactory(new Path(table.location(), "metadata"));
+ return IcebergManifestList.create(table, pathFactory)
+ .read(new
Path(metadata.currentSnapshot().manifestList()).getName()).stream()
+ .map(meta -> new Path(meta.manifestPath()).getName())
+ .collect(Collectors.toCollection(HashSet::new));
+ }
+
@Test
public void testNullPartitionValue() throws Exception {
RowType rowType =
@@ -1642,6 +1750,39 @@ public class IcebergCompatibilityTest {
commit.close();
}
+ @Test
+ public void testDropPartitionInManifestWithOneLiveRowAndNullPartition()
throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), 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(null, 1));
+ write.write(GenericRow.of(1, 2));
+ commit.commit(1, write.prepareCommit(false, 1));
+
+ // the rewritten manifest still reports contains_null for the deleted
entry, and has a
+ // single live row left
+ Map<String, String> partition = new HashMap<>();
+ partition.put("pt", table.coreOptions().partitionDefaultName());
+ commit.truncatePartitions(Collections.singletonList(partition));
+ assertThat(getIcebergResult()).containsExactly("Record(1, 2)");
+
+ partition.put("pt", "1");
+ commit.truncatePartitions(Collections.singletonList(partition));
+ assertThat(getIcebergResult()).isEmpty();
+
+ write.close();
+ commit.close();
+ }
+
@Test
public void testStringPartitionNullPadding() throws Exception {
RowType rowType =
@@ -3183,7 +3324,23 @@ public class IcebergCompatibilityTest {
int numBuckets,
Map<String, String> customOptions)
throws Exception {
- LocalFileIO fileIO = LocalFileIO.create();
+ return createPaimonTable(
+ LocalFileIO.create(),
+ rowType,
+ partitionKeys,
+ primaryKeys,
+ numBuckets,
+ customOptions);
+ }
+
+ private FileStoreTable createPaimonTable(
+ FileIO fileIO,
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ int numBuckets,
+ Map<String, String> customOptions)
+ throws Exception {
Path path = new Path(tempDir.toString());
Options options = new Options(customOptions);
@@ -3373,4 +3530,18 @@ public class IcebergCompatibilityTest {
IcebergCommitCallback.checkVariantNotPublishable(
RowType.of(new DataType[] {DataTypes.INT()}, new String[]
{"k"}));
}
+
+ /** A {@link LocalFileIO} that counts the reads of each file name. */
+ private static class ReadCountingFileIO extends LocalFileIO {
+
+ private final Map<String, AtomicInteger> readCounts = new
ConcurrentHashMap<>();
+
+ @Override
+ public SeekableInputStream newInputStream(Path path) throws
IOException {
+ readCounts
+ .computeIfAbsent(path.getName(), ignored -> new
AtomicInteger())
+ .incrementAndGet();
+ return super.newInputStream(path);
+ }
+ }
}
