refactor: simplify lazy file planning

Author: manuzhang

src/iceberg/manifest/manifest_group.cc

@@ -134,7 +134,7 @@ ManifestGroup& ManifestGroup::operator=(ManifestGroup&&) noexcept = default;
 class ManifestGroup::FilePlanningIterator final
     : public Iterator<std::shared_ptr<FileScanTask>> {
  public:
-  static Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>> Make(
+  static Result<FileScanTaskIterator> Make(
       std::unique_ptr<ManifestGroup> group) {
     ICEBERG_RETURN_UNEXPECTED(group->CheckErrors());
 
@@ -153,7 +153,7 @@ class ManifestGroup::FilePlanningIterator final
                           group->case_sensitive_));
     }
 
-    return std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>(
+    return FileScanTaskIterator(
         new FilePlanningIterator(std::move(group), std::move(delete_index),
                                  std::move(data_file_evaluator), drop_stats));
   }
@@ -310,11 +310,14 @@ class ManifestGroup::FilePlanningIterator final
     ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
                             GetManifestEvaluator(manifest.partition_spec_id));
     ICEBERG_ASSIGN_OR_RAISE(bool should_match, evaluator->Evaluate(manifest));
-    if (!should_match ||
-        (group_->ignore_deleted_ && !manifest.has_added_files() &&
-         !manifest.has_existing_files()) ||
-        (group_->ignore_existing_ && !manifest.has_added_files() &&
-         !manifest.has_deleted_files())) {
+    const bool has_non_deleted_files =
+        manifest.has_added_files() || manifest.has_existing_files();
+    const bool has_non_existing_files =
+        manifest.has_added_files() || manifest.has_deleted_files();
+    const bool has_only_ignored_files =
+        (group_->ignore_deleted_ && !has_non_deleted_files) ||
+        (group_->ignore_existing_ && !has_non_existing_files);
+    if (!should_match || has_only_ignored_files) {
       IncrementSkippedDataManifests();
       return false;
     }
@@ -553,8 +556,7 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> ManifestGroup::PlanFiles() {
   return file_tasks;
 }
 
-Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
-ManifestGroup::PlanFilesIterator() && {
+Result<FileScanTaskIterator> ManifestGroup::PlanFilesIterator() && {
   auto group = std::make_unique<ManifestGroup>(std::move(*this));
   return FilePlanningIterator::Make(std::move(group));
 }

src/iceberg/manifest/manifest_group.h

@@ -145,7 +145,7 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
   /// manifests in each batch are opened in parallel, while entries are consumed one
   /// manifest at a time. Creating the iterator consumes this group's configuration, so
   /// this method may only be called on an rvalue.
-  Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>> PlanFilesIterator() &&;
+  Result<FileScanTaskIterator> PlanFilesIterator() &&;
 
   /// \brief Get all matching manifest entries.
   Result<std::vector<ManifestEntry>> Entries();

src/iceberg/table_scan.cc

@@ -110,7 +110,7 @@ Result<ScanReport> MakeScanReport(const DataTableScan& scan, const Snapshot& sna
 class ReportingFileTaskIterator final : public Iterator<std::shared_ptr<FileScanTask>> {
  public:
   ReportingFileTaskIterator(
-      std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>> iterator,
+      FileScanTaskIterator iterator,
       std::shared_ptr<ScanMetrics> scan_metrics,
       std::chrono::nanoseconds planning_duration,
       std::shared_ptr<MetricsReporter> reporter, ScanReport report)
@@ -147,7 +147,7 @@ class ReportingFileTaskIterator final : public Iterator<std::shared_ptr<FileScan
     std::ignore = reporter_->Report(report_);
   }
 
-  std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>> iterator_;
+  FileScanTaskIterator iterator_;
   std::shared_ptr<ScanMetrics> scan_metrics_;
   std::chrono::nanoseconds planning_duration_;
   std::shared_ptr<MetricsReporter> reporter_;
@@ -721,8 +721,7 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() co
   return tasks;
 }
 
-Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
-DataTableScan::PlanFilesIterator() const {
+Result<FileScanTaskIterator> DataTableScan::PlanFilesIterator() const {
   ICEBERG_ASSIGN_OR_RAISE(auto snapshot, this->snapshot());
   if (!snapshot) {
     return std::make_unique<EmptyIterator<std::shared_ptr<FileScanTask>>>();

src/iceberg/table_scan.h

@@ -468,8 +468,7 @@ class ICEBERG_EXPORT DataTableScan : public TableScan {
   ///
   /// Unlike PlanFiles(), this method does not materialize all manifest entries and scan
   /// tasks. The iterator owns its planning resources and can outlive this scan.
-  Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>> PlanFilesIterator()
-      const;
+  Result<FileScanTaskIterator> PlanFilesIterator() const;
 
  private:
   Status ReportScan(const Snapshot& snapshot, const ScanMetrics& scan_metrics) const;

src/iceberg/type_fwd.h

@@ -24,6 +24,8 @@
 /// you can include this instead of the "full" headers to help reduce compile
 /// times.
 
+#include <memory>
+
 namespace iceberg {
 
 /// \brief A data type.
@@ -232,6 +234,8 @@ struct SessionContext;
 class Executor;
 template <typename T>
 class Iterator;
+using FileScanTaskIterator =
+    std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>;
 
 /// \brief Metrics reporting.
 class MetricsReporter;