Core: Support IncrementalChangelogScan with deletes

Author: manuzhang

Diff for: core/src/main/java/org/apache/iceberg/BaseIncrementalChangelogScan.java

@@ -21,15 +21,17 @@
 import java.util.ArrayDeque;
 import java.util.Collection;
 import java.util.Deque;
+import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.iceberg.ManifestGroup.CreateTasksFunction;
 import org.apache.iceberg.ManifestGroup.TaskContext;
 import org.apache.iceberg.io.CloseableIterable;
-import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.SnapshotUtil;
 import org.apache.iceberg.util.TableScanUtil;
 
@@ -63,21 +65,27 @@ protected CloseableIterable<ChangelogScanTask> doPlanFiles(
       return CloseableIterable.empty();
     }
 
-    Set<Long> changelogSnapshotIds = toSnapshotIds(changelogSnapshots);
-
-    Set<ManifestFile> newDataManifests =
-        FluentIterable.from(changelogSnapshots)
-            .transformAndConcat(snapshot -> snapshot.dataManifests(table().io()))
-            .filter(manifest -> changelogSnapshotIds.contains(manifest.snapshotId()))
-            .toSet();
+    Set<ManifestFile> newDataManifests = Sets.newHashSet();
+    Set<ManifestFile> newDeleteManifests = Sets.newHashSet();
+    Map<Long, Snapshot> addedToChangedSnapshots = Maps.newHashMap();
+    for (Snapshot snapshot : changelogSnapshots) {
+      List<ManifestFile> dataManifests = snapshot.dataManifests(table().io());
+      for (ManifestFile manifest : dataManifests) {
+        if (!newDataManifests.contains(manifest)) {
+          addedToChangedSnapshots.put(manifest.snapshotId(), snapshot);
+          newDataManifests.add(manifest);
+        }
+      }
+      newDeleteManifests.addAll(snapshot.deleteManifests(table().io()));
+    }
 
     ManifestGroup manifestGroup =
-        new ManifestGroup(table().io(), newDataManifests, ImmutableList.of())
+        new ManifestGroup(table().io(), newDataManifests, newDeleteManifests)
             .specsById(table().specs())
             .caseSensitive(isCaseSensitive())
             .select(scanColumns())
             .filterData(filter())
-            .filterManifestEntries(entry -> changelogSnapshotIds.contains(entry.snapshotId()))
+            .filterManifestEntries(entry -> addedToChangedSnapshots.containsKey(entry.snapshotId()))
             .ignoreExisting()
             .columnsToKeepStats(columnsToKeepStats());
 
@@ -89,7 +97,8 @@ protected CloseableIterable<ChangelogScanTask> doPlanFiles(
       manifestGroup = manifestGroup.planWith(planExecutor());
     }
 
-    return manifestGroup.plan(new CreateDataFileChangeTasks(changelogSnapshots));
+    return manifestGroup.plan(
+        new CreateDataFileChangeTasks(changelogSnapshots, addedToChangedSnapshots));
   }
 
   @Override
@@ -105,11 +114,6 @@ private Deque<Snapshot> orderedChangelogSnapshots(Long fromIdExcl, long toIdIncl
 
     for (Snapshot snapshot : SnapshotUtil.ancestorsBetween(table(), toIdIncl, fromIdExcl)) {
       if (!snapshot.operation().equals(DataOperations.REPLACE)) {
-        if (!snapshot.deleteManifests(table().io()).isEmpty()) {
-          throw new UnsupportedOperationException(
-              "Delete files are currently not supported in changelog scans");
-        }
-
         changelogSnapshots.addFirst(snapshot);
       }
     }
@@ -134,50 +138,81 @@ private static Map<Long, Integer> computeSnapshotOrdinals(Deque<Snapshot> snapsh
   }
 
   private static class CreateDataFileChangeTasks implements CreateTasksFunction<ChangelogScanTask> {
-    private static final DeleteFile[] NO_DELETES = new DeleteFile[0];
 
     private final Map<Long, Integer> snapshotOrdinals;
+    private final Map<Long, Snapshot> addedToChangedSnapshots;
 
-    CreateDataFileChangeTasks(Deque<Snapshot> snapshots) {
+    CreateDataFileChangeTasks(
+        Deque<Snapshot> snapshots, Map<Long, Snapshot> addedToChangedSnapshots) {
       this.snapshotOrdinals = computeSnapshotOrdinals(snapshots);
+      this.addedToChangedSnapshots = addedToChangedSnapshots;
     }
 
     @Override
     public CloseableIterable<ChangelogScanTask> apply(
         CloseableIterable<ManifestEntry<DataFile>> entries, TaskContext context) {
 
-      return CloseableIterable.transform(
-          entries,
-          entry -> {
-            long commitSnapshotId = entry.snapshotId();
-            int changeOrdinal = snapshotOrdinals.get(commitSnapshotId);
-            DataFile dataFile = entry.file().copy(context.shouldKeepStats());
-
-            switch (entry.status()) {
-              case ADDED:
-                return new BaseAddedRowsScanTask(
-                    changeOrdinal,
-                    commitSnapshotId,
-                    dataFile,
-                    NO_DELETES,
-                    context.schemaAsString(),
-                    context.specAsString(),
-                    context.residuals());
-
-              case DELETED:
-                return new BaseDeletedDataFileScanTask(
-                    changeOrdinal,
-                    commitSnapshotId,
-                    dataFile,
-                    NO_DELETES,
-                    context.schemaAsString(),
-                    context.specAsString(),
-                    context.residuals());
-
-              default:
-                throw new IllegalArgumentException("Unexpected entry status: " + entry.status());
-            }
-          });
+      return CloseableIterable.filter(
+          CloseableIterable.transform(
+              entries,
+              entry -> {
+                long snapshotId = entry.snapshotId();
+                Snapshot snapshot = addedToChangedSnapshots.get(snapshotId);
+                long commitSnapshotId = snapshot.snapshotId();
+                int changeOrdinal = snapshotOrdinals.get(snapshot.snapshotId());
+                DataFile dataFile = entry.file().copy(context.shouldKeepStats());
+                DeleteFile[] deleteFiles = context.deletes().forDataFile(dataFile);
+                List<DeleteFile> addedDeletes = Lists.newArrayList();
+                List<DeleteFile> existingDeletes = Lists.newArrayList();
+                for (DeleteFile file : deleteFiles) {
+                  if (file.dataSequenceNumber() == snapshot.sequenceNumber()) {
+                    addedDeletes.add(file);
+                  } else {
+                    existingDeletes.add(file);
+                  }
+                }
+
+                switch (entry.status()) {
+                  case ADDED:
+                    if (snapshotId == commitSnapshotId) {
+                      return new BaseAddedRowsScanTask(
+                          changeOrdinal,
+                          commitSnapshotId,
+                          dataFile,
+                          addedDeletes.toArray(new DeleteFile[0]),
+                          context.schemaAsString(),
+                          context.specAsString(),
+                          context.residuals());
+                    } else if (deleteFiles.length > 0) {
+                      return new BaseDeletedRowsScanTask(
+                          changeOrdinal,
+                          commitSnapshotId,
+                          dataFile,
+                          addedDeletes.toArray(new DeleteFile[0]),
+                          existingDeletes.toArray(new DeleteFile[0]),
+                          context.schemaAsString(),
+                          context.specAsString(),
+                          context.residuals());
+                    } else {
+                      return null;
+                    }
+
+                  case DELETED:
+                    return new BaseDeletedDataFileScanTask(
+                        changeOrdinal,
+                        commitSnapshotId,
+                        dataFile,
+                        existingDeletes.toArray(new DeleteFile[0]),
+                        context.schemaAsString(),
+                        context.specAsString(),
+                        context.residuals());
+
+                  default:
+                    throw new IllegalArgumentException(
+                        "Unexpected entry status: " + entry.status());
+                }
+              }),
+          Objects::nonNull);
     }
   }
 }

Diff for: core/src/test/java/org/apache/iceberg/TestBaseIncrementalChangelogScan.java

@@ -33,7 +33,6 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
-import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.TestTemplate;
 import org.junit.jupiter.api.extension.ExtendWith;
 
@@ -132,6 +131,57 @@ public void testFileDeletes() {
     assertThat(t1.existingDeletes()).as("Must be no deletes").isEmpty();
   }
 
+  @TestTemplate
+  public void testRowDeletes() {
+    assumeThat(formatVersion).isEqualTo(2);
+
+    table.newFastAppend().appendFile(FILE_A).commit();
+    Snapshot snap1 = table.currentSnapshot();
+
+    table.newRowDelta().addDeletes(FILE_A_DELETES).commit();
+    Snapshot snap2 = table.currentSnapshot();
+
+    table.newRowDelta().addRows(FILE_B).addDeletes(FILE_B_DELETES).commit();
+
+    Snapshot snap3 = table.currentSnapshot();
+
+    table.newDelete().deleteFile(FILE_B).commit();
+    Snapshot snap4 = table.currentSnapshot();
+
+    IncrementalChangelogScan scan =
+        newScan().fromSnapshotExclusive(snap1.snapshotId()).toSnapshot(snap4.snapshotId());
+
+    List<ChangelogScanTask> tasks = plan(scan);
+
+    assertThat(tasks).as("Must have 3 tasks").hasSize(3);
+
+    DeletedRowsScanTask t1 = (DeletedRowsScanTask) Iterables.get(tasks, 0);
+    assertThat(t1.changeOrdinal()).as("Ordinal must match").isEqualTo(0);
+    assertThat(t1.commitSnapshotId()).as("Snapshot must match").isEqualTo(snap2.snapshotId());
+    assertThat(t1.file().path()).as("Data file must match").isEqualTo(FILE_A.path());
+    assertThat(t1.addedDeletes().get(0).path())
+        .as("Delete file must match")
+        .isEqualTo(FILE_A_DELETES.path());
+    assertThat(t1.existingDeletes()).as("Must be no deletes").isEmpty();
+
+    AddedRowsScanTask t2 = (AddedRowsScanTask) Iterables.get(tasks, 1);
+    assertThat(t2.changeOrdinal()).as("Ordinal must match").isEqualTo(1);
+    assertThat(t2.commitSnapshotId()).as("Snapshot must match").isEqualTo(snap3.snapshotId());
+    assertThat(t2.file().path()).as("Data file must match").isEqualTo(FILE_B.path());
+    assertThat(t2.deletes().get(0).path())
+        .as("Delete file must match")
+        .isEqualTo(FILE_B_DELETES.path());
+
+    DeletedDataFileScanTask t3 = (DeletedDataFileScanTask) Iterables.get(tasks, 2);
+    assertThat(t3.changeOrdinal()).as("Ordinal must match").isEqualTo(2);
+    assertThat(t3.commitSnapshotId()).as("Snapshot must match").isEqualTo(snap4.snapshotId());
+    assertThat(t3.file().path()).as("Data file must match").isEqualTo(FILE_B.path());
+    assertThat(t3.existingDeletes()).hasSize(1);
+    assertThat(t3.existingDeletes().get(0).path())
+        .as("Existing delete file must match")
+        .isEqualTo(FILE_B_DELETES.path());
+  }
+
   @TestTemplate
   public void testExistingEntriesInNewDataManifestsAreIgnored() {
     table
@@ -247,19 +297,6 @@ public void testDataFileRewrites() {
     assertThat(t2.deletes()).as("Must be no deletes").isEmpty();
   }
 
-  @TestTemplate
-  public void testDeleteFilesAreNotSupported() {
-    assumeThat(formatVersion).isEqualTo(2);
-
-    table.newFastAppend().appendFile(FILE_A2).appendFile(FILE_B).commit();
-
-    table.newRowDelta().addDeletes(FILE_A2_DELETES).commit();
-
-    Assertions.assertThatThrownBy(() -> plan(newScan()))
-        .isInstanceOf(UnsupportedOperationException.class)
-        .hasMessage("Delete files are currently not supported in changelog scans");
-  }
-
   // plans tasks and reorders them to have deterministic order
   private List<ChangelogScanTask> plan(IncrementalChangelogScan scan) {
     try (CloseableIterable<ChangelogScanTask> tasks = scan.planFiles()) {