From 68a98141d845ff5225eacf621df4fcce0c657692 Mon Sep 17 00:00:00 2001 From: Aaron Niskode-Dossett Date: Tue, 14 Jul 2026 13:25:25 -0500 Subject: [PATCH] perf: reuse metrics evaluator within manifests --- pyiceberg/table/__init__.py | 25 +++-- .../test_metrics_evaluator_benchmark.py | 96 +++++++++++++++++++ .../table/test_metrics_evaluator_planning.py | 92 ++++++++++++++++++ 3 files changed, 204 insertions(+), 9 deletions(-) create mode 100644 tests/benchmark/test_metrics_evaluator_benchmark.py create mode 100644 tests/table/test_metrics_evaluator_planning.py diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 63b87d290e..85e708195a 100644 --- a/pyiceberg/table/__init__.py +++ b/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] diff --git a/tests/benchmark/test_metrics_evaluator_benchmark.py b/tests/benchmark/test_metrics_evaluator_benchmark.py new file mode 100644 index 0000000000..24ba56e993 --- /dev/null +++ b/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)" + ) diff --git a/tests/table/test_metrics_evaluator_planning.py b/tests/table/test_metrics_evaluator_planning.py new file mode 100644 index 0000000000..ee07d346be --- /dev/null +++ b/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)