-
Notifications
You must be signed in to change notification settings - Fork 302
feat: validation_history
and ancestors_between
#1935
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
Show all changes
29 commits
Select commit
Hold shift + click to select a range
beff92b
feat: validation history
jayceslesar c369720
format
jayceslesar 41bb8a4
almost a working test
jayceslesar 763e9f4
allow content_filter in snapshot.manifests
jayceslesar f200beb
simplify order of arguments to validation_history
jayceslesar 7f6bf9d
simplify return in snapshot.manifests
jayceslesar c63cc55
tests passing
jayceslesar f2f3a88
correct ancestors_between
jayceslesar 74d5569
fix to/from logic and allow optional `to_snapshot` arg in `validation…
jayceslesar 167f9e4
remove a level of nesting with smarter clause
jayceslesar efe50b4
fix bad accessor
jayceslesar 0793713
fix docstring
jayceslesar b10a0bf
Merge branch 'main' into feat/validation-history
jayceslesar d57133e
move `ValidationException` to `exceptions.py`, make `to_snapshot` req…
jayceslesar 6ef1aa2
default to `Operation.OVERWRITE` in `validation_history` if summary i…
jayceslesar fce608f
preserve order of snapshots by changing response to a list instead of…
jayceslesar a6624d9
validation_history and from_ancestor argument alignment
jayceslesar 0763056
raise error on summary operation being none in validation_history
jayceslesar cd0146f
Merge branch 'main' into feat/validation-history
jayceslesar d7c7088
Merge branch 'main' into feat/validation-history
jayceslesar fe8d103
Update pyiceberg/table/update/validate.py
jayceslesar f8c5fee
Update pyiceberg/table/update/validate.py
jayceslesar d95e6ed
formatting
jayceslesar 87ee601
remove emoji (lol)
jayceslesar c16a3e6
vix validation exception check
jayceslesar e272b26
fix bug in ancestors_between
jayceslesar 727845b
break into separate test
jayceslesar e2bba3f
Merge branch 'main' into feat/validation-history
jayceslesar a74d7a9
add test that ensures we raise an error
jayceslesar File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,71 @@ | ||
# 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.exceptions import ValidationException | ||
from pyiceberg.manifest import ManifestContent, ManifestFile | ||
from pyiceberg.table import Table | ||
from pyiceberg.table.snapshots import Operation, Snapshot, ancestors_between | ||
|
||
|
||
def validation_history( | ||
table: Table, | ||
to_snapshot: Snapshot, | ||
from_snapshot: Snapshot, | ||
matching_operations: set[Operation], | ||
manifest_content_filter: ManifestContent, | ||
) -> tuple[list[ManifestFile], set[int]]: | ||
"""Return newly added manifests and snapshot IDs between the starting snapshot and parent snapshot. | ||
|
||
Args: | ||
table: Table to get the history from | ||
to_snapshot: Starting snapshot | ||
from_snapshot: Parent snapshot to get the history from | ||
matching_operations: Operations to match on | ||
manifest_content_filter: Manifest content type to filter | ||
|
||
Raises: | ||
ValidationException: If no matching snapshot is found or only one snapshot is found | ||
|
||
Returns: | ||
List of manifest files and set of snapshots ID's matching conditions | ||
""" | ||
manifests_files: list[ManifestFile] = [] | ||
snapshots: set[int] = set() | ||
|
||
last_snapshot = None | ||
for snapshot in ancestors_between(to_snapshot, from_snapshot, table.metadata): | ||
last_snapshot = snapshot | ||
summary = snapshot.summary | ||
if summary is None: | ||
raise ValidationException(f"No summary found for snapshot {snapshot}!") | ||
if summary.operation not in matching_operations: | ||
continue | ||
|
||
snapshots.add(snapshot.snapshot_id) | ||
# TODO: Maybe do the IO in a separate thread at some point, and collect at the bottom (we can easily merge the sets | ||
manifests_files.extend( | ||
[ | ||
manifest | ||
for manifest in snapshot.manifests(table.io) | ||
if manifest.added_snapshot_id == snapshot.snapshot_id and manifest.content == manifest_content_filter | ||
] | ||
) | ||
|
||
if last_snapshot is not None and last_snapshot.snapshot_id != from_snapshot.snapshot_id: | ||
raise ValidationException("No matching snapshot found.") | ||
|
||
return manifests_files, snapshots |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,138 @@ | ||
# 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 unittest.mock import patch | ||
|
||
import pytest | ||
|
||
from pyiceberg.exceptions import ValidationException | ||
from pyiceberg.io import FileIO | ||
from pyiceberg.manifest import ManifestContent, ManifestFile | ||
from pyiceberg.table import Table | ||
from pyiceberg.table.snapshots import Operation, Snapshot | ||
from pyiceberg.table.update.validate import validation_history | ||
|
||
|
||
@pytest.fixture | ||
def table_v2_with_extensive_snapshots_and_manifests( | ||
table_v2_with_extensive_snapshots: Table, | ||
) -> tuple[Table, dict[int, list[ManifestFile]]]: | ||
"""Fixture to create a table with extensive snapshots and manifests.""" | ||
mock_manifests = {} | ||
|
||
for i, snapshot in enumerate(table_v2_with_extensive_snapshots.snapshots()): | ||
mock_manifest = ManifestFile.from_args( | ||
manifest_path=f"foo/bar/{i}", | ||
manifest_length=1, | ||
partition_spec_id=1, | ||
content=ManifestContent.DATA if i % 2 == 0 else ManifestContent.DELETES, | ||
sequence_number=1, | ||
min_sequence_number=1, | ||
added_snapshot_id=snapshot.snapshot_id, | ||
) | ||
|
||
# Store the manifest for this specific snapshot | ||
mock_manifests[snapshot.snapshot_id] = [mock_manifest] | ||
|
||
return table_v2_with_extensive_snapshots, mock_manifests | ||
|
||
|
||
def test_validation_history(table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]]) -> None: | ||
"""Test the validation history function.""" | ||
table, mock_manifests = table_v2_with_extensive_snapshots_and_manifests | ||
|
||
expected_manifest_data_counts = len([m for m in mock_manifests.values() if m[0].content == ManifestContent.DATA]) | ||
|
||
oldest_snapshot = table.snapshots()[0] | ||
newest_snapshot = cast(Snapshot, table.current_snapshot()) | ||
|
||
def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]: | ||
"""Mock the manifests method to use the snapshot_id for lookup.""" | ||
snapshot_id = self.snapshot_id | ||
if snapshot_id in mock_manifests: | ||
return mock_manifests[snapshot_id] | ||
return [] | ||
|
||
with patch("pyiceberg.table.snapshots.Snapshot.manifests", new=mock_read_manifest_side_effect): | ||
manifests, snapshots = validation_history( | ||
table, | ||
newest_snapshot, | ||
oldest_snapshot, | ||
{Operation.APPEND}, | ||
ManifestContent.DATA, | ||
) | ||
|
||
assert len(manifests) == expected_manifest_data_counts | ||
|
||
|
||
def test_validation_history_fails_on_snapshot_with_no_summary( | ||
table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]], | ||
) -> None: | ||
"""Test the validation history function fails on snapshot with no summary.""" | ||
table, _ = table_v2_with_extensive_snapshots_and_manifests | ||
oldest_snapshot = table.snapshots()[0] | ||
newest_snapshot = cast(Snapshot, table.current_snapshot()) | ||
|
||
# Create a snapshot with no summary | ||
snapshot_with_no_summary = Snapshot( | ||
snapshot_id="1234", | ||
parent_id="5678", | ||
timestamp_ms=0, | ||
operation=Operation.APPEND, | ||
summary=None, | ||
manifest_list="foo/bar", | ||
) | ||
with patch("pyiceberg.table.update.validate.ancestors_between", return_value=[snapshot_with_no_summary]): | ||
with pytest.raises(ValidationException): | ||
validation_history( | ||
table, | ||
newest_snapshot, | ||
oldest_snapshot, | ||
{Operation.APPEND}, | ||
ManifestContent.DATA, | ||
) | ||
|
||
|
||
def test_validation_history_fails_on_from_snapshot_not_matching_last_snapshot( | ||
table_v2_with_extensive_snapshots_and_manifests: tuple[Table, dict[int, list[ManifestFile]]], | ||
) -> None: | ||
"""Test the validation history function fails when from_snapshot doesn't match last_snapshot.""" | ||
table, mock_manifests = table_v2_with_extensive_snapshots_and_manifests | ||
|
||
oldest_snapshot = table.snapshots()[0] | ||
newest_snapshot = cast(Snapshot, table.current_snapshot()) | ||
|
||
def mock_read_manifest_side_effect(self: Snapshot, io: FileIO) -> list[ManifestFile]: | ||
"""Mock the manifests method to use the snapshot_id for lookup.""" | ||
snapshot_id = self.snapshot_id | ||
if snapshot_id in mock_manifests: | ||
return mock_manifests[snapshot_id] | ||
return [] | ||
|
||
missing_oldest_snapshot = table.snapshots()[1:] | ||
|
||
with patch("pyiceberg.table.snapshots.Snapshot.manifests", new=mock_read_manifest_side_effect): | ||
with patch("pyiceberg.table.update.validate.ancestors_between", return_value=missing_oldest_snapshot): | ||
with pytest.raises(ValidationException): | ||
validation_history( | ||
table, | ||
newest_snapshot, | ||
oldest_snapshot, | ||
{Operation.APPEND}, | ||
ManifestContent.DATA, | ||
) |
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.