perf: reuse metrics evaluator within manifests

pyiceberg/table/__init__.py

@@ -2673,16 +2673,23 @@ def _build_partition_evaluator(self, spec_id: int) -> Callable[[DataFile], bool]
     def _build_metrics_evaluator(self) -> Callable[[DataFile], bool]:
         schema = self.table_metadata.schema()
         include_empty_files = strtobool(self.options.get("include_empty_files", "false"))
+        evaluator: _InclusiveMetricsEvaluator | None = None
+
+        # This callable is scoped to one manifest task, whose entries are processed
+        # sequentially. Initialize lazily so files rejected by the partition filter
+        # do not pay the metrics-evaluator setup cost.
+        def metrics_evaluator(data_file: DataFile) -> bool:
+            nonlocal evaluator
+            if evaluator is None:
+                evaluator = _InclusiveMetricsEvaluator(
+                    schema,
+                    self.row_filter,
+                    self.case_sensitive,
+                    include_empty_files,
+                )
+            return evaluator.eval(data_file)
 
-        # The lambda created here is run in multiple threads.
-        # So we avoid creating _InclusiveMetricsEvaluator methods bound to a single
-        # shared instance across multiple threads.
-        return lambda data_file: _InclusiveMetricsEvaluator(
-            schema,
-            self.row_filter,
-            self.case_sensitive,
-            include_empty_files,
-        ).eval(data_file)
+        return metrics_evaluator
 
     def _build_residual_evaluator(self, spec_id: int) -> Callable[[DataFile], ResidualEvaluator]:
         spec = self.table_metadata.specs()[spec_id]

tests/benchmark/test_metrics_evaluator_benchmark.py

@@ -0,0 +1,96 @@
+# 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.
+"""Benchmark metrics evaluation across the sequential entries in one manifest.
+
+Run with:
+    uv run pytest tests/benchmark/test_metrics_evaluator_benchmark.py -v -s -m benchmark
+"""
+
+from __future__ import annotations
+
+import statistics
+import timeit
+
+import pytest
+
+from pyiceberg.conversions import to_bytes
+from pyiceberg.expressions import And, BooleanExpression, EqualTo, GreaterThanOrEqual, LessThanOrEqual, Or
+from pyiceberg.manifest import DataFile, FileFormat
+from pyiceberg.schema import Schema
+from pyiceberg.table import ManifestGroupPlanner, Table
+from pyiceberg.types import LongType, NestedField
+
+
+def _combined_filter() -> BooleanExpression:
+    branches: list[BooleanExpression] = []
+    for value in range(11):
+        branch: BooleanExpression = GreaterThanOrEqual("x", 0)
+        for predicate in (
+            LessThanOrEqual("x", 10),
+            EqualTo("y", value),
+            EqualTo("y", value + 1),
+            EqualTo("y", value + 2),
+            EqualTo("y", value + 3),
+        ):
+            branch = And(branch, predicate)
+        branches.append(branch)
+
+    combined = branches[0]
+    for branch in branches[1:]:
+        combined = Or(combined, branch)
+    return combined
+
+
+def _data_file(file_number: int) -> DataFile:
+    long_type = LongType()
+    return DataFile.from_args(
+        file_path=f"s3://bucket/data-{file_number}.parquet",
+        file_format=FileFormat.PARQUET,
+        partition={},
+        record_count=100,
+        file_size_in_bytes=1,
+        value_counts={1: 100, 2: 100},
+        null_value_counts={1: 0, 2: 0},
+        lower_bounds={1: to_bytes(long_type, 0), 2: to_bytes(long_type, 0)},
+        upper_bounds={1: to_bytes(long_type, 10), 2: to_bytes(long_type, 10)},
+    )
+
+
+@pytest.mark.benchmark
+def test_metrics_evaluator_reuse(table_v2: Table) -> None:
+    num_files = 1_000
+    schema = Schema(
+        NestedField(1, "x", LongType(), required=True),
+        NestedField(2, "y", LongType(), required=True),
+        *(NestedField(field_id, f"unused_{field_id}", LongType(), required=False) for field_id in range(3, 103)),
+        schema_id=table_v2.metadata.current_schema_id,
+    )
+    metadata = table_v2.metadata.model_copy(update={"schemas": [schema]})
+    planner = ManifestGroupPlanner(table_metadata=metadata, io=table_v2.io, row_filter=_combined_filter())
+    data_files = [_data_file(file_number) for file_number in range(num_files)]
+
+    def evaluate_files() -> int:
+        metrics_evaluator = planner._build_metrics_evaluator()
+        return sum(metrics_evaluator(data_file) for data_file in data_files)
+
+    assert evaluate_files() == num_files
+    timings = timeit.repeat(evaluate_files, number=1, repeat=3)
+
+    print(
+        f"Evaluated metrics for {num_files} files with a 102-column schema and 66-leaf predicate in "
+        f"{statistics.mean(timings):.3f}s (best: {min(timings):.3f}s)"
+    )

tests/table/test_metrics_evaluator_planning.py

@@ -0,0 +1,92 @@
+# 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 __future__ import annotations
+
+import pytest
+
+import pyiceberg.table as table_module
+from pyiceberg.conversions import to_bytes
+from pyiceberg.expressions import BooleanExpression, EqualTo
+from pyiceberg.manifest import DataFile, FileFormat
+from pyiceberg.schema import Schema
+from pyiceberg.table import ManifestGroupPlanner, Table
+from pyiceberg.types import LongType
+
+
+def _data_file(file_number: int, lower_bound: int | None = None, upper_bound: int | None = None) -> DataFile:
+    long_type = LongType()
+    return DataFile.from_args(
+        file_path=f"s3://bucket/data-{file_number}.parquet",
+        file_format=FileFormat.PARQUET,
+        partition={},
+        record_count=100,
+        file_size_in_bytes=1,
+        value_counts={1: 100},
+        null_value_counts={1: 0},
+        lower_bounds={1: to_bytes(long_type, lower_bound)} if lower_bound is not None else None,
+        upper_bounds={1: to_bytes(long_type, upper_bound)} if upper_bound is not None else None,
+    )
+
+
+def test_build_metrics_evaluator_reuses_one_instance_per_callable(table_v2: Table, monkeypatch: pytest.MonkeyPatch) -> None:
+    class CountingMetricsEvaluator:
+        def __init__(
+            self,
+            schema: Schema,
+            expr: BooleanExpression,
+            case_sensitive: bool = True,
+            include_empty_files: bool = False,
+        ) -> None:
+            self.calls: list[DataFile] = []
+            instances.append(self)
+
+        def eval(self, data_file: DataFile) -> bool:
+            self.calls.append(data_file)
+            return True
+
+    instances: list[CountingMetricsEvaluator] = []
+    monkeypatch.setattr(table_module, "_InclusiveMetricsEvaluator", CountingMetricsEvaluator)
+    planner = ManifestGroupPlanner(table_metadata=table_v2.metadata, io=table_v2.io, row_filter=EqualTo("x", 10))
+    first_file = _data_file(1)
+    second_file = _data_file(2)
+
+    first_callable = planner._build_metrics_evaluator()
+    assert not instances
+    assert first_callable(first_file)
+    assert first_callable(second_file)
+
+    second_callable = planner._build_metrics_evaluator()
+    assert len(instances) == 1
+    assert second_callable(first_file)
+
+    assert len(instances) == 2
+    assert instances[0].calls == [first_file, second_file]
+    assert instances[1].calls == [first_file]
+
+
+def test_reused_metrics_evaluator_replaces_file_state(table_v2: Table) -> None:
+    planner = ManifestGroupPlanner(table_metadata=table_v2.metadata, io=table_v2.io, row_filter=EqualTo("x", 10))
+    metrics_evaluator = planner._build_metrics_evaluator()
+    cannot_match = _data_file(1, lower_bound=0, upper_bound=5)
+    might_match = _data_file(2, lower_bound=10, upper_bound=15)
+    missing_metrics = _data_file(3)
+
+    assert not metrics_evaluator(cannot_match)
+    assert metrics_evaluator(might_match)
+    assert metrics_evaluator(missing_metrics)
+    assert not metrics_evaluator(cannot_match)