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 dd32086552 [iceberg] Fix incorrect snapshot summary totals (#10575)
dd32086552 is described below
commit dd320865523ba93695a0fcea1496363c130bb8d3
Author: Karan Venkatesh <[email protected]>
AuthorDate: Sat Oct 17 00:57:15 2026 +1100
[iceberg] Fix incorrect snapshot summary totals (#10575)
---
.../paimon/iceberg/IcebergCommitCallback.java | 105 +++++++----
.../manifest/IcebergDataFileMetaSerializer.java | 6 +-
.../iceberg/manifest/IcebergManifestFileMeta.java | 4 +
.../paimon/iceberg/IcebergCompatibilityTest.java | 210 +++++++++++++++++++++
.../iceberg/manifest/IcebergDataFileMetaTest.java | 24 +++
.../manifest/IcebergManifestFileMetaTest.java | 101 ++++++++++
.../paimon/core/IcebergDVCompatibilityTest.java | 84 +++++++++
7 files changed, 492 insertions(+), 42 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 1a74f12309..abfc1dd31e 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
@@ -47,6 +47,7 @@ import org.apache.paimon.index.DeletionVectorMeta;
import org.apache.paimon.index.IndexFileHandler;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
+import org.apache.paimon.manifest.FileKind;
import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.manifest.ManifestEntry;
@@ -1067,6 +1068,7 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
// In this case, if `baseMetadata` already contains this file, we
should not add a
// duplicate.
List<IcebergManifestFileMeta> newDataManifestFileMetas;
+ List<IcebergDataFileMeta> deletedFiles = new ArrayList<>();
String operation;
if (isAddOnly) {
// Fast case. We don't need to remove files from `baseMetadata`.
We only need to append
@@ -1083,7 +1085,8 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
modifiedPartitions,
baseDataManifestFileMetas,
snapshotId,
- snapshot.commitKind());
+ snapshot.commitKind(),
+ deletedFiles);
newDataManifestFileMetas = result.getLeft();
operation = result.getRight();
}
@@ -1108,39 +1111,37 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
addedFiles.values().stream().mapToLong(p ->
p.getRight().rowCount()).sum();
metrics.addedFilesSize =
addedFiles.values().stream().mapToLong(p ->
p.getRight().fileSize()).sum();
- metrics.deletedDataFiles = removedFiles.size();
+ metrics.deletedDataFiles = deletedFiles.size();
metrics.deletedRecords =
- removedFiles.values().stream().mapToLong(p ->
p.getRight().rowCount()).sum();
+
deletedFiles.stream().mapToLong(IcebergDataFileMeta::recordCount).sum();
metrics.deletedFilesSize =
- removedFiles.values().stream().mapToLong(p ->
p.getRight().fileSize()).sum();
- metrics.changedPartitionCount = modifiedPartitionsSet.size();
-
- IcebergSnapshot baseSnapshot = baseMetadata.currentSnapshot();
+
deletedFiles.stream().mapToLong(IcebergDataFileMeta::fileSizeInBytes).sum();
+ Set<BinaryRow> changedPartitions = new HashSet<>();
+
addedFiles.values().stream().map(Pair::getLeft).forEach(changedPartitions::add);
+ deletedFiles.stream()
+ .map(file -> removedFiles.get(file.filePath()).getLeft())
+ .forEach(changedPartitions::add);
+ if (needAddDvToIceberg && !indexFiles.isEmpty()) {
+ changedPartitions.addAll(
+ partitionsWithChangedDeleteFiles(indexFiles,
baseDVManifestFileMetas));
+ }
+ metrics.changedPartitionCount = changedPartitions.size();
- Long previousTotalRecordsValue =
- getSummaryLong(baseSnapshot, SNAPSHOT_SUMMARY_TOTAL_RECORDS);
- long previousTotalRecords =
- previousTotalRecordsValue != null
- ? previousTotalRecordsValue
- : computeLiveRowCount(baseDataManifestFileMetas);
- metrics.totalRecords =
- Math.max(0, previousTotalRecords + metrics.addedRecords -
metrics.deletedRecords);
+ metrics.totalDataFiles =
computeLiveFileCount(newDataManifestFileMetas);
+ metrics.totalRecords = computeLiveRowCount(newDataManifestFileMetas);
+ // file sizes are not in the manifest list, so the previous total is
reused when its
+ // file count matches the base manifests
+ IcebergSnapshot baseSnapshot = baseMetadata.currentSnapshot();
Long previousTotalDataFilesValue =
getSummaryLong(baseSnapshot,
SNAPSHOT_SUMMARY_TOTAL_DATA_FILES);
- long previousTotalDataFiles =
- previousTotalDataFilesValue != null
- ? previousTotalDataFilesValue
- : computeLiveFileCount(baseDataManifestFileMetas);
- metrics.totalDataFiles =
- Math.max(
- 0,
- previousTotalDataFiles + metrics.addedDataFiles -
metrics.deletedDataFiles);
-
Long previousTotalFilesSizeValue =
getSummaryLong(baseSnapshot,
SNAPSHOT_SUMMARY_TOTAL_FILES_SIZE);
long previousTotalFilesSize =
previousTotalFilesSizeValue != null
+ && previousTotalDataFilesValue != null
+ && previousTotalDataFilesValue
+ ==
computeLiveFileCount(baseDataManifestFileMetas)
? previousTotalFilesSizeValue
:
computeTotalFilesSizeFromManifests(baseDataManifestFileMetas);
metrics.totalFilesSize =
@@ -1395,7 +1396,8 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
List<BinaryRow> modifiedPartitions,
List<IcebergManifestFileMeta> baseManifestFileMetas,
long currentSnapshotId,
- Snapshot.CommitKind commitKind)
+ Snapshot.CommitKind commitKind,
+ List<IcebergDataFileMeta> deletedFiles)
throws IOException {
String operation = IcebergSnapshotSummary.APPEND.operation();
List<IcebergManifestFileMeta> newManifestFileMetas = new ArrayList<>();
@@ -1465,6 +1467,9 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
for (IcebergManifestEntry entry : sourceEntries) {
if (entry.isLive()) {
boolean removed =
removedFiles.containsKey(entry.file().filePath());
+ if (removed) {
+ deletedFiles.add(entry.file());
+ }
newEntries.add(
new IcebergManifestEntry(
removed
@@ -1888,6 +1893,40 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
icebergDvEntries.iterator(), snapshotId,
IcebergManifestFileMeta.Content.DELETES);
}
+ private Set<BinaryRow> partitionsWithChangedDeleteFiles(
+ List<IndexManifestEntry> indexFiles,
+ List<IcebergManifestFileMeta> baseDvManifestFileMetas) {
+ Set<String> baseDeleteFiles = new HashSet<>();
+ for (IcebergManifestFileMeta meta : baseDvManifestFileMetas) {
+ for (IcebergManifestEntry entry :
+ manifestFile.read(new
Path(meta.manifestPath()).getName())) {
+ if (entry.isLive()) {
+ baseDeleteFiles.add(entry.file().filePath());
+ }
+ }
+ }
+
+ Set<BinaryRow> partitions = new HashSet<>();
+ for (IndexManifestEntry entry : indexFiles) {
+ if (!DELETION_VECTORS_INDEX.equals(entry.indexFile().indexType()))
{
+ continue;
+ }
+ boolean inBase =
baseDeleteFiles.contains(indexFileHandler.filePath(entry).toString());
+ boolean changed;
+ if (entry.kind() == FileKind.ADD) {
+ // a retry passes every index file of the snapshot as ADD,
unchanged ones included
+ LinkedHashMap<String, DeletionVectorMeta> dvRanges =
entry.indexFile().dvRanges();
+ changed = !inBase && dvRanges != null && !dvRanges.isEmpty();
+ } else {
+ changed = inBase;
+ }
+ if (changed) {
+ partitions.add(entry.partition());
+ }
+ }
+ return partitions;
+ }
+
//
-------------------------------------------------------------------------------------
// Snapshot Summary Computation
//
-------------------------------------------------------------------------------------
@@ -1975,23 +2014,11 @@ public class IcebergCommitCallback implements
CommitCallback, TagCallback {
}
private long computeLiveFileCount(List<IcebergManifestFileMeta>
manifestMetas) {
- return manifestMetas.stream()
- .mapToLong(
- meta ->
- meta.addedFilesCount()
- + meta.existingFilesCount()
- - meta.deletedFilesCount())
- .sum();
+ return
manifestMetas.stream().mapToLong(IcebergManifestFileMeta::liveFilesCount).sum();
}
private long computeLiveRowCount(List<IcebergManifestFileMeta>
manifestMetas) {
- return manifestMetas.stream()
- .mapToLong(
- meta ->
- meta.addedRowsCount()
- + meta.existingRowsCount()
- - meta.deletedRowsCount())
- .sum();
+ return
manifestMetas.stream().mapToLong(IcebergManifestFileMeta::liveRowsCount).sum();
}
@Nullable
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
index 5db5dea27d..aa95d29691 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaSerializer.java
@@ -94,9 +94,9 @@ public class IcebergDataFileMetaSerializer extends
ObjectSerializer<IcebergDataF
partSerializer.toBinaryRow(row.getRow(3,
partSerializer.getArity())).copy(),
row.getLong(4),
row.getLong(5),
- nullValueCountsSerializer.copy(row.getMap(6)),
- lowerBoundsSerializer.copy(row.getMap(7)),
- upperBoundsSerializer.copy(row.getMap(8)),
+ row.isNullAt(6) ? null :
nullValueCountsSerializer.copy(row.getMap(6)),
+ row.isNullAt(7) ? null :
lowerBoundsSerializer.copy(row.getMap(7)),
+ row.isNullAt(8) ? null :
upperBoundsSerializer.copy(row.getMap(8)),
row.isNullAt(9) ? null : row.getString(9).toString(),
row.isNullAt(10) ? null : row.getLong(10),
row.isNullAt(11) ? null : row.getLong(11),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
index d5ac07c962..daaeca3196 100644
---
a/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
+++
b/paimon-core/src/main/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMeta.java
@@ -162,6 +162,10 @@ public class IcebergManifestFileMeta {
return deletedRowsCount;
}
+ public long liveFilesCount() {
+ return addedFilesCount + existingFilesCount;
+ }
+
public long liveRowsCount() {
return addedRowsCount + existingRowsCount;
}
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 023df89537..07ab14c501 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
@@ -66,8 +66,11 @@ import org.apache.paimon.types.DataTypeRoot;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.JsonSerdeUtil;
import
org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
@@ -305,6 +308,150 @@ public class IcebergCompatibilityTest {
commit.close();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testSnapshotSummaryTotalsMatchManifests(boolean
deletionVector) throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+ Map<String, String> customOptions = new HashMap<>();
+ if (deletionVector) {
+ customOptions.put(CoreOptions.DELETION_VECTORS_ENABLED.key(),
"true");
+ }
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ customOptions);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() +
"/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ // after the first snapshot, level-0 files are not exported to
Iceberg, and each full
+ // compaction removes them
+ long id = 1;
+ for (int round = 0; round < 3; round++) {
+ for (int i = 0; i < 3; i++) {
+ int k = round * 3 + i;
+ write.write(GenericRow.of(k, k * 10));
+ commit.commit(id, write.prepareCommit(false, id));
+ id++;
+ assertSummaryTotalsMatchManifests(table);
+ }
+ write.write(GenericRow.ofKind(RowKind.DELETE, round * 3, round *
30));
+ write.write(GenericRow.of(round * 3 + 1, -1));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(id, write.prepareCommit(true, id));
+ id++;
+ assertSummaryTotalsMatchManifests(table);
+ }
+ assertThat(getIcebergResult()).hasSize(6);
+
+ write.close();
+ commit.close();
+ }
+
+ @Test
+ public void testSnapshotSummaryIgnoresRemovedFilesNotInIceberg() throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT(),
DataTypes.INT()},
+ new String[] {"pt", "k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.singletonList("pt"),
+ Arrays.asList("pt", "k"),
+ 1,
+ Collections.emptyMap());
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write = table.newWrite(commitUser);
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 1, 10));
+ commit.commit(1, write.prepareCommit(false, 1));
+ // the only file of pt=2 is a level-0 file, which is not exported to
Iceberg
+ write.write(GenericRow.of(2, 1, 20));
+ commit.commit(2, write.prepareCommit(false, 2));
+ assertSummaryTotalsMatchManifests(table);
+
+ Map<String, String> partition = new HashMap<>();
+ partition.put("pt", "2");
+ commit.truncatePartitions(Collections.singletonList(partition));
+ assertSummaryTotalsMatchManifests(table);
+ assertThat(getIcebergResult()).containsExactly("Record(1, 1, 10)");
+
+ write.close();
+ commit.close();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void
testSnapshotSummaryTotalsRepairedWhenBaseTotalsAreUnusable(boolean missing)
+ throws Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT()}, new
String[] {"k", "v"});
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.emptyList(),
+ Collections.singletonList("k"),
+ 1,
+ Collections.emptyMap());
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() +
"/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+
+ write.write(GenericRow.of(1, 10));
+ write.write(GenericRow.of(2, 20));
+ commit.commit(1, write.prepareCommit(false, 1));
+ // leaves DELETED entries in the current manifests
+ write.write(GenericRow.of(3, 30));
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(2, write.prepareCommit(true, 2));
+
+ String[] totals = {"total-data-files", "total-records",
"total-files-size"};
+ long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+ Path metadataPath =
+ new Path(table.location(), "metadata/v" + latestSnapshotId +
".metadata.json");
+ ObjectNode metadataJson =
+ JsonSerdeUtil.asSpecificNodeType(
+ table.fileIO().readFileUtf8(metadataPath),
ObjectNode.class);
+ JsonNode snapshots = metadataJson.get("snapshots");
+ ObjectNode summary = (ObjectNode) snapshots.get(snapshots.size() -
1).get("summary");
+ for (String total : totals) {
+ if (missing) {
+ summary.remove(total);
+ } else {
+ summary.put(total, "0");
+ }
+ }
+ table.fileIO()
+ .overwriteFileUtf8(metadataPath,
JsonSerdeUtil.writeValueAsString(metadataJson));
+
+ write.write(GenericRow.of(4, 40));
+ commit.commit(3, write.prepareCommit(false, 3));
+ assertSummaryTotalsMatchManifests(table);
+
+ write.compact(BinaryRow.EMPTY_ROW, 0, true);
+ commit.commit(4, write.prepareCommit(true, 4));
+ assertSummaryTotalsMatchManifests(table);
+ assertThat(getIcebergResult()).hasSize(4);
+
+ write.close();
+ commit.close();
+ }
+
@Test
public void testDelete() throws Exception {
testDeleteImpl(false);
@@ -3066,6 +3213,69 @@ public class IcebergCompatibilityTest {
icebergTable -> IcebergGenerics.read(icebergTable).build(),
Record::toString);
}
+ private void assertSummaryTotalsMatchManifests(FileStoreTable table)
throws Exception {
+ long latestSnapshotId = table.snapshotManager().latestSnapshotId();
+ IcebergMetadata metadata =
+ IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(
+ table.location(),
+ "metadata/v" + latestSnapshotId +
".metadata.json"));
+ IcebergSnapshot snapshot = metadata.currentSnapshot();
+
+ IcebergPathFactory pathFactory =
+ new IcebergPathFactory(new Path(table.location(), "metadata"));
+ IcebergManifestList manifestList = IcebergManifestList.create(table,
pathFactory);
+ IcebergManifestFile manifestFile = IcebergManifestFile.create(table,
pathFactory);
+
+ long dataFiles = 0;
+ long records = 0;
+ long filesSize = 0;
+ long deletedDataFiles = 0;
+ long deletedRecords = 0;
+ long removedFilesSize = 0;
+ long deleteFiles = 0;
+ long positionDeletes = 0;
+ Set<BinaryRow> changedPartitions = new HashSet<>();
+ for (IcebergManifestFileMeta fileMeta :
+ manifestList.read(new
Path(snapshot.manifestList()).getName())) {
+ for (IcebergManifestEntry entry :
manifestFile.read(fileMeta.manifestPath())) {
+ if (fileMeta.content() ==
IcebergManifestFileMeta.Content.DELETES) {
+ if (entry.isLive()) {
+ deleteFiles++;
+ positionDeletes += entry.file().recordCount();
+ }
+ continue;
+ }
+ if (entry.isLive()) {
+ dataFiles++;
+ records += entry.file().recordCount();
+ filesSize += entry.file().fileSizeInBytes();
+ } else if (entry.snapshotId() == snapshot.snapshotId()) {
+ deletedDataFiles++;
+ deletedRecords += entry.file().recordCount();
+ removedFilesSize += entry.file().fileSizeInBytes();
+ }
+ if (entry.status() != IcebergManifestEntry.Status.EXISTING
+ && entry.snapshotId() == snapshot.snapshotId()) {
+ changedPartitions.add(entry.file().partition());
+ }
+ }
+ }
+
+ assertThat(snapshot.summary().getSummary())
+ .as("summary of Iceberg snapshot %s", snapshot.snapshotId())
+ .containsEntry("total-data-files", String.valueOf(dataFiles))
+ .containsEntry("total-records", String.valueOf(records))
+ .containsEntry("total-files-size", String.valueOf(filesSize))
+ .containsEntry("deleted-data-files",
String.valueOf(deletedDataFiles))
+ .containsEntry("deleted-records",
String.valueOf(deletedRecords))
+ .containsEntry("removed-files-size",
String.valueOf(removedFilesSize))
+ .containsEntry("changed-partition-count",
String.valueOf(changedPartitions.size()))
+ .containsEntry("total-delete-files",
String.valueOf(deleteFiles))
+ .containsEntry("total-position-deletes",
String.valueOf(positionDeletes));
+ }
+
private List<GenericRecord> readPartitionSummaries(FileStoreTable table,
long version)
throws Exception {
IcebergMetadata metadata =
diff --git
a/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaTest.java
b/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaTest.java
index b0b7154b18..d8e6cb171f 100644
---
a/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergDataFileMetaTest.java
@@ -56,6 +56,30 @@ class IcebergDataFileMetaTest {
assertThat(partitionField.get().type()).isEqualTo(partitionType.notNull());
}
+ @Test
+ @DisplayName("Test delete file meta without column stats survives
serialization")
+ void testDeleteFileMetaRoundTrip() {
+ IcebergDataFileMeta meta =
+ IcebergDataFileMeta.createForDeleteFile(
+ IcebergDataFileMeta.Content.POSITION_DELETES,
+ "index-path",
+ "puffin",
+ BinaryRow.EMPTY_ROW,
+ 3,
+ 100,
+ "data-path",
+ 4L,
+ 20L);
+
+ IcebergDataFileMetaSerializer serializer = new
IcebergDataFileMetaSerializer(RowType.of());
+ IcebergDataFileMeta restored =
serializer.fromRow(serializer.toRow(meta));
+
+ assertThat(restored).isEqualTo(meta);
+ assertThat(restored.nullValueCounts()).isNull();
+ assertThat(restored.lowerBounds()).isNull();
+ assertThat(restored.upperBounds()).isNull();
+ }
+
@Test
@DisplayName("Test unknown null count is omitted instead of published as
0")
void testUnknownNullCountOmitted() {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaTest.java
b/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaTest.java
new file mode 100644
index 0000000000..9dee44c0b8
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/iceberg/manifest/IcebergManifestFileMetaTest.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.iceberg.manifest;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.iceberg.IcebergPathFactory;
+import org.apache.paimon.schema.FileSystemSchemaManager;
+import org.apache.paimon.schema.Schema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+import org.apache.paimon.types.DataTypes;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the file and row counts of {@link IcebergManifestFileMeta}. */
+class IcebergManifestFileMetaTest {
+
+ @TempDir java.nio.file.Path tempDir;
+
+ @Test
+ void testLiveCountsExcludeDeletedEntries() throws Exception {
+ FileStoreTable table = createTable();
+ IcebergManifestFile manifestFile =
+ IcebergManifestFile.create(
+ table, new IcebergPathFactory(new
Path(table.location(), "metadata")));
+ List<IcebergManifestEntry> entries =
+ Arrays.asList(
+ entry(IcebergManifestEntry.Status.ADDED,
"added-1.parquet", 1),
+ entry(IcebergManifestEntry.Status.ADDED,
"added-2.parquet", 2),
+ entry(IcebergManifestEntry.Status.EXISTING,
"existing.parquet", 4),
+ entry(IcebergManifestEntry.Status.DELETED,
"deleted-1.parquet", 8),
+ entry(IcebergManifestEntry.Status.DELETED,
"deleted-2.parquet", 16));
+
+ List<IcebergManifestFileMeta> metas =
manifestFile.rollingWrite(entries.iterator(), 2);
+
+ assertThat(metas).hasSize(1);
+ IcebergManifestFileMeta meta = metas.get(0);
+ assertThat(meta.addedFilesCount()).isEqualTo(2);
+ assertThat(meta.existingFilesCount()).isEqualTo(1);
+ assertThat(meta.deletedFilesCount()).isEqualTo(2);
+ assertThat(meta.liveFilesCount()).isEqualTo(3);
+ assertThat(meta.liveRowsCount()).isEqualTo(7);
+ }
+
+ private static IcebergManifestEntry entry(
+ IcebergManifestEntry.Status status, String path, long recordCount)
{
+ GenericMap emptyMap = new GenericMap(Collections.emptyMap());
+ IcebergDataFileMeta file =
+ new IcebergDataFileMeta(
+ IcebergDataFileMeta.Content.DATA,
+ path,
+ "parquet",
+ BinaryRow.EMPTY_ROW,
+ recordCount,
+ 100,
+ emptyMap,
+ emptyMap,
+ emptyMap);
+ return new IcebergManifestEntry(status, 1, 1, 1, file);
+ }
+
+ private FileStoreTable createTable() throws Exception {
+ Schema schema =
+ new Schema(
+ DataTypes.ROW(DataTypes.INT()).getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.emptyMap(),
+ "");
+ LocalFileIO fileIO = LocalFileIO.create();
+ Path path = new Path(tempDir.toUri());
+ new FileSystemSchemaManager(fileIO, path).createTable(schema);
+ return FileStoreTableFactory.create(fileIO, path);
+ }
+}
diff --git
a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergDVCompatibilityTest.java
b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergDVCompatibilityTest.java
index e24109371e..08bb3d094f 100644
---
a/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergDVCompatibilityTest.java
+++
b/paimon-iceberg/src/test/java/org/apache/paimon/core/IcebergDVCompatibilityTest.java
@@ -22,12 +22,15 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.catalog.FileSystemCatalog;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.deletionvectors.DeletionVectorsIndexFile;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.iceberg.IcebergOptions;
+import org.apache.paimon.iceberg.metadata.IcebergMetadata;
+import org.apache.paimon.manifest.IndexManifestEntry;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
@@ -202,6 +205,69 @@ public class IcebergDVCompatibilityTest {
commit.close();
}
+ @Test
+ public void testChangedPartitionCountWithDeletionVectorOnlyChange() throws
Exception {
+ RowType rowType =
+ RowType.of(
+ new DataType[] {DataTypes.INT(), DataTypes.INT(),
DataTypes.INT()},
+ new String[] {"pt", "k", "v"});
+ Map<String, String> customOptions = new HashMap<>();
+ customOptions.put(CoreOptions.DELETION_VECTORS_ENABLED.key(), "true");
+ customOptions.put(CoreOptions.DELETION_VECTOR_BITMAP64.key(), "true");
+ customOptions.put(IcebergOptions.FORMAT_VERSION.key(), "3");
+
+ FileStoreTable table =
+ createPaimonTable(
+ rowType,
+ Collections.singletonList("pt"),
+ Arrays.asList("pt", "k"),
+ 1,
+ customOptions);
+
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write =
+ table.newWrite(commitUser)
+ .withIOManager(new IOManagerImpl(tempDir.toString() +
"/tmp"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+ BinaryRow pt1 = partition(1);
+ BinaryRow pt2 = partition(2);
+
+ for (int pt = 1; pt <= 2; pt++) {
+ for (int k = 1; k <= 3; k++) {
+ write.write(GenericRow.of(pt, k, k));
+ }
+ }
+ write.compact(pt1, 0, true);
+ write.compact(pt2, 0, true);
+ commit.commit(1, write.prepareCommit(true, 1));
+
+ // compact to generate dv index in both partitions
+ write.write(GenericRow.ofKind(RowKind.DELETE, 1, 1, 1));
+ write.write(GenericRow.ofKind(RowKind.DELETE, 2, 1, 1));
+ write.compact(pt1, 0, false);
+ write.compact(pt2, 0, false);
+ commit.commit(2, write.prepareCommit(true, 2));
+ assertThat(
+ table.store().newIndexFileHandler()
+
.scan(DeletionVectorsIndexFile.DELETION_VECTORS_INDEX).stream()
+ .map(IndexManifestEntry::partition)
+ .collect(Collectors.toSet()))
+ .containsExactlyInAnyOrder(pt1, pt2);
+
assertThat(latestIcebergSummary(table)).containsEntry("changed-partition-count",
"2");
+
+ // only the deletion vector of pt=1 changes
+ write.write(GenericRow.ofKind(RowKind.DELETE, 1, 2, 2));
+ write.compact(pt1, 0, false);
+ commit.commit(3, write.prepareCommit(true, 3));
+ validateIcebergResult(
+ Arrays.asList(
+ new Object[] {1, 3, 3}, new Object[] {2, 2, 2}, new
Object[] {2, 3, 3}));
+
assertThat(latestIcebergSummary(table)).containsEntry("changed-partition-count",
"1");
+
+ write.close();
+ commit.close();
+ }
+
@Test
public void testUpgradeFormatVersion() throws Exception {
RowType rowType =
@@ -283,6 +349,24 @@ public class IcebergDVCompatibilityTest {
}
}
+ private static BinaryRow partition(int pt) {
+ BinaryRow partition = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(partition);
+ writer.writeInt(0, pt);
+ writer.complete();
+ return partition;
+ }
+
+ private Map<String, String> latestIcebergSummary(FileStoreTable table) {
+ long snapshotId = table.snapshotManager().latestSnapshotId();
+ return IcebergMetadata.fromPath(
+ table.fileIO(),
+ new Path(table.location(), "metadata/v" + snapshotId +
".metadata.json"))
+ .currentSnapshot()
+ .summary()
+ .getSummary();
+ }
+
private void validateIcebergResult(List<Object[]> expected) throws
Exception {
HadoopCatalog icebergCatalog = new HadoopCatalog(new Configuration(),
tempDir.toString());
TableIdentifier icebergIdentifier = TableIdentifier.of("mydb.db", "t");
