diff --git a/example/demo_example.cc b/example/demo_example.cc
index 22ecc0c90..6098a98ff 100644
--- a/example/demo_example.cc
+++ b/example/demo_example.cc
@@ -79,7 +79,7 @@ int main(int argc, char** argv) {
   }
 
   auto scan = std::move(scan_result.value());
-  auto plan_result = scan->PlanFiles();
+  auto plan_result = scan->PlanFilesStream();
   if (!plan_result.has_value()) {
     std::cerr << "Failed to plan files: " << plan_result.error().message << std::endl;
     return 1;
@@ -87,8 +87,18 @@ int main(int argc, char** argv) {
 
   std::cout << "Scan tasks: " << std::endl;
   auto scan_tasks = std::move(plan_result.value());
-  for (const auto& scan_task : scan_tasks) {
-    std::cout << " - " << scan_task->data_file()->file_path << std::endl;
+  while (true) {
+    auto task_result = scan_tasks->Next();
+    if (!task_result.has_value()) {
+      std::cerr << "Failed to plan next file: " << task_result.error().message
+                << std::endl;
+      return 1;
+    }
+    if (!task_result.value().has_value()) {
+      break;
+    }
+    std::cout << " - " << task_result.value().value()->data_file()->file_path
+              << std::endl;
   }
 
   return 0;
diff --git a/src/iceberg/manifest/manifest_group.cc b/src/iceberg/manifest/manifest_group.cc
index 02a51b113..fdfab70bd 100644
--- a/src/iceberg/manifest/manifest_group.cc
+++ b/src/iceberg/manifest/manifest_group.cc
@@ -41,7 +41,6 @@
 #include "iceberg/table_scan.h"
 #include "iceberg/type.h"
 #include "iceberg/util/cache_internal.h"
-#include "iceberg/util/checked_cast.h"
 #include "iceberg/util/content_file_util.h"
 #include "iceberg/util/executor_util_internal.h"
 #include "iceberg/util/macros.h"
@@ -131,6 +130,310 @@ ManifestGroup::~ManifestGroup() = default;
 ManifestGroup::ManifestGroup(ManifestGroup&&) noexcept = default;
 ManifestGroup& ManifestGroup::operator=(ManifestGroup&&) noexcept = default;
 
+class ManifestGroup::FilePlanningStream final : public FileScanTaskStream {
+ public:
+  static Result<FileScanTaskStreamPtr> Make(std::unique_ptr<ManifestGroup> group) {
+    ICEBERG_RETURN_UNEXPECTED(group->CheckErrors());
+
+    group->delete_index_builder_.WithScanMetrics(group->scan_metrics_);
+    ICEBERG_ASSIGN_OR_RAISE(auto delete_index, group->delete_index_builder_.Build());
+
+    auto stats_projection =
+        group->PrepareStatsProjection(delete_index->has_equality_deletes());
+
+    std::unique_ptr<Evaluator> data_file_evaluator;
+    if (group->file_filter_ &&
+        group->file_filter_->op() != Expression::Operation::kTrue) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          data_file_evaluator,
+          Evaluator::Make(*DataFileFilterSchema(), group->file_filter_,
+                          group->case_sensitive_));
+    }
+    const bool drop_stats = stats_projection.drop_stats;
+
+    return FileScanTaskStreamPtr(new FilePlanningStream(
+        std::move(group), std::move(delete_index), std::move(data_file_evaluator),
+        std::move(stats_projection.columns), drop_stats));
+  }
+
+  Result<std::optional<std::shared_ptr<FileScanTask>>> NextImpl() override {
+    while (true) {
+      ICEBERG_ASSIGN_OR_RAISE(auto entry, NextEntry());
+      if (!entry.has_value()) {
+        return std::nullopt;
+      }
+
+      auto [spec_id, value] = std::move(entry).value();
+      if (group_->ignore_existing_ && value.status == ManifestStatus::kExisting) {
+        IncrementSkippedDataFiles();
+        continue;
+      }
+
+      ICEBERG_DCHECK(value.data_file != nullptr, "Data file cannot be null");
+      if (data_file_evaluator_) {
+        DataFileStructLike data_file(*value.data_file);
+        ICEBERG_ASSIGN_OR_RAISE(bool should_match,
+                                data_file_evaluator_->Evaluate(data_file));
+        if (!should_match) {
+          IncrementSkippedDataFiles();
+          continue;
+        }
+      }
+
+      if (!group_->manifest_entry_predicate_(value)) {
+        IncrementSkippedDataFiles();
+        continue;
+      }
+
+      ICEBERG_ASSIGN_OR_RAISE(auto delete_files, delete_index_->ForEntry(value));
+
+      // Equality-delete matching uses data-file statistics. Drop unrequested stats only
+      // after the delete index has finished matching this entry.
+      if (drop_stats_) {
+        ContentFileUtil::DropAllStats(*value.data_file);
+      } else if (!group_->columns_to_keep_stats_.empty()) {
+        ContentFileUtil::DropUnselectedStats(*value.data_file,
+                                             group_->columns_to_keep_stats_);
+      }
+
+      UpdateResultMetrics(*value.data_file, delete_files);
+
+      ICEBERG_ASSIGN_OR_RAISE(auto residuals, GetResidualEvaluator(spec_id));
+      ICEBERG_ASSIGN_OR_RAISE(auto residual,
+                              residuals->ResidualFor(value.data_file->partition));
+
+      return std::optional<std::shared_ptr<FileScanTask>>{std::make_shared<FileScanTask>(
+          std::move(value.data_file), std::move(delete_files), std::move(residual))};
+    }
+  }
+
+ private:
+  FilePlanningStream(std::unique_ptr<ManifestGroup> group,
+                     std::unique_ptr<DeleteFileIndex> delete_index,
+                     std::unique_ptr<Evaluator> data_file_evaluator,
+                     std::vector<std::string> columns, bool drop_stats)
+      : group_(std::move(group)),
+        delete_index_(std::move(delete_index)),
+        data_file_evaluator_(std::move(data_file_evaluator)),
+        columns_(std::move(columns)),
+        drop_stats_(drop_stats) {}
+
+  using TaggedEntry = std::pair<int32_t, ManifestEntry>;
+  using TaggedStream = std::pair<int32_t, ManifestEntryStreamPtr>;
+
+  Result<std::optional<TaggedEntry>> NextEntry() {
+    if (!group_->executor_.has_value()) {
+      while (true) {
+        if (!entry_stream_) {
+          ICEBERG_ASSIGN_OR_RAISE(bool opened, OpenNextManifest());
+          if (!opened) {
+            return std::nullopt;
+          }
+        }
+
+        ICEBERG_ASSIGN_OR_RAISE(auto entry, entry_stream_->Next());
+        if (!entry.has_value()) {
+          entry_stream_.reset();
+          continue;
+        }
+        return std::optional<TaggedEntry>{std::in_place, current_spec_id_,
+                                          std::move(entry).value()};
+      }
+    }
+
+    while (true) {
+      if (next_batch_stream_ == batch_streams_.size()) {
+        ICEBERG_ASSIGN_OR_RAISE(bool loaded, LoadNextManifestBatch());
+        if (!loaded) {
+          return std::nullopt;
+        }
+      }
+
+      auto& [spec_id, stream] = batch_streams_[next_batch_stream_];
+      ICEBERG_ASSIGN_OR_RAISE(auto entry, stream->Next());
+      if (!entry.has_value()) {
+        stream.reset();
+        ++next_batch_stream_;
+        continue;
+      }
+      return std::optional<TaggedEntry>{std::in_place, spec_id, std::move(entry).value()};
+    }
+  }
+
+  Result<ManifestEvaluator*> GetManifestEvaluator(int32_t spec_id) {
+    auto cached = manifest_evaluators_.find(spec_id);
+    if (cached != manifest_evaluators_.end()) {
+      return cached->second.get();
+    }
+
+    auto spec_iter = group_->specs_by_id_.find(spec_id);
+    ICEBERG_CHECK(spec_iter != group_->specs_by_id_.cend(),
+                  "Cannot find partition spec for ID {}", spec_id);
+
+    const auto& spec = spec_iter->second;
+    auto projector =
+        Projections::Inclusive(*spec, *group_->schema_, group_->case_sensitive_);
+    ICEBERG_ASSIGN_OR_RAISE(auto partition_filter,
+                            projector->Project(group_->data_filter_));
+    ICEBERG_ASSIGN_OR_RAISE(partition_filter, And::Make(std::move(partition_filter),
+                                                        group_->partition_filter_));
+    ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
+                            ManifestEvaluator::MakePartitionFilter(
+                                std::move(partition_filter), spec, *group_->schema_,
+                                group_->case_sensitive_));
+    auto* result = evaluator.get();
+    manifest_evaluators_.emplace(spec_id, std::move(evaluator));
+    return result;
+  }
+
+  Result<ResidualEvaluator*> GetResidualEvaluator(int32_t spec_id) {
+    auto cached = residual_evaluators_.find(spec_id);
+    if (cached != residual_evaluators_.end()) {
+      return cached->second.get();
+    }
+
+    auto spec_iter = group_->specs_by_id_.find(spec_id);
+    ICEBERG_CHECK(spec_iter != group_->specs_by_id_.cend(),
+                  "Cannot find partition spec for ID {}", spec_id);
+
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto evaluator,
+        ResidualEvaluator::Make(
+            (group_->ignore_residuals_ ? True::Instance() : group_->data_filter_),
+            *spec_iter->second, *group_->schema_, group_->case_sensitive_));
+    auto* result = evaluator.get();
+    residual_evaluators_.emplace(spec_id, std::move(evaluator));
+    return result;
+  }
+
+  Result<bool> ShouldReadManifest(const ManifestFile& manifest) {
+    ICEBERG_ASSIGN_OR_RAISE(auto evaluator,
+                            GetManifestEvaluator(manifest.partition_spec_id));
+    ICEBERG_ASSIGN_OR_RAISE(bool should_match, evaluator->Evaluate(manifest));
+    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;
+    }
+
+    if (group_->scan_metrics_) {
+      group_->scan_metrics_->scanned_data_manifests->Increment(1);
+    }
+    return true;
+  }
+
+  Result<bool> OpenNextManifest() {
+    while (next_manifest_ < group_->data_manifests_.size()) {
+      const auto& manifest = group_->data_manifests_[next_manifest_++];
+      ICEBERG_ASSIGN_OR_RAISE(bool should_read, ShouldReadManifest(manifest));
+      if (!should_read) {
+        continue;
+      }
+
+      ICEBERG_ASSIGN_OR_RAISE(auto reader, group_->MakeReader(manifest, columns_));
+      ICEBERG_ASSIGN_OR_RAISE(entry_stream_, group_->ignore_deleted_
+                                                 ? reader->LiveEntriesStream()
+                                                 : reader->EntriesStream());
+      current_spec_id_ = manifest.partition_spec_id;
+      return true;
+    }
+    return false;
+  }
+
+  Result<bool> LoadNextManifestBatch() {
+    std::vector<const ManifestFile*> manifests;
+    manifests.reserve(kManifestReadBatchSize);
+    while (next_manifest_ < group_->data_manifests_.size() &&
+           manifests.size() < kManifestReadBatchSize) {
+      const auto& manifest = group_->data_manifests_[next_manifest_++];
+      ICEBERG_ASSIGN_OR_RAISE(bool should_read, ShouldReadManifest(manifest));
+      if (should_read) {
+        manifests.push_back(&manifest);
+      }
+    }
+
+    if (manifests.empty()) {
+      return false;
+    }
+
+    // Open the readers concurrently, but keep their streams instead of collecting
+    // entries here. This preserves bounded memory for large manifests while retaining
+    // parallel manifest initialization when an executor is configured.
+    ICEBERG_ASSIGN_OR_RAISE(
+        batch_streams_,
+        ParallelCollect(
+            group_->executor_, manifests,
+            [this](const ManifestFile* manifest) -> Result<std::vector<TaggedStream>> {
+              ICEBERG_ASSIGN_OR_RAISE(auto reader,
+                                      group_->MakeReader(*manifest, columns_));
+              ICEBERG_ASSIGN_OR_RAISE(auto stream, group_->ignore_deleted_
+                                                       ? reader->LiveEntriesStream()
+                                                       : reader->EntriesStream());
+
+              std::vector<TaggedStream> tagged_streams;
+              tagged_streams.emplace_back(manifest->partition_spec_id, std::move(stream));
+              return tagged_streams;
+            }));
+    next_batch_stream_ = 0;
+    return true;
+  }
+
+  void IncrementSkippedDataManifests() {
+    if (group_->scan_metrics_) {
+      group_->scan_metrics_->skipped_data_manifests->Increment(1);
+    }
+  }
+
+  void IncrementSkippedDataFiles() {
+    if (group_->scan_metrics_) {
+      group_->scan_metrics_->skipped_data_files->Increment(1);
+    }
+  }
+
+  void UpdateResultMetrics(const DataFile& data_file,
+                           const std::vector<std::shared_ptr<DataFile>>& delete_files) {
+    if (!group_->scan_metrics_) {
+      return;
+    }
+
+    group_->scan_metrics_->total_file_size_in_bytes->Increment(
+        ContentFileUtil::ContentSizeInBytes(data_file));
+    group_->scan_metrics_->result_data_files->Increment(1);
+    group_->scan_metrics_->result_delete_files->Increment(
+        static_cast<int64_t>(delete_files.size()));
+    int64_t deletes_size = 0;
+    for (const auto& delete_file : delete_files) {
+      deletes_size += ContentFileUtil::ContentSizeInBytes(*delete_file);
+    }
+    group_->scan_metrics_->total_delete_file_size_in_bytes->Increment(deletes_size);
+  }
+
+  std::unique_ptr<ManifestGroup> group_;
+  std::unique_ptr<DeleteFileIndex> delete_index_;
+  std::unique_ptr<Evaluator> data_file_evaluator_;
+  std::vector<std::string> columns_;
+  std::unordered_map<int32_t, std::unique_ptr<ManifestEvaluator>> manifest_evaluators_;
+  std::unordered_map<int32_t, std::shared_ptr<ResidualEvaluator>> residual_evaluators_;
+  ManifestEntryStreamPtr entry_stream_;
+  std::vector<TaggedStream> batch_streams_;
+  size_t next_manifest_ = 0;
+  size_t next_batch_stream_ = 0;
+  int32_t current_spec_id_ = 0;
+  bool drop_stats_;
+
+  // Limit the number of manifest readers and streams retained by executor-backed
+  // planning. The executor still controls actual task concurrency, while this fixed
+  // cap prevents resource use from scaling with the total manifest count. Entries
+  // within each manifest remain streamed, so this does not cap manifest size.
+  static constexpr size_t kManifestReadBatchSize = 32;
+};
+
 ManifestGroup& ManifestGroup::FilterData(std::shared_ptr<Expression> filter) {
   ICEBERG_BUILDER_ASSIGN_OR_RETURN(data_filter_, And::Make(data_filter_, filter));
   delete_index_builder_.DataFilter(std::move(filter));
@@ -203,52 +506,14 @@ ManifestGroup& ManifestGroup::WithScanMetrics(std::shared_ptr<ScanMetrics> scan_
   return *this;
 }
 
-Result<std::vector<std::shared_ptr<FileScanTask>>> ManifestGroup::PlanFiles() {
-  auto create_file_scan_tasks =
-      [this](std::vector<ManifestEntry>&& entries,
-             const TaskContext& ctx) -> Result<std::vector<std::shared_ptr<ScanTask>>> {
-    std::vector<std::shared_ptr<ScanTask>> tasks;
-    tasks.reserve(entries.size());
-
-    for (auto& entry : entries) {
-      if (ctx.drop_stats) {
-        ContentFileUtil::DropAllStats(*entry.data_file);
-      } else if (!ctx.columns_to_keep_stats.empty()) {
-        ContentFileUtil::DropUnselectedStats(*entry.data_file, ctx.columns_to_keep_stats);
-      }
-      ICEBERG_ASSIGN_OR_RAISE(auto delete_files, ctx.deletes->ForEntry(entry));
-      // Count result metrics once per data file task. A delete file shared by
-      // multiple data files contributes once to each task, unlike indexed delete files.
-      if (scan_metrics_) {
-        scan_metrics_->total_file_size_in_bytes->Increment(
-            ContentFileUtil::ContentSizeInBytes(*entry.data_file));
-        scan_metrics_->result_data_files->Increment(1);
-        scan_metrics_->result_delete_files->Increment(
-            static_cast<int64_t>(delete_files.size()));
-        int64_t deletes_size = 0;
-        for (const auto& delete_file : delete_files) {
-          deletes_size += ContentFileUtil::ContentSizeInBytes(*delete_file);
-        }
-        scan_metrics_->total_delete_file_size_in_bytes->Increment(deletes_size);
-      }
-      ICEBERG_ASSIGN_OR_RAISE(auto residual,
-                              ctx.residuals->ResidualFor(entry.data_file->partition));
-      tasks.push_back(std::make_shared<FileScanTask>(
-          std::move(entry.data_file), std::move(delete_files), std::move(residual)));
-    }
-
-    return tasks;
-  };
-
-  ICEBERG_ASSIGN_OR_RAISE(auto tasks, Plan(create_file_scan_tasks));
+Result<std::vector<std::shared_ptr<FileScanTask>>> ManifestGroup::PlanFiles() && {
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, std::move(*this).PlanFilesStream());
+  return stream->ToVector();
+}
 
-  // Convert ScanTask to FileScanTask
-  std::vector<std::shared_ptr<FileScanTask>> file_tasks;
-  file_tasks.reserve(tasks.size());
-  for (auto& task : tasks) {
-    file_tasks.push_back(internal::checked_pointer_cast<FileScanTask>(task));
-  }
-  return file_tasks;
+Result<FileScanTaskStreamPtr> ManifestGroup::PlanFilesStream() && {
+  auto group = std::make_unique<ManifestGroup>(std::move(*this));
+  return FilePlanningStream::Make(std::move(group));
 }
 
 Result<std::vector<std::shared_ptr<ScanTask>>> ManifestGroup::Plan(
@@ -276,10 +541,7 @@ Result<std::vector<std::shared_ptr<ScanTask>>> ManifestGroup::Plan(
   delete_index_builder_.WithScanMetrics(scan_metrics_);
   ICEBERG_ASSIGN_OR_RAISE(auto delete_index, delete_index_builder_.Build());
 
-  bool drop_stats = ManifestReader::ShouldDropStats(columns_);
-  if (delete_index->has_equality_deletes()) {
-    columns_ = ManifestReader::WithStatsColumns(columns_);
-  }
+  auto stats_projection = PrepareStatsProjection(delete_index->has_equality_deletes());
 
   std::unordered_map<int32_t, std::unique_ptr<TaskContext>> task_context_cache;
   auto get_task_context = [&](int32_t spec_id) -> Result<TaskContext*> {
@@ -297,13 +559,13 @@ Result<std::vector<std::shared_ptr<ScanTask>>> ManifestGroup::Plan(
         TaskContext{.spec = spec,
                     .deletes = delete_index.get(),
                     .residuals = residuals,
-                    .drop_stats = drop_stats,
+                    .drop_stats = stats_projection.drop_stats,
                     .columns_to_keep_stats = columns_to_keep_stats_});
 
     return task_context_cache[spec_id].get();
   };
 
-  ICEBERG_ASSIGN_OR_RAISE(auto entry_groups, ReadEntries());
+  ICEBERG_ASSIGN_OR_RAISE(auto entry_groups, ReadEntries(stats_projection.columns));
 
   std::vector<std::shared_ptr<ScanTask>> all_tasks;
   for (auto& [spec_id, entries] : entry_groups) {
@@ -317,7 +579,7 @@ Result<std::vector<std::shared_ptr<ScanTask>>> ManifestGroup::Plan(
 }
 
 Result<std::vector<ManifestEntry>> ManifestGroup::Entries() {
-  ICEBERG_ASSIGN_OR_RAISE(auto entry_groups, ReadEntries());
+  ICEBERG_ASSIGN_OR_RAISE(auto entry_groups, ReadEntries(columns_));
 
   std::vector<ManifestEntry> all_entries;
   for (auto& [_, entries] : entry_groups) {
@@ -329,13 +591,14 @@ Result<std::vector<ManifestEntry>> ManifestGroup::Entries() {
 }
 
 Result<std::unique_ptr<ManifestReader>> ManifestGroup::MakeReader(
-    const ManifestFile& manifest) {
+    const ManifestFile& manifest, const std::vector<std::string>& columns) {
   ICEBERG_ASSIGN_OR_RAISE(auto reader,
                           ManifestReader::Make(manifest, io_, schema_, specs_by_id_));
 
-  auto columns = columns_;
+  auto reader_columns = columns;
   if (file_filter_ && file_filter_->op() != Expression::Operation::kTrue &&
-      !columns.empty() && !std::ranges::contains(columns, Schema::kAllColumns)) {
+      !reader_columns.empty() &&
+      !std::ranges::contains(reader_columns, Schema::kAllColumns)) {
     auto data_file_schema = DataFileFilterSchema();
     ICEBERG_ASSIGN_OR_RAISE(
         auto bound_file_filter,
@@ -343,7 +606,8 @@ Result<std::unique_ptr<ManifestReader>> ManifestGroup::MakeReader(
     ICEBERG_ASSIGN_OR_RAISE(auto referenced_field_ids,
                             ReferenceVisitor::GetReferencedFieldIds(bound_file_filter));
 
-    std::unordered_set<std::string> selected_columns(columns.cbegin(), columns.cend());
+    std::unordered_set<std::string> selected_columns(reader_columns.cbegin(),
+                                                     reader_columns.cend());
     for (const auto field_id : referenced_field_ids) {
       if (field_id == DataFile::kSpecIdFieldId) {
         continue;
@@ -355,8 +619,8 @@ Result<std::unique_ptr<ManifestReader>> ManifestGroup::MakeReader(
         if (selected_columns.contains(column_name_str)) {
           continue;
         }
-        columns.push_back(std::move(column_name_str));
-        selected_columns.insert(columns.back());
+        reader_columns.push_back(std::move(column_name_str));
+        selected_columns.insert(reader_columns.back());
       }
     }
   }
@@ -364,7 +628,7 @@ Result<std::unique_ptr<ManifestReader>> ManifestGroup::MakeReader(
   reader->FilterRows(data_filter_)
       .FilterPartitions(partition_filter_)
       .CaseSensitive(case_sensitive_)
-      .Select(std::move(columns));
+      .Select(std::move(reader_columns));
 
   if (scan_metrics_) {
     reader->SkipCounter(scan_metrics_->skipped_data_files);
@@ -373,8 +637,29 @@ Result<std::unique_ptr<ManifestReader>> ManifestGroup::MakeReader(
   return reader;
 }
 
+ManifestGroup::StatsProjection ManifestGroup::PrepareStatsProjection(
+    bool has_equality_deletes) const {
+  // The caller's projection records whether stats were requested. Equality-delete
+  // matching may add stats temporarily, but they should still be dropped from the
+  // result when the original projection did not request them. Keeping this decision
+  // here ensures eager and stream planning use identical semantics.
+  StatsProjection result{.columns = columns_,
+                         .drop_stats = ManifestReader::ShouldDropStats(columns_)};
+  // Delete matching and residual evaluation require partition values even when the
+  // caller does not select them. Do not narrow an empty or select-all projection.
+  if (!result.columns.empty() &&
+      !std::ranges::contains(result.columns, Schema::kAllColumns) &&
+      !std::ranges::contains(result.columns, DataFile::kPartitionField)) {
+    result.columns.emplace_back(DataFile::kPartitionField);
+  }
+  if (has_equality_deletes) {
+    result.columns = ManifestReader::WithStatsColumns(result.columns);
+  }
+  return result;
+}
+
 Result<std::unordered_map<int32_t, std::vector<ManifestEntry>>>
-ManifestGroup::ReadEntries() {
+ManifestGroup::ReadEntries(const std::vector<std::string>& columns) {
   const auto cache_capacity = static_cast<int32_t>(specs_by_id_.size());
   auto get_manifest_evaluator = internal::MemoizeLru(
       [this](int32_t spec_id) -> Result<std::shared_ptr<ManifestEvaluator>> {
@@ -441,7 +726,7 @@ ManifestGroup::ReadEntries() {
         }
 
         // Read manifest entries
-        ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeReader(manifest));
+        ICEBERG_ASSIGN_OR_RAISE(auto reader, MakeReader(manifest, columns));
         ICEBERG_ASSIGN_OR_RAISE(
             auto entries, ignore_deleted_ ? reader->LiveEntries() : reader->Entries());
 
diff --git a/src/iceberg/manifest/manifest_group.h b/src/iceberg/manifest/manifest_group.h
index be0ca4b8b..a53dc590b 100644
--- a/src/iceberg/manifest/manifest_group.h
+++ b/src/iceberg/manifest/manifest_group.h
@@ -34,6 +34,7 @@
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_list.h"
 #include "iceberg/result.h"
+#include "iceberg/table_scan.h"
 #include "iceberg/type_fwd.h"
 #include "iceberg/util/error_collector.h"
 #include "iceberg/util/executor.h"
@@ -113,6 +114,9 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
 
   /// \brief Select specific columns from manifest entries.
   ///
+  /// Task planning also reads partition values for delete matching and residuals, and
+  /// may temporarily read statistics needed for equality-delete matching.
+  ///
   /// \param columns Column names to select from manifest entries.
   ManifestGroup& Select(std::vector<std::string> columns);
 
@@ -126,6 +130,9 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
 
   /// \brief Configure an optional executor for manifest planning.
   ///
+  /// The executor is borrowed and must remain alive throughout planning and until any
+  /// stream returned by PlanFilesStream() is destroyed.
+  ///
   /// \param executor Executor to use, or std::nullopt to plan manifests serially.
   /// \return Reference to this for method chaining.
   ManifestGroup& PlanWith(OptionalExecutor executor);
@@ -134,7 +141,29 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
   ManifestGroup& WithScanMetrics(std::shared_ptr<ScanMetrics> scan_metrics);
 
   /// \brief Plan scan tasks for all matching data files.
-  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles();
+  ///
+  /// Consumes this group and collects PlanFilesStream() into a vector.
+  /// Callers must use std::move(group).PlanFiles(), or std::move(*group).PlanFiles()
+  /// for an owning pointer, instead of calling on an lvalue. Do not reuse the consumed
+  /// group; construct a new group to plan again.
+  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() &&;
+
+  /// \brief Lazily plan scan tasks for matching data files.
+  ///
+  /// The returned stream owns the planning state and may outlive this ManifestGroup.
+  /// An executor configured through PlanWith() is borrowed and must remain alive until
+  /// the stream is destroyed, as later Next() calls may submit work to it.
+  ///
+  /// It reads one bounded manifest batch at a time instead of materializing all manifest
+  /// entries and scan tasks. When PlanWith() configures an executor, entry streams for
+  /// manifests in each batch are opened in parallel, while entries are consumed one
+  /// manifest at a time. Delete manifests are still read eagerly when creating the
+  /// stream because delete files must be indexed before data-file planning can begin.
+  /// Creating the stream consumes this group's configuration, so this method may only
+  /// be called on an rvalue. Use std::move(group).PlanFilesStream(), or
+  /// std::move(*group).PlanFilesStream() for an owning pointer, and do not reuse the
+  /// group.
+  Result<FileScanTaskStreamPtr> PlanFilesStream() &&;
 
   /// \brief Get all matching manifest entries.
   Result<std::vector<ManifestEntry>> Entries();
@@ -151,14 +180,25 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
       const CreateTasksFunction& create_tasks);
 
  private:
+  class FilePlanningStream;
+
+  struct StatsProjection {
+    std::vector<std::string> columns;
+    bool drop_stats;
+  };
+
   ManifestGroup(std::shared_ptr<FileIO> io, std::shared_ptr<Schema> schema,
                 std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> specs_by_id,
                 std::vector<ManifestFile> data_manifests,
                 DeleteFileIndex::Builder&& delete_index_builder);
 
-  Result<std::unordered_map<int32_t, std::vector<ManifestEntry>>> ReadEntries();
+  Result<std::unordered_map<int32_t, std::vector<ManifestEntry>>> ReadEntries(
+      const std::vector<std::string>& columns);
+
+  Result<std::unique_ptr<ManifestReader>> MakeReader(
+      const ManifestFile& manifest, const std::vector<std::string>& columns);
 
-  Result<std::unique_ptr<ManifestReader>> MakeReader(const ManifestFile& manifest);
+  StatsProjection PrepareStatsProjection(bool has_equality_deletes) const;
 
   std::shared_ptr<FileIO> io_;
   std::shared_ptr<Schema> schema_;
diff --git a/src/iceberg/manifest/manifest_reader.cc b/src/iceberg/manifest/manifest_reader.cc
index 8da5befcb..1e196df72 100644
--- a/src/iceberg/manifest/manifest_reader.cc
+++ b/src/iceberg/manifest/manifest_reader.cc
@@ -25,6 +25,7 @@
 #include <ranges>
 #include <type_traits>
 #include <unordered_set>
+#include <utility>
 
 #include <nanoarrow/nanoarrow.h>
 
@@ -689,6 +690,112 @@ Result<std::shared_ptr<Schema>> ProjectSchema(std::shared_ptr<Schema> schema,
   return schema;
 }
 
+class ManifestEntryStreamImpl final : public ManifestEntryStream {
+ public:
+  ManifestEntryStreamImpl(std::unique_ptr<Reader> reader,
+                          std::shared_ptr<Schema> file_schema, ArrowSchema arrow_schema,
+                          std::shared_ptr<InheritableMetadata> inheritable_metadata,
+                          std::optional<int64_t> first_row_id, bool is_committed,
+                          bool only_live, std::unique_ptr<Evaluator> evaluator,
+                          std::unique_ptr<InclusiveMetricsEvaluator> metrics_evaluator,
+                          std::shared_ptr<PartitionSet> partition_set,
+                          std::shared_ptr<Counter> skip_counter, bool drop_stats)
+      : reader_(std::move(reader)),
+        file_schema_(std::move(file_schema)),
+        arrow_schema_(std::exchange(arrow_schema, ArrowSchema{})),
+        arrow_schema_guard_(&arrow_schema_),
+        inheritable_metadata_(std::move(inheritable_metadata)),
+        first_row_id_(first_row_id),
+        is_committed_(is_committed),
+        only_live_(only_live),
+        evaluator_(std::move(evaluator)),
+        metrics_evaluator_(std::move(metrics_evaluator)),
+        partition_set_(std::move(partition_set)),
+        skip_counter_(std::move(skip_counter)),
+        drop_stats_(drop_stats) {}
+
+  Result<std::optional<ManifestEntry>> NextImpl() override {
+    while (true) {
+      while (next_entry_ < entries_.size()) {
+        auto entry = std::move(entries_[next_entry_++]);
+        ICEBERG_RETURN_UNEXPECTED(inheritable_metadata_->Apply(entry));
+
+        if (only_live_ && !entry.IsAlive()) {
+          continue;
+        }
+
+        ICEBERG_DCHECK(entry.data_file != nullptr, "Data file cannot be null");
+        if (evaluator_) {
+          ICEBERG_ASSIGN_OR_RAISE(bool partition_match,
+                                  evaluator_->Evaluate(entry.data_file->partition));
+          if (!partition_match) {
+            IncrementSkipCounter();
+            continue;
+          }
+        }
+        if (metrics_evaluator_) {
+          ICEBERG_ASSIGN_OR_RAISE(bool metrics_match,
+                                  metrics_evaluator_->Evaluate(*entry.data_file));
+          if (!metrics_match) {
+            IncrementSkipCounter();
+            continue;
+          }
+        }
+        if (partition_set_) {
+          ICEBERG_PRECHECK(entry.data_file->partition_spec_id.has_value(),
+                           "Missing partition spec id from data file {}",
+                           entry.data_file->file_path);
+          if (!partition_set_->contains(entry.data_file->partition_spec_id.value(),
+                                        entry.data_file->partition)) {
+            IncrementSkipCounter();
+            continue;
+          }
+        }
+
+        if (drop_stats_) {
+          ContentFileUtil::DropAllStats(*entry.data_file);
+        }
+        return std::optional<ManifestEntry>{std::move(entry)};
+      }
+
+      entries_.clear();
+      next_entry_ = 0;
+      ICEBERG_ASSIGN_OR_RAISE(auto batch, reader_->Next());
+      if (!batch.has_value()) {
+        return std::nullopt;
+      }
+
+      internal::ArrowArrayGuard array_guard(&batch.value());
+      ICEBERG_ASSIGN_OR_RAISE(
+          entries_, ParseManifestEntry(&arrow_schema_, &batch.value(), *file_schema_,
+                                       first_row_id_, is_committed_));
+    }
+  }
+
+ private:
+  void IncrementSkipCounter() {
+    if (skip_counter_) {
+      skip_counter_->Increment(1);
+    }
+  }
+
+  std::unique_ptr<Reader> reader_;
+  std::shared_ptr<Schema> file_schema_;
+  ArrowSchema arrow_schema_{};
+  internal::ArrowSchemaGuard arrow_schema_guard_;
+  std::shared_ptr<InheritableMetadata> inheritable_metadata_;
+  std::optional<int64_t> first_row_id_;
+  bool is_committed_;
+  bool only_live_;
+  std::unique_ptr<Evaluator> evaluator_;
+  std::unique_ptr<InclusiveMetricsEvaluator> metrics_evaluator_;
+  std::shared_ptr<PartitionSet> partition_set_;
+  std::shared_ptr<Counter> skip_counter_;
+  bool drop_stats_;
+  std::vector<ManifestEntry> entries_;
+  size_t next_entry_ = 0;
+};
+
 }  // namespace
 
 bool ManifestReader::ShouldDropStats(const std::vector<std::string>& columns) {
@@ -716,7 +823,7 @@ bool ManifestReader::ShouldDropStats(const std::vector<std::string>& columns) {
 
 std::vector<std::string> ManifestReader::WithStatsColumns(
     const std::vector<std::string>& columns) {
-  if (std::ranges::contains(columns, Schema::kAllColumns)) {
+  if (columns.empty() || std::ranges::contains(columns, Schema::kAllColumns)) {
     return columns;
   } else {
     std::vector<std::string> updated_columns{columns};
@@ -788,7 +895,7 @@ bool ManifestReaderImpl::HasRowFilter() const {
   return row_filter_->op() != Expression::Operation::kTrue;
 }
 
-Result<Evaluator*> ManifestReaderImpl::GetEvaluator() {
+Result<std::unique_ptr<Evaluator>> ManifestReaderImpl::TakeEvaluator() {
   if (!evaluator_) {
     auto projection_evaluator = Projections::Inclusive(*spec_, *schema_, case_sensitive_);
     ICEBERG_ASSIGN_OR_RAISE(auto projected, projection_evaluator->Project(row_filter_));
@@ -801,25 +908,17 @@ Result<Evaluator*> ManifestReaderImpl::GetEvaluator() {
         evaluator_, Evaluator::Make(*partition_schema, std::move(final_part_filter),
                                     case_sensitive_));
   }
-  return evaluator_.get();
+  return std::move(evaluator_);
 }
 
-Result<InclusiveMetricsEvaluator*> ManifestReaderImpl::GetMetricsEvaluator() {
+Result<std::unique_ptr<InclusiveMetricsEvaluator>>
+ManifestReaderImpl::TakeMetricsEvaluator() {
   if (!metrics_evaluator_) {
     ICEBERG_ASSIGN_OR_RAISE(
         metrics_evaluator_,
         InclusiveMetricsEvaluator::Make(row_filter_, *schema_, case_sensitive_));
   }
-  return metrics_evaluator_.get();
-}
-
-Result<bool> ManifestReaderImpl::InPartitionSet(const DataFile& file) const {
-  if (!partition_set_) {
-    return true;
-  }
-  ICEBERG_PRECHECK(file.partition_spec_id.has_value(),
-                   "Missing partition spec id from data file {}", file.file_path);
-  return partition_set_->contains(file.partition_spec_id.value(), file.partition);
+  return std::move(metrics_evaluator_);
 }
 
 Status ManifestReaderImpl::OpenReader(std::shared_ptr<Schema> projection) {
@@ -860,15 +959,25 @@ Status ManifestReaderImpl::OpenReader(std::shared_ptr<Schema> projection) {
   return {};
 }
 
-Result<std::vector<ManifestEntry>> ManifestReaderImpl::Entries() {
-  return ReadEntries(/*only_live=*/false);
+Result<std::vector<ManifestEntry>> ManifestReader::Entries() {
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, EntriesStream());
+  return entries->ToVector();
+}
+
+Result<std::vector<ManifestEntry>> ManifestReader::LiveEntries() {
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, LiveEntriesStream());
+  return entries->ToVector();
 }
 
-Result<std::vector<ManifestEntry>> ManifestReaderImpl::LiveEntries() {
-  return ReadEntries(/*only_live=*/true);
+Result<ManifestEntryStreamPtr> ManifestReaderImpl::EntriesStream() {
+  return MakeEntriesStream(/*only_live=*/false);
 }
 
-Result<std::vector<ManifestEntry>> ManifestReaderImpl::ReadEntries(bool only_live) {
+Result<ManifestEntryStreamPtr> ManifestReaderImpl::LiveEntriesStream() {
+  return MakeEntriesStream(/*only_live=*/true);
+}
+
+Result<ManifestEntryStreamPtr> ManifestReaderImpl::MakeEntriesStream(bool only_live) {
   ICEBERG_ASSIGN_OR_RAISE(auto partition_type, spec_->RawPartitionType(*schema_));
   auto data_file_schema = DataFile::Type(std::move(partition_type))->ToSchema();
 
@@ -894,74 +1003,27 @@ Result<std::vector<ManifestEntry>> ManifestReaderImpl::ReadEntries(bool only_liv
   ICEBERG_RETURN_UNEXPECTED(OpenReader(std::move(projected_data_file_schema)));
   ICEBERG_DCHECK(file_reader_ != nullptr, "File reader should be initialized");
 
-  std::vector<ManifestEntry> manifest_entries;
   ICEBERG_ASSIGN_OR_RAISE(auto arrow_schema, file_reader_->Schema());
   internal::ArrowSchemaGuard schema_guard(&arrow_schema);
 
   // Get evaluators if needed
-  Evaluator* evaluator = nullptr;
-  InclusiveMetricsEvaluator* metrics_evaluator = nullptr;
+  std::unique_ptr<Evaluator> evaluator;
+  std::unique_ptr<InclusiveMetricsEvaluator> metrics_evaluator;
   if (HasPartitionFilter() || HasRowFilter()) {
-    ICEBERG_ASSIGN_OR_RAISE(evaluator, GetEvaluator());
+    ICEBERG_ASSIGN_OR_RAISE(evaluator, TakeEvaluator());
   }
   if (HasRowFilter()) {
-    ICEBERG_ASSIGN_OR_RAISE(metrics_evaluator, GetMetricsEvaluator());
+    ICEBERG_ASSIGN_OR_RAISE(metrics_evaluator, TakeMetricsEvaluator());
   }
 
   bool drop_stats = drop_stats_ && ShouldDropStats(columns_);
-
-  while (true) {
-    ICEBERG_ASSIGN_OR_RAISE(auto result, file_reader_->Next());
-    if (!result.has_value()) {
-      break;  // EOF
-    }
-
-    internal::ArrowArrayGuard array_guard(&result.value());
-    ICEBERG_ASSIGN_OR_RAISE(
-        auto entries, ParseManifestEntry(&arrow_schema, &result.value(), *file_schema_,
-                                         first_row_id_, is_committed_));
-
-    for (auto& entry : entries) {
-      ICEBERG_RETURN_UNEXPECTED(inheritable_metadata_->Apply(entry));
-
-      if (only_live && !entry.IsAlive()) {
-        continue;
-      }
-
-      if (needs_filtering) {
-        ICEBERG_DCHECK(entry.data_file != nullptr, "Data file cannot be null");
-        if (evaluator) {
-          ICEBERG_ASSIGN_OR_RAISE(bool partition_match,
-                                  evaluator->Evaluate(entry.data_file->partition));
-          if (!partition_match) {
-            if (skip_counter_) skip_counter_->Increment(1);
-            continue;
-          }
-        }
-        if (metrics_evaluator) {
-          ICEBERG_ASSIGN_OR_RAISE(bool metrics_match,
-                                  metrics_evaluator->Evaluate(*entry.data_file));
-          if (!metrics_match) {
-            if (skip_counter_) skip_counter_->Increment(1);
-            continue;
-          }
-        }
-        ICEBERG_ASSIGN_OR_RAISE(bool in_partition_set, InPartitionSet(*entry.data_file));
-        if (!in_partition_set) {
-          if (skip_counter_) skip_counter_->Increment(1);
-          continue;
-        }
-      }
-
-      if (drop_stats) {
-        ContentFileUtil::DropAllStats(*entry.data_file);
-      }
-
-      manifest_entries.push_back(std::move(entry));
-    }
-  }
-
-  return manifest_entries;
+  auto stream = ManifestEntryStreamPtr(new ManifestEntryStreamImpl(
+      std::move(file_reader_), file_schema_, std::move(arrow_schema),
+      inheritable_metadata_, first_row_id_, is_committed_, only_live,
+      std::move(evaluator), std::move(metrics_evaluator), partition_set_, skip_counter_,
+      drop_stats));
+  schema_guard.Release();
+  return stream;
 }
 
 Result<std::vector<ManifestFile>> ManifestListReaderImpl::Files() const {
diff --git a/src/iceberg/manifest/manifest_reader.h b/src/iceberg/manifest/manifest_reader.h
index 72cb9ae56..073bba9c4 100644
--- a/src/iceberg/manifest/manifest_reader.h
+++ b/src/iceberg/manifest/manifest_reader.h
@@ -30,24 +30,52 @@
 #include <vector>
 
 #include "iceberg/iceberg_export.h"
+#include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/metrics/counter.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
+#include "iceberg/util/stream.h"
 
 namespace iceberg {
 
+/// \brief Stream of manifest entries.
+using ManifestEntryStream = Stream<ManifestEntry>;
+
+/// \brief Owning pointer to a manifest entry stream.
+using ManifestEntryStreamPtr = std::unique_ptr<ManifestEntryStream>;
+
 /// \brief Read manifest entries from a manifest file.
+///
+/// Implementations must override EntriesStream() and LiveEntriesStream(), returning
+/// self-contained streams that may outlive the reader. These are the extension points;
+/// Entries() and LiveEntries() are non-virtual eager collection helpers. Custom readers
+/// that previously overrode the eager methods must migrate to the stream methods;
+/// no eager-to-stream fallback is provided.
 class ICEBERG_EXPORT ManifestReader {
  public:
   virtual ~ManifestReader() = default;
 
   /// \brief Read all manifest entries in the manifest file.
   ///
-  /// TODO(gangwu): provide a lazy-evaluated iterator interface for better performance.
-  virtual Result<std::vector<ManifestEntry>> Entries() = 0;
+  /// Collects EntriesStream() into a vector.
+  Result<std::vector<ManifestEntry>> Entries();
 
   /// \brief Read only live (non-deleted) manifest entries.
-  virtual Result<std::vector<ManifestEntry>> LiveEntries() = 0;
+  ///
+  /// Collects LiveEntriesStream() into a vector.
+  Result<std::vector<ManifestEntry>> LiveEntries();
+
+  /// \brief Lazily read manifest entries.
+  ///
+  /// The returned stream is fallible and single-pass. It must own all resources
+  /// required for consumption and must not depend on this reader remaining alive.
+  virtual Result<ManifestEntryStreamPtr> EntriesStream() = 0;
+
+  /// \brief Lazily read only live (non-deleted) manifest entries.
+  ///
+  /// The returned stream is fallible and single-pass. It must own all resources
+  /// required for consumption and must not depend on this reader remaining alive.
+  virtual Result<ManifestEntryStreamPtr> LiveEntriesStream() = 0;
 
   /// \brief Select specific columns of data file to read from the manifest entries.
   ///
diff --git a/src/iceberg/manifest/manifest_reader_internal.h b/src/iceberg/manifest/manifest_reader_internal.h
index 4ad708e43..99621fc88 100644
--- a/src/iceberg/manifest/manifest_reader_internal.h
+++ b/src/iceberg/manifest/manifest_reader_internal.h
@@ -62,9 +62,11 @@ class ManifestReaderImpl : public ManifestReader {
                      std::unique_ptr<InheritableMetadata> inheritable_metadata,
                      std::optional<int64_t> first_row_id, bool is_committed);
 
-  Result<std::vector<ManifestEntry>> Entries() override;
+  /// \brief Lazily read manifest entries.
+  Result<ManifestEntryStreamPtr> EntriesStream() override;
 
-  Result<std::vector<ManifestEntry>> LiveEntries() override;
+  /// \brief Lazily read only live (non-deleted) manifest entries.
+  Result<ManifestEntryStreamPtr> LiveEntriesStream() override;
 
   ManifestReader& Select(const std::vector<std::string>& columns) override;
 
@@ -81,8 +83,8 @@ class ManifestReaderImpl : public ManifestReader {
   ManifestReader& SkipCounter(std::shared_ptr<Counter> counter) override;
 
  private:
-  /// \brief Read entries with optional live-only filtering.
-  Result<std::vector<ManifestEntry>> ReadEntries(bool only_live);
+  /// \brief Create an entry stream with optional live-only filtering.
+  Result<ManifestEntryStreamPtr> MakeEntriesStream(bool only_live);
 
   /// \brief Lazily open the underlying Avro reader with appropriate schema projection.
   Status OpenReader(std::shared_ptr<Schema> projection);
@@ -93,14 +95,11 @@ class ManifestReaderImpl : public ManifestReader {
   /// \brief Check if there's a non-trivial row filter.
   bool HasRowFilter() const;
 
-  /// \brief Get or create the partition evaluator.
-  Result<Evaluator*> GetEvaluator();
+  /// \brief Get or create and transfer ownership of the partition evaluator.
+  Result<std::unique_ptr<Evaluator>> TakeEvaluator();
 
-  /// \brief Get or create the metrics evaluator.
-  Result<InclusiveMetricsEvaluator*> GetMetricsEvaluator();
-
-  /// \brief Check if a partition is in the partition set.
-  Result<bool> InPartitionSet(const DataFile& file) const;
+  /// \brief Get or create and transfer ownership of the metrics evaluator.
+  Result<std::unique_ptr<InclusiveMetricsEvaluator>> TakeMetricsEvaluator();
 
   // Fields set at construction
   const std::string manifest_path_;
@@ -108,7 +107,7 @@ class ManifestReaderImpl : public ManifestReader {
   const std::shared_ptr<FileIO> file_io_;
   const std::shared_ptr<Schema> schema_;
   const std::shared_ptr<PartitionSpec> spec_;
-  const std::unique_ptr<InheritableMetadata> inheritable_metadata_;
+  const std::shared_ptr<InheritableMetadata> inheritable_metadata_;
   std::optional<int64_t> first_row_id_;
   bool is_committed_;
 
diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index ef4e94c5b..99689e6dc 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -20,7 +20,9 @@
 #include "iceberg/table_scan.h"
 
 #include <algorithm>
+#include <chrono>
 #include <cstdint>
+#include <iterator>
 #include <utility>
 
 #include "iceberg/expression/binder.h"
@@ -64,6 +66,95 @@ const std::vector<std::string> kScanColumnsWithStats = [] {
   return cols;
 }();
 
+template <typename T>
+class EmptyStream final : public Stream<T> {
+ public:
+  Result<std::optional<T>> NextImpl() override { return std::nullopt; }
+};
+
+Result<ScanReport> MakeScanReport(const DataTableScan& scan, const Snapshot& snapshot,
+                                  ScanMetricsResult scan_metrics) {
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_ptr, scan.schema());
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto projected_id_set,
+      GetProjectedIdsVisitor::GetProjectedIds(*schema_ptr, /*include_struct_ids=*/true));
+  std::vector<int32_t> projected_field_ids(projected_id_set.begin(),
+                                           projected_id_set.end());
+  std::ranges::sort(projected_field_ids);
+
+  std::vector<std::string> projected_field_names;
+  projected_field_names.reserve(projected_field_ids.size());
+  for (int32_t field_id : projected_field_ids) {
+    ICEBERG_ASSIGN_OR_RAISE(auto field_name, schema_ptr->FindColumnNameById(field_id));
+    ICEBERG_CHECK(field_name.has_value(), "Projected field {} not found in schema",
+                  field_id);
+    projected_field_names.emplace_back(*field_name);
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto sanitized_filter,
+                          SanitizeExpression::Sanitize(*schema_ptr, scan.filter(),
+                                                       scan.context().case_sensitive));
+
+  return ScanReport{
+      .table_name = scan.context().table_name,
+      .snapshot_id = snapshot.snapshot_id,
+      .filter = std::move(sanitized_filter),
+      .schema_id = schema_ptr->schema_id(),
+      .projected_field_ids = std::move(projected_field_ids),
+      .projected_field_names = std::move(projected_field_names),
+      .scan_metrics = std::move(scan_metrics),
+      .metadata = scan.context().options,
+  };
+}
+
+class ReportingFileTaskStream final : public FileScanTaskStream {
+ public:
+  ReportingFileTaskStream(FileScanTaskStreamPtr stream,
+                          std::shared_ptr<ScanMetrics> scan_metrics,
+                          std::chrono::nanoseconds planning_duration,
+                          std::shared_ptr<MetricsReporter> reporter, ScanReport report)
+      : stream_(std::move(stream)),
+        scan_metrics_(std::move(scan_metrics)),
+        planning_duration_(std::move(planning_duration)),
+        reporter_(std::move(reporter)),
+        report_(std::move(report)) {}
+
+  ~ReportingFileTaskStream() override { Finalize(); }
+
+  Result<std::optional<std::shared_ptr<FileScanTask>>> NextImpl() override {
+    auto start = std::chrono::steady_clock::now();
+    auto result = stream_->Next();
+    planning_duration_ += std::chrono::duration_cast<std::chrono::nanoseconds>(
+        std::chrono::steady_clock::now() - start);
+    if (!result.has_value()) {
+      // Failed planning does not emit a successful scan report.
+      finalized_ = true;
+    } else if (!result.value().has_value()) {
+      Finalize();
+    }
+    return result;
+  }
+
+ private:
+  void Finalize() {
+    if (finalized_) {
+      return;
+    }
+    finalized_ = true;
+    scan_metrics_->total_planning_duration->Record(planning_duration_);
+    report_.scan_metrics = scan_metrics_->ToResult();
+    std::ignore = reporter_->Report(report_);
+  }
+
+  FileScanTaskStreamPtr stream_;
+  std::shared_ptr<ScanMetrics> scan_metrics_;
+  std::chrono::nanoseconds planning_duration_;
+  std::shared_ptr<MetricsReporter> reporter_;
+  ScanReport report_;
+  bool finalized_ = false;
+};
+
 }  // namespace
 
 namespace internal {
@@ -566,60 +657,23 @@ Result<std::unique_ptr<DataTableScan>> DataTableScan::Make(
       std::move(metadata), std::move(schema), std::move(io), std::move(context)));
 }
 
-Status DataTableScan::ReportScan(const Snapshot& snapshot,
-                                 const ScanMetrics& scan_metrics) const {
-  if (!context_.metrics_reporter) {
-    return {};
-  }
-
-  ICEBERG_ASSIGN_OR_RAISE(auto projected_schema, ResolveProjectedSchema());
-  const auto& schema_ptr = projected_schema.get();
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto projected_id_set,
-      GetProjectedIdsVisitor::GetProjectedIds(*schema_ptr, /*include_struct_ids=*/true));
-  std::vector<int32_t> projected_field_ids(projected_id_set.begin(),
-                                           projected_id_set.end());
-  std::ranges::sort(projected_field_ids);
-
-  std::vector<std::string> projected_field_names;
-  projected_field_names.reserve(projected_field_ids.size());
-  for (int32_t field_id : projected_field_ids) {
-    ICEBERG_ASSIGN_OR_RAISE(auto field_name, schema_ptr->FindColumnNameById(field_id));
-    ICEBERG_CHECK(field_name.has_value(), "Projected field {} not found in schema",
-                  field_id);
-    projected_field_names.emplace_back(*field_name);
-  }
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto sanitized_filter,
-      SanitizeExpression::Sanitize(*schema_ptr, filter(), context_.case_sensitive));
-
-  ScanReport report{
-      .table_name = context_.table_name,
-      .snapshot_id = snapshot.snapshot_id,
-      .filter = std::move(sanitized_filter),
-      .schema_id = schema_ptr->schema_id(),
-      .projected_field_ids = std::move(projected_field_ids),
-      .projected_field_names = std::move(projected_field_names),
-      .scan_metrics = scan_metrics.ToResult(),
-      .metadata = context_.options,
-  };
-  return context_.metrics_reporter->Report(report);
+Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() const {
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, PlanFilesStream());
+  return stream->ToVector();
 }
 
-Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() const {
+Result<FileScanTaskStreamPtr> DataTableScan::PlanFilesStream() const {
   ICEBERG_ASSIGN_OR_RAISE(auto snapshot, this->snapshot());
   if (!snapshot) {
-    return std::vector<std::shared_ptr<FileScanTask>>{};
+    return std::make_unique<EmptyStream<std::shared_ptr<FileScanTask>>>();
   }
 
   std::shared_ptr<ScanMetrics> scan_metrics;
-  std::optional<Timer::Timed> planning_duration;
+  std::optional<std::chrono::steady_clock::time_point> planning_start;
   if (context_.metrics_reporter) {
     auto metrics_context = MetricsContext::Default();
     scan_metrics = ScanMetrics::Make(*metrics_context);
-    planning_duration.emplace(scan_metrics->total_planning_duration->Start());
+    planning_start = std::chrono::steady_clock::now();
   }
 
   TableMetadataCache metadata_cache(metadata_.get());
@@ -636,11 +690,17 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() co
         static_cast<int64_t>(delete_manifests.size()));
   }
 
+  std::vector<ManifestFile> owned_data_manifests(
+      std::make_move_iterator(data_manifests.begin()),
+      std::make_move_iterator(data_manifests.end()));
+  std::vector<ManifestFile> owned_delete_manifests(
+      std::make_move_iterator(delete_manifests.begin()),
+      std::make_move_iterator(delete_manifests.end()));
+
   ICEBERG_ASSIGN_OR_RAISE(
       auto manifest_group,
-      ManifestGroup::Make(io_, schema_, specs_by_id,
-                          {data_manifests.begin(), data_manifests.end()},
-                          {delete_manifests.begin(), delete_manifests.end()}));
+      ManifestGroup::Make(io_, schema_, specs_by_id, std::move(owned_data_manifests),
+                          std::move(owned_delete_manifests)));
   manifest_group->CaseSensitive(context_.case_sensitive)
       .Select(ScanColumns())
       .FilterData(filter())
@@ -651,14 +711,24 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() co
   if (context_.ignore_residuals) {
     manifest_group->IgnoreResiduals();
   }
-  ICEBERG_ASSIGN_OR_RAISE(auto tasks, manifest_group->PlanFiles());
 
-  if (planning_duration) {
-    planning_duration->Stop();
-    std::ignore = ReportScan(*snapshot, *scan_metrics);
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, std::move(*manifest_group).PlanFilesStream());
+  if (!planning_start.has_value()) {
+    return stream;
+  }
+
+  auto planning_duration = std::chrono::duration_cast<std::chrono::nanoseconds>(
+      std::chrono::steady_clock::now() - planning_start.value());
+
+  auto report = MakeScanReport(*this, *snapshot, ScanMetricsResult{});
+  if (!report.has_value()) {
+    // Scan reporting is best effort.
+    return stream;
   }
 
-  return tasks;
+  return std::make_unique<ReportingFileTaskStream>(
+      std::move(stream), std::move(scan_metrics), planning_duration,
+      context_.metrics_reporter, std::move(report).value());
 }
 
 // Friend function template for IncrementalScan that implements the shared PlanFiles
@@ -764,7 +834,7 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFi
     manifest_group->IgnoreResiduals();
   }
 
-  return manifest_group->PlanFiles();
+  return std::move(*manifest_group).PlanFiles();
 }
 
 // IncrementalChangelogScan implementation
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index bee2b7d1d..11f2f3612 100644
--- a/src/iceberg/table_scan.h
+++ b/src/iceberg/table_scan.h
@@ -36,6 +36,7 @@
 #include "iceberg/type_fwd.h"
 #include "iceberg/util/error_collector.h"
 #include "iceberg/util/executor.h"
+#include "iceberg/util/stream.h"
 
 namespace iceberg {
 
@@ -96,6 +97,12 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   std::shared_ptr<Expression> residual_filter_;
 };
 
+/// \brief Stream of file scan tasks.
+using FileScanTaskStream = Stream<std::shared_ptr<FileScanTask>>;
+
+/// \brief Owning pointer to a file scan task stream.
+using FileScanTaskStreamPtr = std::unique_ptr<FileScanTaskStream>;
+
 enum class ChangelogOperation : uint8_t {
   kInsert,
   kDelete,
@@ -304,6 +311,9 @@ class ICEBERG_TEMPLATE_CLASS_EXPORT TableScanBuilder : public ErrorCollector {
 
   /// \brief Configure an executor for manifest planning.
   ///
+  /// The executor is borrowed and must remain alive throughout planning by scans built
+  /// from this builder and until any stream returned by PlanFilesStream() is destroyed.
+  ///
   /// \param executor Executor to use while planning manifests.
   /// \return Reference to this for method chaining.
   TableScanBuilder& PlanWith(Executor& executor);
@@ -460,11 +470,18 @@ class ICEBERG_EXPORT DataTableScan : public TableScan {
       std::shared_ptr<FileIO> io, internal::TableScanContext context);
 
   /// \brief Plans the scan tasks by resolving manifests and data files.
+  ///
+  /// Collects PlanFilesStream() into a vector.
   /// \return A Result containing scan tasks or an error.
   Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const;
 
- private:
-  Status ReportScan(const Snapshot& snapshot, const ScanMetrics& scan_metrics) const;
+  /// \brief Lazily plans scan tasks by resolving manifests and data files on demand.
+  ///
+  /// The returned fallible, single-pass stream owns its planning resources and
+  /// can outlive this scan. An executor configured through PlanWith() is borrowed and
+  /// must remain alive until the stream is destroyed, as later Next() calls may submit
+  /// work to it.
+  Result<FileScanTaskStreamPtr> PlanFilesStream() const;
 
  protected:
   using TableScan::TableScan;
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 5ca9fd915..4f9767496 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -135,7 +135,7 @@ add_iceberg_test(util_test
                  endian_test.cc
                  file_io_test.cc
                  formatter_test.cc
-                 iterator_test.cc
+                 stream_test.cc
                  lazy_test.cc
                  location_util_test.cc
                  math_util_internal_test.cc
diff --git a/src/iceberg/test/manifest_group_test.cc b/src/iceberg/test/manifest_group_test.cc
index aa2d6810d..1eef8a430 100644
--- a/src/iceberg/test/manifest_group_test.cc
+++ b/src/iceberg/test/manifest_group_test.cc
@@ -24,6 +24,7 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include <gmock/gmock.h>
@@ -282,7 +283,7 @@ TEST_P(ManifestGroupTest, CreateAndGetEntries) {
       testing::UnorderedElementsAre("/path/to/data1.parquet", "/path/to/data2.parquet"));
 
   // Verify PlanFiles returns data files with associated delete files
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, group->PlanFiles());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
   ASSERT_EQ(tasks.size(), 2);
   EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet",
                                                              "/path/to/data2.parquet"));
@@ -292,6 +293,155 @@ TEST_P(ManifestGroupTest, CreateAndGetEntries) {
   EXPECT_EQ(tasks[1]->delete_files()[0]->file_path, "/path/to/delete.parquet");
 }
 
+TEST_P(ManifestGroupTest, PlanFilesStreamPreservesSelectAllWithEqualityDeletes) {
+  auto version = GetParam();
+  if (version < 2) {
+    GTEST_SKIP() << "Delete files only supported in V2+";
+  }
+
+  constexpr int64_t kSnapshotId = 1000L;
+  const auto part_value = PartitionValues({Literal::Int(0)});
+
+  auto data_file = MakeDataFile("/path/to/data.parquet", part_value,
+                                partitioned_spec_->spec_id(), /*record_count=*/100);
+  data_file->lower_bounds[1] = Literal::Int(20).Serialize().value();
+  data_file->upper_bounds[1] = Literal::Int(30).Serialize().value();
+  std::vector<ManifestEntry> data_entries{MakeEntry(
+      ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1, std::move(data_file))};
+  auto data_manifest =
+      WriteDataManifest(version, kSnapshotId, std::move(data_entries), partitioned_spec_);
+
+  auto equality_delete = MakeEqualityDeleteFile("/path/to/equality-delete.parquet",
+                                                part_value, partitioned_spec_->spec_id());
+  equality_delete->lower_bounds[1] = Literal::Int(20).Serialize().value();
+  equality_delete->upper_bounds[1] = Literal::Int(30).Serialize().value();
+  std::vector<ManifestEntry> delete_entries{MakeEntry(ManifestStatus::kAdded, kSnapshotId,
+                                                      /*sequence_number=*/2,
+                                                      std::move(equality_delete))};
+  auto delete_manifest = WriteDeleteManifest(
+      version, kSnapshotId, std::move(delete_entries), partitioned_spec_);
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto group, ManifestGroup::Make(file_io_, schema_, GetSpecsById(), {data_manifest},
+                                      {delete_manifest}));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, std::move(*group).PlanFilesStream());
+  ICEBERG_UNWRAP_OR_FAIL(auto task, stream->Next());
+
+  ASSERT_TRUE(task.has_value());
+  EXPECT_EQ(task.value()->data_file()->file_path, "/path/to/data.parquet");
+  EXPECT_EQ(task.value()->data_file()->record_count, 100);
+  EXPECT_TRUE(task.value()->data_file()->lower_bounds.contains(1));
+  EXPECT_TRUE(task.value()->data_file()->upper_bounds.contains(1));
+  ASSERT_EQ(task.value()->delete_files().size(), 1);
+  EXPECT_EQ(task.value()->delete_files().front()->file_path,
+            "/path/to/equality-delete.parquet");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto end, stream->Next());
+  EXPECT_FALSE(end.has_value());
+}
+
+TEST_P(ManifestGroupTest, PlanFilesDropsUnselectedStatsWithEqualityDeletes) {
+  auto version = GetParam();
+  if (version < 2) {
+    GTEST_SKIP() << "Equality deletes only supported in V2+";
+  }
+
+  constexpr int64_t kSnapshotId = 1000L;
+  const auto part_value = PartitionValues({Literal::Int(0)});
+
+  auto data_file = MakeDataFile("/path/to/data.parquet", part_value,
+                                partitioned_spec_->spec_id(), /*record_count=*/100);
+  data_file->lower_bounds[1] = Literal::Int(20).Serialize().value();
+  data_file->upper_bounds[1] = Literal::Int(30).Serialize().value();
+  auto data_manifest =
+      WriteDataManifest(version, kSnapshotId,
+                        {MakeEntry(ManifestStatus::kAdded, kSnapshotId,
+                                   /*sequence_number=*/1, std::move(data_file))},
+                        partitioned_spec_);
+
+  auto equality_delete = MakeEqualityDeleteFile("/path/to/equality-delete.parquet",
+                                                part_value, partitioned_spec_->spec_id());
+  equality_delete->lower_bounds[1] = Literal::Int(20).Serialize().value();
+  equality_delete->upper_bounds[1] = Literal::Int(30).Serialize().value();
+  auto delete_manifest =
+      WriteDeleteManifest(version, kSnapshotId,
+                          {MakeEntry(ManifestStatus::kAdded, kSnapshotId,
+                                     /*sequence_number=*/2, std::move(equality_delete))},
+                          partitioned_spec_);
+
+  for (bool use_executor : {false, true}) {
+    SCOPED_TRACE(std::format("use_executor={}", use_executor));
+    for (bool use_stream : {false, true}) {
+      SCOPED_TRACE(std::format("use_stream={}", use_stream));
+      test::ThreadExecutor executor;
+      ICEBERG_UNWRAP_OR_FAIL(auto group,
+                             ManifestGroup::Make(file_io_, schema_, GetSpecsById(),
+                                                 {data_manifest}, {delete_manifest}));
+      group->Select({"file_path"});
+      if (use_executor) {
+        group->PlanWith(std::ref(executor));
+      }
+
+      std::vector<std::shared_ptr<FileScanTask>> tasks;
+      if (use_stream) {
+        ICEBERG_UNWRAP_OR_FAIL(auto stream, std::move(*group).PlanFilesStream());
+        ICEBERG_UNWRAP_OR_FAIL(tasks, stream->ToVector());
+      } else {
+        ICEBERG_UNWRAP_OR_FAIL(tasks, std::move(*group).PlanFiles());
+      }
+      ASSERT_EQ(tasks.size(), 1);
+      EXPECT_EQ(tasks.front()->data_file()->partition, part_value);
+      ASSERT_EQ(tasks.front()->delete_files().size(), 1);
+      EXPECT_EQ(tasks.front()->delete_files().front()->file_path,
+                "/path/to/equality-delete.parquet");
+      EXPECT_TRUE(tasks.front()->data_file()->lower_bounds.empty());
+      EXPECT_TRUE(tasks.front()->data_file()->upper_bounds.empty());
+    }
+  }
+}
+
+TEST_P(ManifestGroupTest, PlanFilesStreamPreservesPartitionWithPositionDeletes) {
+  auto version = GetParam();
+  if (version < 2) {
+    GTEST_SKIP() << "Position deletes only supported in V2+";
+  }
+
+  constexpr int64_t kSnapshotId = 1000L;
+  const auto part_value = PartitionValues({Literal::Int(0)});
+  auto data_manifest = WriteDataManifest(
+      version, kSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1,
+                 MakeDataFile("/path/to/data.parquet", part_value,
+                              partitioned_spec_->spec_id()))},
+      partitioned_spec_);
+  auto delete_manifest = WriteDeleteManifest(
+      version, kSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/2,
+                 MakePositionDeleteFile("/path/to/position-delete.parquet", part_value,
+                                        partitioned_spec_->spec_id()))},
+      partitioned_spec_);
+
+  for (bool use_executor : {false, true}) {
+    SCOPED_TRACE(std::format("use_executor={}", use_executor));
+    test::ThreadExecutor executor;
+    ICEBERG_UNWRAP_OR_FAIL(auto group,
+                           ManifestGroup::Make(file_io_, schema_, GetSpecsById(),
+                                               {data_manifest}, {delete_manifest}));
+    group->Select({"file_path"});
+    if (use_executor) {
+      group->PlanWith(std::ref(executor));
+    }
+
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, std::move(*group).PlanFilesStream());
+    ICEBERG_UNWRAP_OR_FAIL(auto tasks, stream->ToVector());
+    ASSERT_EQ(tasks.size(), 1);
+    EXPECT_EQ(tasks.front()->data_file()->partition, part_value);
+    ASSERT_EQ(tasks.front()->delete_files().size(), 1);
+    EXPECT_EQ(tasks.front()->delete_files().front()->file_path,
+              "/path/to/position-delete.parquet");
+  }
+}
+
 TEST_P(ManifestGroupTest, IgnoreDeleted) {
   auto version = GetParam();
 
@@ -323,7 +473,7 @@ TEST_P(ManifestGroupTest, IgnoreDeleted) {
   group->IgnoreDeleted();
 
   // Plan files - should only return ADDED and EXISTING
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, group->PlanFiles());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
   ASSERT_EQ(tasks.size(), 2);
   EXPECT_THAT(GetPaths(tasks),
               testing::UnorderedElementsAre("/path/to/added.parquet",
@@ -362,7 +512,7 @@ TEST_P(ManifestGroupTest, IgnoreExisting) {
   group->IgnoreExisting();
 
   // Plan files - should only return ADDED
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, group->PlanFiles());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
   ASSERT_EQ(tasks.size(), 2);
   EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/added.parquet",
                                                              "/path/to/deleted.parquet"));
@@ -400,7 +550,7 @@ TEST_P(ManifestGroupTest, CustomManifestEntriesFilter) {
   });
 
   // Plan files - should only return filtered entries
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, group->PlanFiles());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
   ASSERT_EQ(tasks.size(), 2);
   EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet",
                                                              "/path/to/data3.parquet"));
@@ -593,11 +743,11 @@ TEST_P(ManifestGroupTest, EmptyManifestGroup) {
       auto group,
       ManifestGroup::Make(file_io_, schema_, GetSpecsById(), std::move(manifests)));
 
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, group->PlanFiles());
-  EXPECT_TRUE(tasks.empty());
-
   ICEBERG_UNWRAP_OR_FAIL(auto entries, group->Entries());
   EXPECT_TRUE(entries.empty());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
+  EXPECT_TRUE(tasks.empty());
 }
 
 TEST_P(ManifestGroupTest, MultipleDataManifests) {
@@ -630,13 +780,48 @@ TEST_P(ManifestGroupTest, MultipleDataManifests) {
   group->PlanWith(std::ref(executor));
 
   // Plan files - should return files from both manifests
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, group->PlanFiles());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
   ASSERT_EQ(tasks.size(), 2);
   EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet",
                                                              "/path/to/data2.parquet"));
   EXPECT_EQ(executor.submit_count(), 2);
 }
 
+TEST_P(ManifestGroupTest, PlanFilesStreamUsesExecutor) {
+  auto version = GetParam();
+
+  const auto partition_a = PartitionValues({Literal::Int(0)});
+  const auto partition_b = PartitionValues({Literal::Int(1)});
+  auto data_manifest_1 =
+      WriteDataManifest(version, /*snapshot_id=*/1000L,
+                        {MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L,
+                                   /*sequence_number=*/1,
+                                   MakeDataFile("/path/to/data1.parquet", partition_a,
+                                                partitioned_spec_->spec_id()))},
+                        partitioned_spec_);
+  auto data_manifest_2 =
+      WriteDataManifest(version, /*snapshot_id=*/1001L,
+                        {MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1001L,
+                                   /*sequence_number=*/2,
+                                   MakeDataFile("/path/to/data2.parquet", partition_b,
+                                                partitioned_spec_->spec_id()))},
+                        partitioned_spec_);
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto group,
+      ManifestGroup::Make(file_io_, schema_, GetSpecsById(),
+                          {std::move(data_manifest_1), std::move(data_manifest_2)}));
+  test::ThreadExecutor executor;
+  group->PlanWith(std::ref(executor));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, std::move(*group).PlanFilesStream());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, stream->ToVector());
+
+  EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet",
+                                                             "/path/to/data2.parquet"));
+  EXPECT_EQ(executor.submit_count(), 2);
+}
+
 TEST_P(ManifestGroupTest, PartitionFilter) {
   auto version = GetParam();
 
@@ -666,7 +851,7 @@ TEST_P(ManifestGroupTest, PartitionFilter) {
   group->FilterPartitions(std::move(partition_filter));
 
   // Plan files - should only return the file in bucket 0
-  ICEBERG_UNWRAP_OR_FAIL(auto tasks, group->PlanFiles());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, std::move(*group).PlanFiles());
   ASSERT_EQ(tasks.size(), 1);
   EXPECT_THAT(GetPaths(tasks), testing::ElementsAre("/path/to/bucket0.parquet"));
 }
diff --git a/src/iceberg/test/manifest_reader_test.cc b/src/iceberg/test/manifest_reader_test.cc
index b57b0bc4a..d98643c12 100644
--- a/src/iceberg/test/manifest_reader_test.cc
+++ b/src/iceberg/test/manifest_reader_test.cc
@@ -190,6 +190,37 @@ TEST_P(TestManifestReader, TestManifestReaderWithEmptyInheritableMetadata) {
   EXPECT_EQ(read_entry.snapshot_id, 1000L);
 }
 
+TEST_P(TestManifestReader, EntriesStreamOwnsReaderResources) {
+  auto version = GetParam();
+  auto file_a =
+      MakeDataFile("/path/to/data-a.parquet", PartitionValues({Literal::Int(0)}));
+  auto file_b =
+      MakeDataFile("/path/to/data-b.parquet", PartitionValues({Literal::Int(1)}));
+
+  std::vector<ManifestEntry> entries;
+  entries.push_back(
+      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(file_a)));
+  entries.push_back(
+      MakeEntry(ManifestStatus::kAdded, /*snapshot_id=*/1000L, std::move(file_b)));
+  auto manifest = WriteManifest(version, /*snapshot_id=*/1000L, entries);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->EntriesStream());
+  reader.reset();
+
+  ICEBERG_UNWRAP_OR_FAIL(auto first, stream->Next());
+  ASSERT_TRUE(first.has_value());
+  EXPECT_EQ(first->data_file->file_path, "/path/to/data-a.parquet");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto second, stream->Next());
+  ASSERT_TRUE(second.has_value());
+  EXPECT_EQ(second->data_file->file_path, "/path/to/data-b.parquet");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto end, stream->Next());
+  EXPECT_FALSE(end.has_value());
+}
+
 TEST_P(TestManifestReader, DeletedEntriesDoNotInheritFirstRowId) {
   auto version = GetParam();
   if (version < 3) {
@@ -548,6 +579,12 @@ TEST(ManifestReaderStaticTest, TestShouldDropStats) {
       ManifestReader::ShouldDropStats({"file_path", "record_count", "value_counts"}));
 }
 
+TEST(ManifestReaderStaticTest, WithStatsColumnsPreservesSelectAll) {
+  EXPECT_TRUE(ManifestReader::WithStatsColumns({}).empty());
+  EXPECT_THAT(ManifestReader::WithStatsColumns({std::string(Schema::kAllColumns)}),
+              testing::ElementsAre(Schema::kAllColumns));
+}
+
 INSTANTIATE_TEST_SUITE_P(ManifestReaderVersions, TestManifestReader,
                          testing::Values(1, 2, 3));
 
diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build
index 6844df5e9..c1ca0c327 100644
--- a/src/iceberg/test/meson.build
+++ b/src/iceberg/test/meson.build
@@ -110,7 +110,6 @@ iceberg_tests = {
             'executor_util_test.cc',
             'file_io_test.cc',
             'formatter_test.cc',
-            'iterator_test.cc',
             'lazy_test.cc',
             'location_util_test.cc',
             'math_util_internal_test.cc',
@@ -119,6 +118,7 @@ iceberg_tests = {
             'resolving_file_io_test.cc',
             'retry_util_test.cc',
             'roaring_position_bitmap_test.cc',
+            'stream_test.cc',
             'string_util_test.cc',
             'struct_like_set_test.cc',
             'task_group_test.cc',
diff --git a/src/iceberg/test/scan_planning_metrics_test.cc b/src/iceberg/test/scan_planning_metrics_test.cc
index d3ef4fae9..dd01ffecd 100644
--- a/src/iceberg/test/scan_planning_metrics_test.cc
+++ b/src/iceberg/test/scan_planning_metrics_test.cc
@@ -48,6 +48,7 @@ namespace {
 class CapturingReporter final : public MetricsReporter {
  public:
   Status Report(const MetricsReport& report) override {
+    ++report_count_;
     if (std::holds_alternative<ScanReport>(report)) {
       last_ = std::get<ScanReport>(report);
     }
@@ -55,9 +56,11 @@ class CapturingReporter final : public MetricsReporter {
   }
 
   const std::optional<ScanReport>& last() const { return last_; }
+  int report_count() const { return report_count_; }
 
  private:
   std::optional<ScanReport> last_;
+  int report_count_ = 0;
 };
 
 }  // namespace
@@ -227,6 +230,8 @@ TEST_P(ScanPlanningMetricsTest, ReportsToTableAndScanReporters) {
   ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles());
   ASSERT_EQ(tasks.size(), 1u);
 
+  EXPECT_EQ(reporter_->report_count(), 1);
+  EXPECT_EQ(scan_reporter->report_count(), 1);
   ASSERT_TRUE(reporter_->last().has_value());
   const auto& report = *reporter_->last();
   EXPECT_EQ(report.table_name, "test.table");
@@ -241,6 +246,76 @@ TEST_P(ScanPlanningMetricsTest, ReportsToTableAndScanReporters) {
   EXPECT_EQ(scan_reporter->last()->table_name, "test.table");
 }
 
+TEST_P(ScanPlanningMetricsTest, StreamReportsWhenDestroyedEarly) {
+  auto version = GetParam();
+  constexpr int64_t kSnapshotId = 2010L;
+  const auto part = PartitionValues({Literal::Int(0)});
+
+  auto data_manifest = WriteDataManifest(
+      version, kSnapshotId,
+      {MakeEntry(
+           ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1,
+           MakeDataFile("/data/file_a.parquet", part, partitioned_spec_->spec_id())),
+       MakeEntry(
+           ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1,
+           MakeDataFile("/data/file_b.parquet", part, partitioned_spec_->spec_id()))},
+      partitioned_spec_);
+  auto manifest_list =
+      WriteManifestList(version, kSnapshotId, /*sequence_number=*/1, {data_manifest});
+  auto metadata =
+      BuildMetadata(version, kSnapshotId, /*sequence_number=*/1, manifest_list);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata));
+  builder->ReportWith(reporter_);
+  ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream());
+  scan.reset();
+
+  ICEBERG_UNWRAP_OR_FAIL(auto first, stream->Next());
+  ASSERT_TRUE(first.has_value());
+  EXPECT_EQ(reporter_->report_count(), 0);
+
+  stream.reset();
+  ASSERT_EQ(reporter_->report_count(), 1);
+  ASSERT_TRUE(reporter_->last().has_value());
+  const auto& metrics = reporter_->last()->scan_metrics;
+  ASSERT_TRUE(metrics.result_data_files.has_value());
+  EXPECT_EQ(metrics.result_data_files->value, 1);
+}
+
+TEST_P(ScanPlanningMetricsTest, DoesNotReportFailedPlanning) {
+  auto version = GetParam();
+  constexpr int64_t kSnapshotId = 2011L;
+  const auto part = PartitionValues({Literal::Int(0)});
+
+  auto missing_manifest = WriteDataManifest(
+      version, kSnapshotId,
+      {MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1,
+                 MakeDataFile("/data/file.parquet", part, partitioned_spec_->spec_id()))},
+      partitioned_spec_);
+  missing_manifest.manifest_path = "missing-data-manifest.avro";
+  auto manifest_list =
+      WriteManifestList(version, kSnapshotId, /*sequence_number=*/1, {missing_manifest});
+  auto metadata =
+      BuildMetadata(version, kSnapshotId, /*sequence_number=*/1, manifest_list);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto builder, MakeScanBuilder<DataTableScan>(metadata));
+  builder->ReportWith(reporter_);
+  ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream());
+
+  auto next = stream->Next();
+  EXPECT_FALSE(next.has_value());
+  EXPECT_EQ(reporter_->report_count(), 0);
+
+  stream.reset();
+  EXPECT_EQ(reporter_->report_count(), 0);
+
+  auto tasks = scan->PlanFiles();
+  EXPECT_FALSE(tasks.has_value());
+  EXPECT_EQ(reporter_->report_count(), 0);
+}
+
 TEST_P(ScanPlanningMetricsTest, ScanReportFilterUsesBoundCaseInsensitiveResolution) {
   auto version = GetParam();
   constexpr int64_t kSnapshotId = 2000L;
diff --git a/src/iceberg/test/iterator_test.cc b/src/iceberg/test/stream_test.cc
similarity index 66%
rename from src/iceberg/test/iterator_test.cc
rename to src/iceberg/test/stream_test.cc
index 49b338697..91b352167 100644
--- a/src/iceberg/test/iterator_test.cc
+++ b/src/iceberg/test/stream_test.cc
@@ -17,7 +17,7 @@
  * under the License.
  */
 
-#include "iceberg/util/iterator.h"
+#include "iceberg/util/stream.h"
 
 #include <memory>
 #include <optional>
@@ -50,7 +50,7 @@ static_assert(std::is_copy_constructible_v<CopyOnly>);
 static_assert(!std::is_move_constructible_v<CopyOnly>);
 
 // Exercises ToVector() with values that can be copied but not moved.
-class CopyOnlyIterator final : public Iterator<CopyOnly> {
+class CopyOnlyStream final : public Stream<CopyOnly> {
  private:
   Result<std::optional<CopyOnly>> NextImpl() override {
     if (next_ == 3) {
@@ -63,7 +63,7 @@ class CopyOnlyIterator final : public Iterator<CopyOnly> {
 };
 
 // Exercises ToVector() with values that can be moved but not copied.
-class MoveOnlyIterator final : public Iterator<std::unique_ptr<int>> {
+class MoveOnlyStream final : public Stream<std::unique_ptr<int>> {
  public:
   int calls() const { return calls_; }
 
@@ -82,7 +82,7 @@ class MoveOnlyIterator final : public Iterator<std::unique_ptr<int>> {
 };
 
 // Exercises ToVector() error propagation after some values have been consumed.
-class FailingIterator final : public Iterator<int> {
+class FailingStream final : public Stream<int> {
  public:
   int calls() const { return calls_; }
 
@@ -92,22 +92,28 @@ class FailingIterator final : public Iterator<int> {
     if (next_ < 2) {
       return Result<std::optional<int>>(std::in_place, std::in_place, next_++);
     }
-    return Invalid("iteration failed");
+    return Invalid("stream failed");
   }
 
   int next_ = 0;
   int calls_ = 0;
 };
 
-static_assert(std::is_move_constructible_v<CopyOnlyIterator>);
-static_assert(std::is_move_assignable_v<CopyOnlyIterator>);
-static_assert(std::is_move_constructible_v<MoveOnlyIterator>);
-static_assert(std::is_move_assignable_v<MoveOnlyIterator>);
+static_assert(std::is_move_constructible_v<CopyOnlyStream>);
+static_assert(std::is_move_assignable_v<CopyOnlyStream>);
+static_assert(std::is_move_constructible_v<MoveOnlyStream>);
+static_assert(std::is_move_assignable_v<MoveOnlyStream>);
 
-TEST(IteratorTest, ToVectorSupportsCopyOnlyValues) {
-  CopyOnlyIterator iterator;
+TEST(StreamTest, SupportsIncompleteSharedPointerValue) {
+  class IncompleteType;
+  std::unique_ptr<Stream<std::shared_ptr<IncompleteType>>> stream;
+  EXPECT_EQ(stream, nullptr);
+}
+
+TEST(StreamTest, ToVectorSupportsCopyOnlyValues) {
+  CopyOnlyStream stream;
 
-  ICEBERG_UNWRAP_OR_FAIL(auto values, iterator.ToVector());
+  ICEBERG_UNWRAP_OR_FAIL(auto values, stream.ToVector());
 
   ASSERT_EQ(values.size(), 3);
   EXPECT_EQ(values[0].value(), 0);
@@ -115,10 +121,10 @@ TEST(IteratorTest, ToVectorSupportsCopyOnlyValues) {
   EXPECT_EQ(values[2].value(), 2);
 }
 
-TEST(IteratorTest, ToVectorSupportsMoveOnlyValues) {
-  MoveOnlyIterator iterator;
+TEST(StreamTest, ToVectorSupportsMoveOnlyValues) {
+  MoveOnlyStream stream;
 
-  ICEBERG_UNWRAP_OR_FAIL(auto values, iterator.ToVector());
+  ICEBERG_UNWRAP_OR_FAIL(auto values, stream.ToVector());
 
   ASSERT_EQ(values.size(), 3);
   EXPECT_EQ(*values[0], 0);
@@ -126,46 +132,46 @@ TEST(IteratorTest, ToVectorSupportsMoveOnlyValues) {
   EXPECT_EQ(*values[2], 2);
 }
 
-TEST(IteratorTest, NextRemainsAtEndAfterExhaustion) {
-  MoveOnlyIterator iterator;
-  ICEBERG_UNWRAP_OR_FAIL(auto values, iterator.ToVector());
+TEST(StreamTest, NextRemainsAtEndAfterExhaustion) {
+  MoveOnlyStream stream;
+  ICEBERG_UNWRAP_OR_FAIL(auto values, stream.ToVector());
   ASSERT_EQ(values.size(), 3);
-  EXPECT_EQ(iterator.calls(), 4);
+  EXPECT_EQ(stream.calls(), 4);
 
   for (int i = 0; i < 2; ++i) {
-    auto result = iterator.Next();
+    auto result = stream.Next();
     ASSERT_TRUE(result.has_value());
     EXPECT_FALSE(result->has_value());
   }
-  EXPECT_EQ(iterator.calls(), 4);
+  EXPECT_EQ(stream.calls(), 4);
 }
 
-TEST(IteratorTest, ToVectorPropagatesErrorsAfterPartialConsumption) {
-  FailingIterator iterator;
+TEST(StreamTest, ToVectorPropagatesErrorsAfterPartialConsumption) {
+  FailingStream stream;
 
-  ICEBERG_UNWRAP_OR_FAIL(auto first, iterator.Next());
+  ICEBERG_UNWRAP_OR_FAIL(auto first, stream.Next());
   ASSERT_TRUE(first.has_value());
   EXPECT_EQ(first.value(), 0);
 
-  auto result = iterator.ToVector();
+  auto result = stream.ToVector();
 
   EXPECT_THAT(result, IsError(ErrorKind::kInvalid));
-  EXPECT_THAT(result, HasErrorMessage("iteration failed"));
+  EXPECT_THAT(result, HasErrorMessage("stream failed"));
 }
 
-TEST(IteratorTest, NextRepeatsErrorWithoutAdvancing) {
-  FailingIterator iterator;
-  auto first_error = iterator.ToVector();
+TEST(StreamTest, NextRepeatsErrorWithoutAdvancing) {
+  FailingStream stream;
+  auto first_error = stream.ToVector();
   EXPECT_THAT(first_error, IsError(ErrorKind::kInvalid));
-  EXPECT_THAT(first_error, HasErrorMessage("iteration failed"));
-  EXPECT_EQ(iterator.calls(), 3);
+  EXPECT_THAT(first_error, HasErrorMessage("stream failed"));
+  EXPECT_EQ(stream.calls(), 3);
 
   for (int i = 0; i < 2; ++i) {
-    auto result = iterator.Next();
+    auto result = stream.Next();
     EXPECT_THAT(result, IsError(ErrorKind::kInvalid));
-    EXPECT_THAT(result, HasErrorMessage("iteration failed"));
+    EXPECT_THAT(result, HasErrorMessage("stream failed"));
   }
-  EXPECT_EQ(iterator.calls(), 3);
+  EXPECT_EQ(stream.calls(), 3);
 }
 
 }  // namespace
diff --git a/src/iceberg/test/table_scan_test.cc b/src/iceberg/test/table_scan_test.cc
index 375c9aa53..ee447db06 100644
--- a/src/iceberg/test/table_scan_test.cc
+++ b/src/iceberg/test/table_scan_test.cc
@@ -320,6 +320,10 @@ TEST_P(TableScanTest, DataTableScanPlanFilesEmpty) {
   ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
   ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles());
   EXPECT_TRUE(tasks.empty());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream());
+  ICEBERG_UNWRAP_OR_FAIL(auto next, stream->Next());
+  EXPECT_FALSE(next.has_value());
 }
 
 TEST_P(TableScanTest, PlanFilesWithDataManifests) {
@@ -380,6 +384,14 @@ TEST_P(TableScanTest, PlanFilesWithDataManifests) {
   ASSERT_EQ(tasks.size(), 2);
   EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet",
                                                              "/path/to/data2.parquet"));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream());
+  scan.reset();
+  ICEBERG_UNWRAP_OR_FAIL(auto streamed_tasks, stream->ToVector());
+  ASSERT_EQ(streamed_tasks.size(), 2);
+  EXPECT_THAT(
+      GetPaths(streamed_tasks),
+      testing::UnorderedElementsAre("/path/to/data1.parquet", "/path/to/data2.parquet"));
 }
 
 TEST_P(TableScanTest, PlanRowLineage) {
@@ -604,22 +616,27 @@ TEST_P(TableScanTest, PlanFilesWithDeleteFiles) {
   std::vector<ManifestEntry> data_entries{
       MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1,
                 MakeDataFile("/path/to/data1.parquet", part_value,
-                             partitioned_spec_->spec_id(), /*record_count=*/100)),
+                             partitioned_spec_->spec_id(), /*record_count=*/100,
+                             /*lower_id=*/0, /*upper_id=*/10)),
       MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/1,
                 MakeDataFile("/path/to/data2.parquet", part_value,
-                             partitioned_spec_->spec_id(), /*record_count=*/200))};
+                             partitioned_spec_->spec_id(), /*record_count=*/200,
+                             /*lower_id=*/20, /*upper_id=*/30))};
   auto data_manifest =
       WriteDataManifest(version, kSnapshotId, std::move(data_entries), partitioned_spec_);
 
   // Create delete manifest with position delete files
+  auto equality_delete = MakeEqualityDeleteFile("/path/to/eq_delete.parquet", part_value,
+                                                partitioned_spec_->spec_id(), {1});
+  equality_delete->lower_bounds[1] = Literal::Int(20).Serialize().value();
+  equality_delete->upper_bounds[1] = Literal::Int(30).Serialize().value();
   std::vector<ManifestEntry> delete_entries{
       MakeEntry(
           ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/2,
           MakePositionDeleteFile("/path/to/pos_delete.parquet", part_value,
                                  partitioned_spec_->spec_id(), "/path/to/data1.parquet")),
       MakeEntry(ManifestStatus::kAdded, kSnapshotId, /*sequence_number=*/2,
-                MakeEqualityDeleteFile("/path/to/eq_delete.parquet", part_value,
-                                       partitioned_spec_->spec_id(), {1}))};
+                std::move(equality_delete))};
   auto delete_manifest = WriteDeleteManifest(
       version, kSnapshotId, std::move(delete_entries), partitioned_spec_);
   std::string manifest_list_path = WriteManifestList(
@@ -662,13 +679,25 @@ TEST_P(TableScanTest, PlanFilesWithDeleteFiles) {
                          MakeScanBuilder<DataTableScan>(metadata_with_manifests));
   ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
   ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanFiles());
-  ASSERT_EQ(tasks.size(), 2);
-  EXPECT_THAT(GetPaths(tasks), testing::UnorderedElementsAre("/path/to/data1.parquet",
-                                                             "/path/to/data2.parquet"));
-  // Verify that delete files are associated with the tasks
-  for (const auto& task : tasks) {
-    EXPECT_GT(task->delete_files().size(), 0);
-  }
+  auto verify_tasks = [](const auto& planned_tasks) {
+    ASSERT_EQ(planned_tasks.size(), 2);
+    for (const auto& task : planned_tasks) {
+      ASSERT_EQ(task->delete_files().size(), 1);
+      if (task->data_file()->file_path == "/path/to/data1.parquet") {
+        EXPECT_EQ(task->delete_files().front()->file_path, "/path/to/pos_delete.parquet");
+      } else {
+        EXPECT_EQ(task->data_file()->file_path, "/path/to/data2.parquet");
+        EXPECT_EQ(task->delete_files().front()->file_path, "/path/to/eq_delete.parquet");
+      }
+      EXPECT_TRUE(task->data_file()->lower_bounds.empty());
+      EXPECT_TRUE(task->data_file()->upper_bounds.empty());
+    }
+  };
+  verify_tasks(tasks);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->PlanFilesStream());
+  ICEBERG_UNWRAP_OR_FAIL(auto streamed_tasks, stream->ToVector());
+  verify_tasks(streamed_tasks);
 }
 
 TEST_P(TableScanTest, SchemaWithSelectedColumnsAndFilter) {
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
index 0b19adaf5..f45dd0115 100644
--- a/src/iceberg/type_fwd.h
+++ b/src/iceberg/type_fwd.h
@@ -230,7 +230,7 @@ struct SessionContext;
 /// \brief Task execution.
 class Executor;
 template <typename T>
-class Iterator;
+class Stream;
 
 /// \brief Metrics reporting.
 class MetricsReporter;
diff --git a/src/iceberg/util/meson.build b/src/iceberg/util/meson.build
index 831cc5888..c819bbfe1 100644
--- a/src/iceberg/util/meson.build
+++ b/src/iceberg/util/meson.build
@@ -32,7 +32,7 @@ install_headers(
         'formatter.h',
         'functional.h',
         'int128.h',
-        'iterator.h',
+        'stream.h',
         'lazy.h',
         'location_util.h',
         'macros.h',
diff --git a/src/iceberg/util/iterator.h b/src/iceberg/util/stream.h
similarity index 69%
rename from src/iceberg/util/iterator.h
rename to src/iceberg/util/stream.h
index 16717499f..5bca51cbc 100644
--- a/src/iceberg/util/iterator.h
+++ b/src/iceberg/util/stream.h
@@ -19,8 +19,8 @@
 
 #pragma once
 
-/// \file iceberg/util/iterator.h
-/// \brief Pull-based iterator interface for fallible, lazily produced values.
+/// \file iceberg/util/stream.h
+/// \brief Pull-based stream interface for fallible, lazily produced values.
 
 #include <deque>
 #include <optional>
@@ -32,27 +32,41 @@
 
 namespace iceberg {
 
-/// \brief A pull-based iterator whose reads may fail.
+/// \brief A pull-based stream whose reads may fail.
 ///
-/// Iterator implementations own any resources needed to produce values. Destroying an
-/// iterator releases those resources, including when iteration stops before reaching the
-/// end. Iterators are not thread-safe unless an implementation explicitly says otherwise.
-/// Once Next() returns an error or std::nullopt, the iterator is terminal. Subsequent
+/// Stream implementations own any resources needed to produce values. Destroying a
+/// stream releases those resources, including when consumption stops before reaching the
+/// end. Streams are not thread-safe unless an implementation explicitly says otherwise.
+/// Once Next() returns an error or std::nullopt, the stream is terminal. Subsequent
 /// calls return the same terminal result without invoking the implementation again.
 ///
-/// \tparam T Value returned by the iterator.
+/// This interface replaces Iterator<T>. Include iceberg/util/stream.h instead of
+/// iceberg/util/iterator.h and derive from Stream<T>, retaining the NextImpl() override.
+/// The former header and type alias are not provided.
+///
+/// \tparam T Value returned by the stream.
 template <typename T>
-class Iterator {
+class Stream {
  public:
-  virtual ~Iterator() = default;
+  /// \brief Destroy this stream and release its producer resources.
+  virtual ~Stream() = default;
+
+  /// \brief Construct a stream in its initial state.
+  Stream() = default;
+
+  /// \brief Streams cannot be copied.
+  Stream(const Stream&) = delete;
+
+  /// \brief Streams cannot be copy-assigned.
+  Stream& operator=(const Stream&) = delete;
+
+  /// \brief Move a stream and its terminal state.
+  Stream(Stream&&) noexcept = default;
 
-  Iterator() = default;
-  Iterator(const Iterator&) = delete;
-  Iterator& operator=(const Iterator&) = delete;
-  Iterator(Iterator&&) noexcept = default;
-  Iterator& operator=(Iterator&&) noexcept = default;
+  /// \brief Move-assign a stream and its terminal state.
+  Stream& operator=(Stream&&) noexcept = default;
 
-  /// \brief Return the next value, or std::nullopt when the iterator is exhausted.
+  /// \brief Return the next value, or std::nullopt when the stream is exhausted.
   ///
   /// After this method returns an error or std::nullopt, subsequent calls return the same
   /// terminal result without invoking NextImpl().
@@ -92,7 +106,7 @@ class Iterator {
 
     if constexpr (!std::is_move_constructible_v<T>) {
       static_assert(std::is_copy_constructible_v<T>,
-                    "Iterator::ToVector requires T to be move- or copy-constructible");
+                    "Stream::ToVector requires T to be move- or copy-constructible");
 
       // For strictly copy-only T, collecting directly into a vector can repeatedly copy
       // previously collected elements during vector growth. Stage values in a deque,