feat: validation_history and ancestors_between

pyiceberg/table/snapshots.py

@@ -25,7 +25,7 @@
 from pydantic import Field, PrivateAttr, model_serializer
 
 from pyiceberg.io import FileIO
-from pyiceberg.manifest import DataFile, DataFileContent, ManifestFile, _manifests
+from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestFile, _manifests
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
 from pyiceberg.schema import Schema
 
@@ -255,6 +255,14 @@ def manifests(self, io: FileIO) -> List[ManifestFile]:
         """Return the manifests for the given snapshot."""
         return list(_manifests(io, self.manifest_list))
 
+    def data_manifests(self, io: FileIO) -> List[ManifestFile]:
+        """Return the data manifests for the given snapshot."""
+        return [manifest for manifest in self.manifests(io) if manifest.content == ManifestContent.DATA]
+
+    def delete_manifests(self, io: FileIO) -> List[ManifestFile]:
+        """Return the delete manifests for the given snapshot."""
+        return [manifest for manifest in self.manifests(io) if manifest.content == ManifestContent.DELETES]
+
 
 class MetadataLogEntry(IcebergBaseModel):
     metadata_file: str = Field(alias="metadata-file")
@@ -429,3 +437,13 @@ def ancestors_of(current_snapshot: Optional[Snapshot], table_metadata: TableMeta
         if snapshot.parent_snapshot_id is None:
             break
         snapshot = table_metadata.snapshot_by_id(snapshot.parent_snapshot_id)
+
+
+def ancestors_between(
+    current_snapshot: Optional[Snapshot], oldest_snapshot: Snapshot, table_metadata: TableMetadata
+) -> Iterable[Snapshot]:
+    """Get the ancestors of and including the given snapshot between the latest and oldest snapshot."""
+    for snapshot in ancestors_of(current_snapshot, table_metadata):
+        if snapshot.snapshot_id == oldest_snapshot.snapshot_id:
+            break
+        yield snapshot

pyiceberg/table/update/validate.py

@@ -0,0 +1,79 @@
+# 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.
+from pyiceberg.manifest import ManifestContent, ManifestFile
+from pyiceberg.table import Table
+from pyiceberg.table.snapshots import Operation, Snapshot, ancestors_between
+
+
+class ValidationException(Exception):
+    """Raised when validation fails."""
+
+
+def validation_history(
+    table: Table,
+    starting_snapshot: Snapshot,
+    matching_operations: set[Operation],
+    manifest_content: ManifestContent,
+    parent_snapshot: Snapshot,
+) -> tuple[list[ManifestFile], set[Snapshot]]:
+    """Return newly added manifests and snapshot IDs between the starting snapshot and parent snapshot.
+
+    Args:
+        table: Table to get the history from
+        starting_snapshot: Starting snapshot
+        matching_operations: Operations to match on
+        manifest_content: Manifest content type to filter
+        parent_snapshot: Parent snapshot to get the history from
+
+    Raises:
+        ValidationException: If no matching snapshot is found or only one snapshot is found
+
+    Returns:
+        List of manifest files and set of snapshots matching conditions
+    """
+    manifests_files: list[ManifestFile] = []
+    snapshots: set[Snapshot] = set()
+
+    last_snapshot = None
+    for snapshot in ancestors_between(starting_snapshot, parent_snapshot, table.metadata):
+        last_snapshot = snapshot
+        summary = snapshot.summary
+        if summary is None:
+            continue
+        if summary.operation in matching_operations:
+            snapshots.add(snapshot)
+            if manifest_content == ManifestContent.DATA:
+                manifests_files.extend(
+                    [
+                        manifest
+                        for manifest in snapshot.data_manifests(table.io)
+                        if manifest.added_snapshot_id == snapshot.snapshot_id
+                    ]
+                )
+            else:
+                manifests_files.extend(
+                    [
+                        manifest
+                        for manifest in snapshot.delete_manifests(table.io)
+                        if manifest.added_snapshot_id == snapshot.snapshot_id
+                    ]
+                )
+
+    if last_snapshot is None or last_snapshot.snapshot_id == starting_snapshot.snapshot_id:
+        raise ValidationException("No matching snapshot found.")
+
+    return manifests_files, snapshots

tests/table/test_init.py

@@ -57,7 +57,6 @@
     Snapshot,
     SnapshotLogEntry,
     Summary,
-    ancestors_of,
 )
 from pyiceberg.table.sorting import (
     NullOrder,
@@ -225,44 +224,6 @@ def test_snapshot_by_timestamp(table_v2: Table) -> None:
     assert table_v2.snapshot_as_of_timestamp(1515100955770, inclusive=False) is None
 
 
-def test_ancestors_of(table_v2: Table) -> None:
-    assert list(ancestors_of(table_v2.current_snapshot(), table_v2.metadata)) == [
-        Snapshot(
-            snapshot_id=3055729675574597004,
-            parent_snapshot_id=3051729675574597004,
-            sequence_number=1,
-            timestamp_ms=1555100955770,
-            manifest_list="s3://a/b/2.avro",
-            summary=Summary(Operation.APPEND),
-            schema_id=1,
-        ),
-        Snapshot(
-            snapshot_id=3051729675574597004,
-            parent_snapshot_id=None,
-            sequence_number=0,
-            timestamp_ms=1515100955770,
-            manifest_list="s3://a/b/1.avro",
-            summary=Summary(Operation.APPEND),
-            schema_id=None,
-        ),
-    ]
-
-
-def test_ancestors_of_recursive_error(table_v2_with_extensive_snapshots: Table) -> None:
-    # Test RecursionError: maximum recursion depth exceeded
-    assert (
-        len(
-            list(
-                ancestors_of(
-                    table_v2_with_extensive_snapshots.current_snapshot(),
-                    table_v2_with_extensive_snapshots.metadata,
-                )
-            )
-        )
-        == 2000
-    )
-
-
 def test_snapshot_by_id_does_not_exist(table_v2: Table) -> None:
     assert table_v2.snapshot_by_id(-1) is None
 

tests/table/test_snapshots.py

@@ -20,7 +20,16 @@
 from pyiceberg.manifest import DataFile, DataFileContent, ManifestContent, ManifestFile
 from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
-from pyiceberg.table.snapshots import Operation, Snapshot, SnapshotSummaryCollector, Summary, update_snapshot_summaries
+from pyiceberg.table import Table
+from pyiceberg.table.snapshots import (
+    Operation,
+    Snapshot,
+    SnapshotSummaryCollector,
+    Summary,
+    ancestors_between,
+    ancestors_of,
+    update_snapshot_summaries,
+)
 from pyiceberg.transforms import IdentityTransform
 from pyiceberg.typedef import Record
 from pyiceberg.types import (
@@ -341,3 +350,57 @@ def test_invalid_type() -> None:
         )
 
     assert "Could not parse summary property total-data-files to an int: abc" in str(e.value)
+
+
+def test_ancestors_of(table_v2: Table) -> None:
+    assert list(ancestors_of(table_v2.current_snapshot(), table_v2.metadata)) == [
+        Snapshot(
+            snapshot_id=3055729675574597004,
+            parent_snapshot_id=3051729675574597004,
+            sequence_number=1,
+            timestamp_ms=1555100955770,
+            manifest_list="s3://a/b/2.avro",
+            summary=Summary(Operation.APPEND),
+            schema_id=1,
+        ),
+        Snapshot(
+            snapshot_id=3051729675574597004,
+            parent_snapshot_id=None,
+            sequence_number=0,
+            timestamp_ms=1515100955770,
+            manifest_list="s3://a/b/1.avro",
+            summary=Summary(Operation.APPEND),
+            schema_id=None,
+        ),
+    ]
+
+
+def test_ancestors_of_recursive_error(table_v2_with_extensive_snapshots: Table) -> None:
+    # Test RecursionError: maximum recursion depth exceeded
+    assert (
+        len(
+            list(
+                ancestors_of(
+                    table_v2_with_extensive_snapshots.current_snapshot(),
+                    table_v2_with_extensive_snapshots.metadata,
+                )
+            )
+        )
+        == 2000
+    )
+
+
+def test_ancestors_between(table_v2_with_extensive_snapshots: Table) -> None:
+    oldest_snapshot = table_v2_with_extensive_snapshots.snapshots()[0]
+    assert (
+        len(
+            list(
+                ancestors_between(
+                    table_v2_with_extensive_snapshots.current_snapshot(),
+                    oldest_snapshot,
+                    table_v2_with_extensive_snapshots.metadata,
+                )
+            )
+        )
+        == 1999
+    )

tests/table/test_validate.py

@@ -0,0 +1,33 @@
+# 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.
+# pylint:disable=redefined-outer-name,eval-used
+from typing import cast
+
+from pyiceberg.manifest import ManifestContent
+from pyiceberg.table import Table
+from pyiceberg.table.snapshots import Operation, Snapshot
+from pyiceberg.table.update.validate import validation_history
+
+
+def test_validation_history(table_v2_with_extensive_snapshots: Table) -> None:
+    """Test the validation history function."""
+    oldest_snapshot = table_v2_with_extensive_snapshots.snapshots()[0]
+    newest_snapshot = cast(Snapshot, table_v2_with_extensive_snapshots.current_snapshot())
+    manifests, snapshots = validation_history(
+        table_v2_with_extensive_snapshots, newest_snapshot, {Operation.APPEND}, ManifestContent.DATA, oldest_snapshot
+    )
+    assert len(snapshots) == 2