[wip] feat: validate_deleted_data_files

Author: jayceslesar

pyiceberg/table/update/validate.py

@@ -14,11 +14,16 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from typing import Optional
+from typing import Iterator, Optional
 
-from pyiceberg.manifest import ManifestContent, ManifestFile
+from pyiceberg.expressions import BooleanExpression
+from pyiceberg.expressions.visitors import _StrictMetricsEvaluator
+from pyiceberg.manifest import ManifestContent, ManifestEntry, ManifestEntryStatus, ManifestFile
 from pyiceberg.table import Table
 from pyiceberg.table.snapshots import Operation, Snapshot, ancestors_between
+from pyiceberg.typedef import Record
+
+VALIDATE_DATA_FILES_EXIST_OPERATIONS = {Operation.OVERWRITE, Operation.REPLACE, Operation.DELETE}
 
 
 class ValidationException(Exception):
@@ -70,3 +75,76 @@ def validation_history(
         raise ValidationException("No matching snapshot found.")
 
     return manifests_files, snapshots
+
+
+def deleted_data_files(
+    table: Table,
+    starting_snapshot: Snapshot,
+    data_filter: Optional[BooleanExpression],
+    parent_snapshot: Optional[Snapshot],
+    partition_set: Optional[set[Record]],
+) -> Iterator[ManifestEntry]:
+    """Find deleted data files matching a filter since a starting snapshot.
+
+    Args:
+        table: Table to validate
+        starting_snapshot: Snapshot current at the start of the operation
+        data_filter: Expression used to find deleted data files
+        partition_set: a set of partitions to find deleted data files
+        parent_snapshot: Ending snapshot on the branch being validated
+
+    Returns:
+        List of deleted data files matching the filter
+    """
+    # if there is no current table state, no files have been deleted
+    if parent_snapshot is None:
+        return
+
+    manifests, new_snapshots = validation_history(
+        table,
+        starting_snapshot,
+        parent_snapshot,
+        VALIDATE_DATA_FILES_EXIST_OPERATIONS,
+        ManifestContent.DATA,
+    )
+
+    if data_filter is not None:
+        evaluator = _StrictMetricsEvaluator(table.schema(), data_filter).eval
+
+    new_snapshot_ids = {s.snapshot_id for s in new_snapshots}
+
+    for manifest in manifests:
+        for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=False):
+            if entry.snapshot_id not in new_snapshot_ids:
+                continue
+
+            if entry.status != ManifestEntryStatus.DELETED:
+                continue
+
+            if data_filter is not None and not evaluator(entry.data_file):
+                continue
+
+            if partition_set is not None and (entry.data_file.spec_id, entry.data_file.partition) not in partition_set:
+                continue
+
+            yield entry
+
+
+def validate_deleted_data_files(
+    table: Table,
+    starting_snapshot: Snapshot,
+    data_filter: Optional[BooleanExpression],
+    parent_snapshot: Snapshot,
+) -> None:
+    """Validate that no files matching a filter have been deleted from the table since a starting snapshot.
+
+    Args:
+        table: Table to validate
+        starting_snapshot: Snapshot current at the start of the operation
+        data_filter: Expression used to find deleted data files
+        parent_snapshot: Ending snapshot on the branch being validated
+
+    """
+    conflicting_entries = deleted_data_files(table, starting_snapshot, data_filter, None, parent_snapshot)
+    if any(conflicting_entries):
+        raise ValidationException("Deleted data files were found matching the filter.")