From e7bf4b1e9a3f59041f0997778f1b3c8928cb9b79 Mon Sep 17 00:00:00 2001 From: Pralay Ghosh Date: Fri, 2 Oct 2026 11:55:00 +0100 Subject: [PATCH] fix(expressions): stop a shared manifest evaluator from mixing up manifests manifest_evaluator returns one _ManifestEvalVisitor, and callers such as _OverwriteFiles._deleted_entries evaluate the parent snapshot's manifests on the ExecutorFactory thread pool. eval stored the manifest's partition summaries on the instance before visiting the filter, so 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, those files were never found, and the commit failed with "Missing required files to delete" although the files were live. Evaluating on a shallow copy keeps the bound filter shared and read-only while giving each call its own summaries. Closes #4031 --- pyiceberg/expressions/visitors.py | 10 ++++- tests/expressions/test_visitors.py | 61 +++++++++++++++++++++++++++++- 2 files changed, 68 insertions(+), 3 deletions(-) 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"