Skip to content

Commit fc4e2f1

Browse files
committed
feat: Fuse partition and metrics filtering into manifest entry deserialization
1 parent 2c75523 commit fc4e2f1

2 files changed

Lines changed: 36 additions & 6 deletions

File tree

pyiceberg/manifest.py

Lines changed: 35 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
import math
2020
import threading
2121
from abc import ABC, abstractmethod
22-
from collections.abc import Iterator
22+
from collections.abc import Callable, Iterator
2323
from copy import copy
2424
from enum import Enum
2525
from types import TracebackType
@@ -883,6 +883,40 @@ def fetch_manifest_entry(self, io: FileIO, discard_deleted: bool = True) -> list
883883
if not discard_deleted or entry.status != ManifestEntryStatus.DELETED
884884
]
885885

886+
def prune_manifest_entry(
887+
self, io: FileIO, partition_filter: Callable[[DataFile], bool], metrics_evaluator: Callable[[DataFile], bool]
888+
) -> list[ManifestEntry]:
889+
"""
890+
Read manifest entries, applying partition and metrics evaluator during deserialization.
891+
892+
Unlike fetch_manifest_entry followed by a separate filer pass, this fuses filtering
893+
into the deserialization loop, avoiding the intermediate list allocation for
894+
non-matching entries.
895+
896+
Args:
897+
io: The FileIO to fetch the file.
898+
partition_filter: Evaluates the entry's partition data.
899+
metrics_evaluator: Evaluates the entry's column-level metrics.
900+
901+
Returns:
902+
An Iterator of manifest entries matching both filters.
903+
"""
904+
input_file = io.new_input(self.manifest_path)
905+
with AvroFile[ManifestEntry](
906+
input_file,
907+
MANIFEST_ENTRY_SCHEMAS[DEFAULT_READ_VERSION],
908+
read_types={-1: ManifestEntry, 2: DataFile},
909+
read_enums={0: ManifestEntryStatus, 101: DataFileContent},
910+
) as reader:
911+
result = []
912+
for entry in reader:
913+
if entry.status != ManifestEntryStatus.DELETED:
914+
_inherit_from_manifest(entry, self)
915+
if partition_filter(entry.data_file) and metrics_evaluator(entry.data_file):
916+
result.append(entry)
917+
918+
return result
919+
886920
def __eq__(self, other: Any) -> bool:
887921
"""Return the equality of two instances of the ManifestFile class."""
888922
return self.manifest_path == other.manifest_path if isinstance(other, ManifestFile) else False

pyiceberg/table/__init__.py

Lines changed: 1 addition & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2151,11 +2151,7 @@ def _open_manifest(
21512151
Returns:
21522152
A list of ManifestEntry that matches the provided filters.
21532153
"""
2154-
return [
2155-
manifest_entry
2156-
for manifest_entry in manifest.fetch_manifest_entry(io, discard_deleted=True)
2157-
if partition_filter(manifest_entry.data_file) and metrics_evaluator(manifest_entry.data_file)
2158-
]
2154+
return manifest.prune_manifest_entry(io, partition_filter, metrics_evaluator)
21592155

21602156

21612157
def _min_sequence_number(manifests: list[ManifestFile]) -> int:

0 commit comments

Comments
 (0)