Refactor the helper function

Author: Hanzhi Wang

pyiceberg/table/__init__.py

@@ -1943,7 +1943,7 @@ def _check_sequence_number(min_sequence_number: int, manifest: ManifestFile) ->
             and (manifest.sequence_number or INITIAL_SEQUENCE_NUMBER) >= min_sequence_number
         )
 
-    def scan_plan_helper(self) -> Iterator[ManifestEntry]:
+    def scan_plan_helper(self) -> Iterator[List[ManifestEntry]]:
         """Filter and return manifest entries based on partition and metrics evaluators.
 
         Returns:
@@ -1973,20 +1973,18 @@ def scan_plan_helper(self) -> Iterator[ManifestEntry]:
 
         executor = ExecutorFactory.get_or_create()
 
-        return chain(
-            *executor.map(
-                lambda args: _open_manifest(*args),
-                [
-                    (
-                        self.io,
-                        manifest,
-                        partition_evaluators[manifest.partition_spec_id],
-                        self._build_metrics_evaluator(),
-                    )
-                    for manifest in manifests
-                    if self._check_sequence_number(min_sequence_number, manifest)
-                ],
-            )
+        return executor.map(
+            lambda args: _open_manifest(*args),
+            [
+                (
+                    self.io,
+                    manifest,
+                    partition_evaluators[manifest.partition_spec_id],
+                    self._build_metrics_evaluator(),
+                )
+                for manifest in manifests
+                if self._check_sequence_number(min_sequence_number, manifest)
+            ],
         )
 
     def plan_files(self) -> Iterable[FileScanTask]:
@@ -2000,7 +1998,7 @@ def plan_files(self) -> Iterable[FileScanTask]:
 
         residual_evaluators: Dict[int, Callable[[DataFile], ResidualEvaluator]] = KeyDefaultDict(self._build_residual_evaluator)
 
-        for manifest_entry in self.scan_plan_helper():
+        for manifest_entry in chain(*self.scan_plan_helper()):
             data_file = manifest_entry.data_file
             if data_file.content == DataFileContent.DATA:
                 data_entries.append(manifest_entry)

pyiceberg/table/inspect.py

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import itertools
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Optional, Set, Tuple, Union
 
@@ -309,7 +310,7 @@ def partitions(
 
         partitions_map: Dict[Tuple[str, Any], Any] = {}
 
-        for entry in scan.scan_plan_helper():
+        for entry in itertools.chain(*scan.scan_plan_helper()):
             partition = entry.data_file.partition
             partition_record_dict = {
                 field.name: partition[pos] for pos, field in enumerate(self.tbl.metadata.specs()[entry.data_file.spec_id].fields)