diff --git a/pyiceberg/expressions/visitors.py b/pyiceberg/expressions/visitors.py index 5072d3de11..bf1ff88a3d 100644 --- a/pyiceberg/expressions/visitors.py +++ b/pyiceberg/expressions/visitors.py @@ -14,6 +14,7 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +import copy import math from abc import ABC, abstractmethod from collections.abc import Callable, Mapping @@ -561,8 +562,13 @@ def __init__(self, partition_struct_schema: Schema, partition_filter: BooleanExp def eval(self, manifest: ManifestFile) -> bool: if partitions := manifest.partitions: - self.partition_fields = partitions - return visit(self.partition_filter, self) + # manifest_evaluator hands this one instance to every caller, and callers such as + # _OverwriteFiles._deleted_entries evaluate manifests on a thread pool. Keeping the + # per-manifest summaries off self means a thread switch between storing and reading + # them cannot make this manifest be judged by another manifest's bounds. + visitor = copy.copy(self) + visitor.partition_fields = partitions + return visit(self.partition_filter, visitor) # No partition information return ROWS_MIGHT_MATCH diff --git a/tests/expressions/test_visitors.py b/tests/expressions/test_visitors.py index 2bc77d50e9..5fb4deb5f3 100644 --- a/tests/expressions/test_visitors.py +++ b/tests/expressions/test_visitors.py @@ -17,7 +17,7 @@ # pylint:disable=redefined-outer-name from concurrent.futures import ThreadPoolExecutor -from threading import Event +from threading import Event, get_ident from typing import Any import pytest @@ -1982,3 +1982,62 @@ def test_translate_column_names_missing_column_projected_field_ignores_initial_d # Should evaluate to AlwaysFalse since projected field value doesn't match the expression literal assert translated_expr == AlwaysFalse() + + +def test_manifest_evaluator_judges_each_manifest_by_its_own_summaries(schema: Schema, manifest: ManifestFile) -> None: + """A shared evaluator must not let one manifest's partition summaries decide another's fate. + + manifest_evaluator returns a single _ManifestEvalVisitor, and callers such as + _OverwriteFiles._deleted_entries evaluate the parent snapshot's manifests on a thread pool. + While eval stored the summaries on the instance, a thread switch between storing and reading + them made one manifest be judged by another manifest's bounds. When the mis-judged manifest + held the files being deleted it was skipped, so the files were never found and the overwrite + failed with "Missing required files to delete" even though they were live. + """ + # id is within [INT_MIN_VALUE, INT_MAX_VALUE] in `manifest` and far outside it in `other`. + other = _to_manifest_file( + PartitionFieldSummary.from_args( + contains_null=False, + contains_nan=None, + lower_bound=_to_byte_buffer(IntegerType(), INT_MAX_VALUE + 1000), + upper_bound=_to_byte_buffer(IntegerType(), INT_MAX_VALUE + 2000), + ), + ) + + evaluate = _ManifestEvalVisitor(schema, EqualTo(Reference("id"), INT_MIN_VALUE), case_sensitive=True).eval + + at_first_leaf = Event() + other_evaluated = Event() + threads: dict[str, int] = {} + original_visit_equal = _ManifestEvalVisitor.visit_equal + + def visit_equal_pausing_once(self: Any, *args: Any, **kwargs: Any) -> bool: + # Hold the thread evaluating `manifest` at its first predicate leaf, after it has stored + # the summaries, until the other thread has evaluated `other`. + if get_ident() == threads.get("under_test") and not at_first_leaf.is_set(): + at_first_leaf.set() + assert other_evaluated.wait(timeout=10), "the interleaving never completed" + return original_visit_equal(self, *args, **kwargs) + + def evaluate_manifest() -> bool: + threads["under_test"] = get_ident() + return evaluate(manifest) + + def evaluate_other() -> bool: + assert at_first_leaf.wait(timeout=10), "the manifest under test never reached a leaf" + try: + return evaluate(other) + finally: + other_evaluated.set() + + _ManifestEvalVisitor.visit_equal = visit_equal_pausing_once # type: ignore[method-assign] + try: + with ThreadPoolExecutor(max_workers=2) as executor: + under_test = executor.submit(evaluate_manifest) + concurrent = executor.submit(evaluate_other) + result = under_test.result() + assert not concurrent.result(), "id == INT_MIN_VALUE cannot be in `other`, so it must be skipped" + finally: + _ManifestEvalVisitor.visit_equal = original_visit_equal # type: ignore[method-assign] + + assert result, "`manifest` holds id == INT_MIN_VALUE and must be read, whatever another thread evaluates"