Core: Add validation for row-level deletes with rewrites (#2865)

Author: rdblue

api/src/main/java/org/apache/iceberg/RewriteFiles.java

@@ -54,12 +54,23 @@ default RewriteFiles rewriteFiles(Set<DataFile> filesToDelete, Set<DataFile> fil
   /**
    * Add a rewrite that replaces one set of files with another set that contains the same data.
    *
-   * @param dataFilesToDelete   data files that will be replaced (deleted).
-   * @param deleteFilesToDelete delete files that will be replaced (deleted).
+   * @param dataFilesToReplace   data files that will be replaced (deleted).
+   * @param deleteFilesToReplace delete files that will be replaced (deleted).
    * @param dataFilesToAdd      data files that will be added.
    * @param deleteFilesToAdd    delete files that will be added.
    * @return this for method chaining.
    */
-  RewriteFiles rewriteFiles(Set<DataFile> dataFilesToDelete, Set<DeleteFile> deleteFilesToDelete,
+  RewriteFiles rewriteFiles(Set<DataFile> dataFilesToReplace, Set<DeleteFile> deleteFilesToReplace,
                             Set<DataFile> dataFilesToAdd, Set<DeleteFile> deleteFilesToAdd);
+
+  /**
+   * Set the snapshot ID used in any reads for this operation.
+   * <p>
+   * Validations will check changes after this snapshot ID. If this is not called, all ancestor snapshots through the
+   * table's initial snapshot are validated.
+   *
+   * @param snapshotId a snapshot ID
+   * @return this for method chaining
+   */
+  RewriteFiles validateFromSnapshot(long snapshotId);
 }

core/src/main/java/org/apache/iceberg/BaseRewriteFiles.java

@@ -21,8 +21,12 @@
 
 import java.util.Set;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 
 class BaseRewriteFiles extends MergingSnapshotProducer<RewriteFiles> implements RewriteFiles {
+  private final Set<DataFile> replacedDataFiles = Sets.newHashSet();
+  private Long startingSnapshotId = null;
+
   BaseRewriteFiles(String tableName, TableOperations ops) {
     super(tableName, ops);
 
@@ -63,15 +67,16 @@ private void verifyInputAndOutputFiles(Set<DataFile> dataFilesToDelete, Set<Dele
   }
 
   @Override
-  public RewriteFiles rewriteFiles(Set<DataFile> dataFilesToDelete, Set<DeleteFile> deleteFilesToDelete,
+  public RewriteFiles rewriteFiles(Set<DataFile> dataFilesToReplace, Set<DeleteFile> deleteFilesToReplace,
                                    Set<DataFile> dataFilesToAdd, Set<DeleteFile> deleteFilesToAdd) {
-    verifyInputAndOutputFiles(dataFilesToDelete, deleteFilesToDelete, dataFilesToAdd, deleteFilesToAdd);
+    verifyInputAndOutputFiles(dataFilesToReplace, deleteFilesToReplace, dataFilesToAdd, deleteFilesToAdd);
+    replacedDataFiles.addAll(dataFilesToReplace);
 
-    for (DataFile dataFile : dataFilesToDelete) {
+    for (DataFile dataFile : dataFilesToReplace) {
       delete(dataFile);
     }
 
-    for (DeleteFile deleteFile : deleteFilesToDelete) {
+    for (DeleteFile deleteFile : deleteFilesToReplace) {
       delete(deleteFile);
     }
 
@@ -85,4 +90,18 @@ public RewriteFiles rewriteFiles(Set<DataFile> dataFilesToDelete, Set<DeleteFile
 
     return this;
   }
+
+  @Override
+  public RewriteFiles validateFromSnapshot(long snapshotId) {
+    this.startingSnapshotId = snapshotId;
+    return this;
+  }
+
+  @Override
+  protected void validate(TableMetadata base) {
+    if (replacedDataFiles.size() > 0) {
+      // if there are replaced data files, there cannot be any new row-level deletes for those data files
+      validateNoNewDeletesForDataFiles(base, startingSnapshotId, replacedDataFiles);
+    }
+  }
 }

core/src/main/java/org/apache/iceberg/BaseRowDelta.java

@@ -94,6 +94,7 @@ protected void validate(TableMetadata base) {
         validateDataFilesExist(base, startingSnapshotId, referencedDataFiles, !validateDeletes);
       }
 
+      // TODO: does this need to check new delete files?
       if (conflictDetectionFilter != null) {
         validateAddedDataFiles(base, startingSnapshotId, conflictDetectionFilter, caseSensitive);
       }

core/src/main/java/org/apache/iceberg/DeleteFileIndex.java

@@ -312,6 +312,7 @@ static Builder builderFor(FileIO io, Iterable<ManifestFile> deleteManifests) {
   static class Builder {
     private final FileIO io;
     private final Set<ManifestFile> deleteManifests;
+    private long minSequenceNumber = 0L;
     private Map<Integer, PartitionSpec> specsById = null;
     private Expression dataFilter = Expressions.alwaysTrue();
     private Expression partitionFilter = Expressions.alwaysTrue();
@@ -323,6 +324,11 @@ static class Builder {
       this.deleteManifests = Sets.newHashSet(deleteManifests);
     }
 
+    Builder afterSequenceNumber(long seq) {
+      this.minSequenceNumber = seq;
+      return this;
+    }
+
     Builder specsById(Map<Integer, PartitionSpec> newSpecsById) {
       this.specsById = newSpecsById;
       return this;
@@ -357,8 +363,10 @@ DeleteFileIndex build() {
           .run(deleteFile -> {
             try (CloseableIterable<ManifestEntry<DeleteFile>> reader = deleteFile) {
               for (ManifestEntry<DeleteFile> entry : reader) {
-                // copy with stats for better filtering against data file stats
-                deleteEntries.add(entry.copy());
+                if (entry.sequenceNumber() > minSequenceNumber) {
+                  // copy with stats for better filtering against data file stats
+                  deleteEntries.add(entry.copy());
+                }
               }
             } catch (IOException e) {
               throw new RuntimeIOException(e, "Failed to close");

core/src/main/java/org/apache/iceberg/MergingSnapshotProducer.java

@@ -41,6 +41,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.util.CharSequenceSet;
+import org.apache.iceberg.util.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -62,6 +63,9 @@ abstract class MergingSnapshotProducer<ThisT> extends SnapshotProducer<ThisT> {
       ImmutableSet.of(DataOperations.OVERWRITE, DataOperations.REPLACE, DataOperations.DELETE);
   private static final Set<String> VALIDATE_DATA_FILES_EXIST_SKIP_DELETE_OPERATIONS =
       ImmutableSet.of(DataOperations.OVERWRITE, DataOperations.REPLACE);
+  // delete files can be added in "overwrite" or "delete" operations
+  private static final Set<String> VALIDATE_REPLACED_DATA_FILES_OPERATIONS =
+      ImmutableSet.of(DataOperations.OVERWRITE, DataOperations.DELETE);
 
   private final String tableName;
   private final TableOperations ops;
@@ -253,28 +257,10 @@ protected void validateAddedDataFiles(TableMetadata base, Long startingSnapshotI
       return;
     }
 
-    List<ManifestFile> manifests = Lists.newArrayList();
-    Set<Long> newSnapshots = Sets.newHashSet();
-
-    Long currentSnapshotId = base.currentSnapshot().snapshotId();
-    while (currentSnapshotId != null && !currentSnapshotId.equals(startingSnapshotId)) {
-      Snapshot currentSnapshot = ops.current().snapshot(currentSnapshotId);
-
-      ValidationException.check(currentSnapshot != null,
-          "Cannot determine history between starting snapshot %s and current %s",
-          startingSnapshotId, currentSnapshotId);
-
-      if (VALIDATE_ADDED_FILES_OPERATIONS.contains(currentSnapshot.operation())) {
-        newSnapshots.add(currentSnapshotId);
-        for (ManifestFile manifest : currentSnapshot.dataManifests()) {
-          if (manifest.snapshotId() == (long) currentSnapshotId) {
-            manifests.add(manifest);
-          }
-        }
-      }
-
-      currentSnapshotId = currentSnapshot.parentId();
-    }
+    Pair<List<ManifestFile>, Set<Long>> history =
+        validationHistory(base, startingSnapshotId, VALIDATE_ADDED_FILES_OPERATIONS, ManifestContent.DATA);
+    List<ManifestFile> manifests = history.first();
+    Set<Long> newSnapshots = history.second();
 
     ManifestGroup conflictGroup = new ManifestGroup(ops.io(), manifests, ImmutableList.of())
         .caseSensitive(caseSensitive)
@@ -297,6 +283,39 @@ protected void validateAddedDataFiles(TableMetadata base, Long startingSnapshotI
     }
   }
 
+  /**
+   * Validates that no new delete files that must be applied to the given data files have been added to the table since
+   * a starting snapshot.
+   *
+   * @param base table metadata to validate
+   * @param startingSnapshotId id of the snapshot current at the start of the operation
+   * @param dataFiles data files to validate have no new row deletes
+   */
+  protected void validateNoNewDeletesForDataFiles(TableMetadata base, Long startingSnapshotId,
+                                                  Iterable<DataFile> dataFiles) {
+    // if there is no current table state, no files have been added
+    if (base.currentSnapshot() == null) {
+      return;
+    }
+
+    Pair<List<ManifestFile>, Set<Long>> history =
+        validationHistory(base, startingSnapshotId, VALIDATE_REPLACED_DATA_FILES_OPERATIONS, ManifestContent.DELETES);
+    List<ManifestFile> deleteManifests = history.first();
+
+    long startingSequenceNumber = startingSnapshotId == null ? 0 : base.snapshot(startingSnapshotId).sequenceNumber();
+    DeleteFileIndex deletes = DeleteFileIndex.builderFor(ops.io(), deleteManifests)
+        .afterSequenceNumber(startingSequenceNumber)
+        .specsById(ops.current().specsById())
+        .build();
+
+    for (DataFile dataFile : dataFiles) {
+      // if any delete is found that applies to files written in or before the starting snapshot, fail
+      if (deletes.forDataFile(startingSequenceNumber, dataFile).length > 0) {
+        throw new ValidationException("Cannot commit, found new delete for replaced data file: %s", dataFile);
+      }
+    }
+  }
+
   @SuppressWarnings("CollectionUndefinedEquality")
   protected void validateDataFilesExist(TableMetadata base, Long startingSnapshotId,
                                         CharSequenceSet requiredDataFiles, boolean skipDeletes) {
@@ -309,6 +328,31 @@ protected void validateDataFilesExist(TableMetadata base, Long startingSnapshotI
         VALIDATE_DATA_FILES_EXIST_SKIP_DELETE_OPERATIONS :
         VALIDATE_DATA_FILES_EXIST_OPERATIONS;
 
+    Pair<List<ManifestFile>, Set<Long>> history =
+        validationHistory(base, startingSnapshotId, matchingOperations, ManifestContent.DATA);
+    List<ManifestFile> manifests = history.first();
+    Set<Long> newSnapshots = history.second();
+
+    ManifestGroup matchingDeletesGroup = new ManifestGroup(ops.io(), manifests, ImmutableList.of())
+        .filterManifestEntries(entry -> entry.status() != ManifestEntry.Status.ADDED &&
+            newSnapshots.contains(entry.snapshotId()) && requiredDataFiles.contains(entry.file().path()))
+        .specsById(base.specsById())
+        .ignoreExisting();
+
+    try (CloseableIterator<ManifestEntry<DataFile>> deletes = matchingDeletesGroup.entries().iterator()) {
+      if (deletes.hasNext()) {
+        throw new ValidationException("Cannot commit, missing data files: %s",
+            Iterators.toString(Iterators.transform(deletes, entry -> entry.file().path().toString())));
+      }
+
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to validate required files exist", e);
+    }
+  }
+
+  private Pair<List<ManifestFile>, Set<Long>> validationHistory(TableMetadata base, Long startingSnapshotId,
+                                                                Set<String> matchingOperations,
+                                                                ManifestContent content) {
     List<ManifestFile> manifests = Lists.newArrayList();
     Set<Long> newSnapshots = Sets.newHashSet();
 
@@ -322,31 +366,25 @@ protected void validateDataFilesExist(TableMetadata base, Long startingSnapshotI
 
       if (matchingOperations.contains(currentSnapshot.operation())) {
         newSnapshots.add(currentSnapshotId);
-        for (ManifestFile manifest : currentSnapshot.dataManifests()) {
-          if (manifest.snapshotId() == (long) currentSnapshotId) {
-            manifests.add(manifest);
+        if (content == ManifestContent.DATA) {
+          for (ManifestFile manifest : currentSnapshot.dataManifests()) {
+            if (manifest.snapshotId() == (long) currentSnapshotId) {
+              manifests.add(manifest);
+            }
+          }
+        } else {
+          for (ManifestFile manifest : currentSnapshot.deleteManifests()) {
+            if (manifest.snapshotId() == (long) currentSnapshotId) {
+              manifests.add(manifest);
+            }
           }
         }
       }
 
       currentSnapshotId = currentSnapshot.parentId();
     }
 
-    ManifestGroup matchingDeletesGroup = new ManifestGroup(ops.io(), manifests, ImmutableList.of())
-        .filterManifestEntries(entry -> entry.status() != ManifestEntry.Status.ADDED &&
-            newSnapshots.contains(entry.snapshotId()) && requiredDataFiles.contains(entry.file().path()))
-        .specsById(base.specsById())
-        .ignoreExisting();
-
-    try (CloseableIterator<ManifestEntry<DataFile>> deletes = matchingDeletesGroup.entries().iterator()) {
-      if (deletes.hasNext()) {
-        throw new ValidationException("Cannot commit, missing data files: %s",
-            Iterators.toString(Iterators.transform(deletes, entry -> entry.file().path().toString())));
-      }
-
-    } catch (IOException e) {
-      throw new UncheckedIOException("Failed to validate required files exist", e);
-    }
+    return Pair.of(manifests, newSnapshots);
   }
 
   @Override

core/src/main/java/org/apache/iceberg/actions/BaseRewriteDataFilesAction.java

@@ -197,8 +197,14 @@ public BaseRewriteDataFilesAction<ThisT> filter(Expression expr) {
   @Override
   public RewriteDataFilesActionResult execute() {
     CloseableIterable<FileScanTask> fileScanTasks = null;
+    if (table.currentSnapshot() == null) {
+      return RewriteDataFilesActionResult.empty();
+    }
+
+    long startingSnapshotId = table.currentSnapshot().snapshotId();
     try {
       fileScanTasks = table.newScan()
+          .useSnapshot(startingSnapshotId)
           .caseSensitive(caseSensitive)
           .ignoreResiduals()
           .filter(filter)
@@ -241,7 +247,7 @@ public RewriteDataFilesActionResult execute() {
     List<DataFile> currentDataFiles = combinedScanTasks.stream()
         .flatMap(tasks -> tasks.files().stream().map(FileScanTask::file))
         .collect(Collectors.toList());
-    replaceDataFiles(currentDataFiles, addedDataFiles);
+    replaceDataFiles(currentDataFiles, addedDataFiles, startingSnapshotId);
 
     return new RewriteDataFilesActionResult(currentDataFiles, addedDataFiles);
   }
@@ -262,10 +268,12 @@ private Map<StructLikeWrapper, Collection<FileScanTask>> groupTasksByPartition(
     return tasksGroupedByPartition.asMap();
   }
 
-  private void replaceDataFiles(Iterable<DataFile> deletedDataFiles, Iterable<DataFile> addedDataFiles) {
+  private void replaceDataFiles(Iterable<DataFile> deletedDataFiles, Iterable<DataFile> addedDataFiles,
+                                long startingSnapshotId) {
     try {
-      RewriteFiles rewriteFiles = table.newRewrite();
-      rewriteFiles.rewriteFiles(Sets.newHashSet(deletedDataFiles), Sets.newHashSet(addedDataFiles));
+      RewriteFiles rewriteFiles = table.newRewrite()
+          .validateFromSnapshot(startingSnapshotId)
+          .rewriteFiles(Sets.newHashSet(deletedDataFiles), Sets.newHashSet(addedDataFiles));
       commit(rewriteFiles);
     } catch (Exception e) {
       Tasks.foreach(Iterables.transform(addedDataFiles, f -> f.path().toString()))

core/src/main/java/org/apache/iceberg/actions/RewriteDataFilesCommitManager.java

@@ -28,6 +28,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.iceberg.DataFile;
+import org.apache.iceberg.RewriteFiles;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -46,9 +47,16 @@ public class RewriteDataFilesCommitManager {
   private static final Logger LOG = LoggerFactory.getLogger(RewriteDataFilesCommitManager.class);
 
   private final Table table;
+  private final long startingSnapshotId;
 
+  // constructor used for testing
   public RewriteDataFilesCommitManager(Table table) {
+    this(table, table.currentSnapshot().snapshotId());
+  }
+
+  public RewriteDataFilesCommitManager(Table table, long startingSnapshotId) {
     this.table = table;
+    this.startingSnapshotId = startingSnapshotId;
   }
 
   /**
@@ -64,9 +72,10 @@ public void commitFileGroups(Set<RewriteFileGroup> fileGroups) {
       addedDataFiles = Sets.union(addedDataFiles, group.addedFiles());
     }
 
-    table.newRewrite()
-        .rewriteFiles(rewrittenDataFiles, addedDataFiles)
-        .commit();
+    RewriteFiles rewrite = table.newRewrite()
+        .validateFromSnapshot(startingSnapshotId)
+        .rewriteFiles(rewrittenDataFiles, addedDataFiles);
+    rewrite.commit();
   }
 
   /**