From 9492802c5f2ea58e3870a7d02d8811681ac0b5d8 Mon Sep 17 00:00:00 2001 From: zenfenan Date: Tue, 14 Jul 2026 23:48:12 +0200 Subject: [PATCH] feat: Fuse partition and metrics filtering into manifest entry deserialization --- pyiceberg/manifest.py | 36 +++++++++++++++++++++++++++++++++++- pyiceberg/table/__init__.py | 6 +----- 2 files changed, 36 insertions(+), 6 deletions(-) diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 9842f79d8e..cd19e74cac 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -19,7 +19,7 @@ import math import threading from abc import ABC, abstractmethod -from collections.abc import Iterator +from collections.abc import Callable, Iterator from copy import copy from enum import Enum from types import TracebackType @@ -883,6 +883,40 @@ def fetch_manifest_entry(self, io: FileIO, discard_deleted: bool = True) -> list if not discard_deleted or entry.status != ManifestEntryStatus.DELETED ] + def prune_manifest_entry( + self, io: FileIO, partition_filter: Callable[[DataFile], bool], metrics_evaluator: Callable[[DataFile], bool] + ) -> list[ManifestEntry]: + """ + Read manifest entries, applying partition and metrics evaluator during deserialization. + + Unlike fetch_manifest_entry followed by a separate filer pass, this fuses filtering + into the deserialization loop, avoiding the intermediate list allocation for + non-matching entries. + + Args: + io: The FileIO to fetch the file. + partition_filter: Evaluates the entry's partition data. + metrics_evaluator: Evaluates the entry's column-level metrics. + + Returns: + An Iterator of manifest entries matching both filters. + """ + input_file = io.new_input(self.manifest_path) + with AvroFile[ManifestEntry]( + input_file, + MANIFEST_ENTRY_SCHEMAS[DEFAULT_READ_VERSION], + read_types={-1: ManifestEntry, 2: DataFile}, + read_enums={0: ManifestEntryStatus, 101: FileFormat, 134: DataFileContent}, + ) as reader: + result = [] + for entry in reader: + if entry.status != ManifestEntryStatus.DELETED: + _inherit_from_manifest(entry, self) + if partition_filter(entry.data_file) and metrics_evaluator(entry.data_file): + result.append(entry) + + return result + def __eq__(self, other: Any) -> bool: """Return the equality of two instances of the ManifestFile class.""" return self.manifest_path == other.manifest_path if isinstance(other, ManifestFile) else False diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 63b87d290e..a02a607c4c 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -2151,11 +2151,7 @@ def _open_manifest( Returns: A list of ManifestEntry that matches the provided filters. """ - return [ - manifest_entry - for manifest_entry in manifest.fetch_manifest_entry(io, discard_deleted=True) - if partition_filter(manifest_entry.data_file) and metrics_evaluator(manifest_entry.data_file) - ] + return manifest.prune_manifest_entry(io, partition_filter, metrics_evaluator) def _min_sequence_number(manifests: list[ManifestFile]) -> int: