diff --git a/example/demo_example.cc b/example/demo_example.cc
index 22ecc0c90..a477e5af1 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->PlanFilesIterator();
   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 3db0a1df4..c47c9a0bc 100644
--- a/src/iceberg/manifest/manifest_group.cc
+++ b/src/iceberg/manifest/manifest_group.cc
@@ -132,6 +132,223 @@ ManifestGroup::~ManifestGroup() = default;
 ManifestGroup::ManifestGroup(ManifestGroup&&) noexcept = default;
 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(
+      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());
+
+    const bool drop_stats = ManifestReader::ShouldDropStats(group->columns_);
+    if (delete_index->has_equality_deletes()) {
+      group->columns_ = ManifestReader::WithStatsColumns(group->columns_);
+    }
+
+    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_));
+    }
+
+    return std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>(
+        new FilePlanningIterator(std::move(group), std::move(delete_index),
+                                 std::move(data_file_evaluator), drop_stats));
+  }
+
+  Result<std::optional<std::shared_ptr<FileScanTask>>> Next() override {
+    while (true) {
+      if (!entry_iterator_) {
+        ICEBERG_ASSIGN_OR_RAISE(bool opened, OpenNextManifest());
+        if (!opened) {
+          return std::nullopt;
+        }
+      }
+
+      ICEBERG_ASSIGN_OR_RAISE(auto entry, entry_iterator_->Next());
+      if (!entry.has_value()) {
+        entry_iterator_.reset();
+        continue;
+      }
+
+      auto 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;
+      }
+
+      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_);
+      }
+
+      ICEBERG_ASSIGN_OR_RAISE(auto delete_files, delete_index_->ForEntry(value));
+      UpdateResultMetrics(*value.data_file, delete_files);
+
+      ICEBERG_ASSIGN_OR_RAISE(auto residuals, GetResidualEvaluator(current_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:
+  FilePlanningIterator(std::unique_ptr<ManifestGroup> group,
+                       std::unique_ptr<DeleteFileIndex> delete_index,
+                       std::unique_ptr<Evaluator> data_file_evaluator, bool drop_stats)
+      : group_(std::move(group)),
+        delete_index_(std::move(delete_index)),
+        data_file_evaluator_(std::move(data_file_evaluator)),
+        drop_stats_(drop_stats) {}
+
+  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> OpenNextManifest() {
+    while (next_manifest_ < group_->data_manifests_.size()) {
+      const auto& manifest = group_->data_manifests_[next_manifest_++];
+      const int32_t spec_id = manifest.partition_spec_id;
+
+      ICEBERG_ASSIGN_OR_RAISE(auto evaluator, GetManifestEvaluator(spec_id));
+      ICEBERG_ASSIGN_OR_RAISE(bool should_match, evaluator->Evaluate(manifest));
+      if (!should_match) {
+        IncrementSkippedDataManifests();
+        continue;
+      }
+      if (group_->ignore_deleted_ && !manifest.has_added_files() &&
+          !manifest.has_existing_files()) {
+        IncrementSkippedDataManifests();
+        continue;
+      }
+      if (group_->ignore_existing_ && !manifest.has_added_files() &&
+          !manifest.has_deleted_files()) {
+        IncrementSkippedDataManifests();
+        continue;
+      }
+
+      if (group_->scan_metrics_) {
+        group_->scan_metrics_->scanned_data_manifests->Increment(1);
+      }
+
+      ICEBERG_ASSIGN_OR_RAISE(auto reader, group_->MakeReader(manifest));
+      ICEBERG_ASSIGN_OR_RAISE(entry_iterator_, group_->ignore_deleted_
+                                                   ? reader->LiveEntriesIterator()
+                                                   : reader->EntriesIterator());
+      current_spec_id_ = spec_id;
+      return true;
+    }
+    return false;
+  }
+
+  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::unordered_map<int32_t, std::unique_ptr<ManifestEvaluator>> manifest_evaluators_;
+  std::unordered_map<int32_t, std::shared_ptr<ResidualEvaluator>> residual_evaluators_;
+  std::unique_ptr<Iterator<ManifestEntry>> entry_iterator_;
+  size_t next_manifest_ = 0;
+  int32_t current_spec_id_ = 0;
+  bool drop_stats_;
+};
+
 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));
@@ -252,6 +469,12 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> ManifestGroup::PlanFiles() {
   return file_tasks;
 }
 
+Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
+ManifestGroup::PlanFilesIterator() {
+  auto group = std::make_unique<ManifestGroup>(std::move(*this));
+  return FilePlanningIterator::Make(std::move(group));
+}
+
 Result<std::vector<std::shared_ptr<ScanTask>>> ManifestGroup::Plan(
     const CreateTasksFunction& create_tasks) {
   std::unordered_map<int32_t, std::shared_ptr<ResidualEvaluator>> residual_cache;
diff --git a/src/iceberg/manifest/manifest_group.h b/src/iceberg/manifest/manifest_group.h
index be0ca4b8b..c9778747a 100644
--- a/src/iceberg/manifest/manifest_group.h
+++ b/src/iceberg/manifest/manifest_group.h
@@ -37,6 +37,7 @@
 #include "iceberg/type_fwd.h"
 #include "iceberg/util/error_collector.h"
 #include "iceberg/util/executor.h"
+#include "iceberg/util/iterator.h"
 
 namespace iceberg {
 
@@ -136,6 +137,15 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
   /// \brief Plan scan tasks for all matching data files.
   Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles();
 
+  /// \brief Lazily plan scan tasks for matching data files.
+  ///
+  /// The returned iterator owns the planning state and may outlive this ManifestGroup.
+  /// It reads one manifest batch at a time instead of materializing all manifest entries
+  /// and scan tasks. Creating the iterator consumes this group's configuration. Streaming
+  /// planning is pull-based and does not eagerly submit manifests to the executor set by
+  /// PlanWith().
+  Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>> PlanFilesIterator();
+
   /// \brief Get all matching manifest entries.
   Result<std::vector<ManifestEntry>> Entries();
 
@@ -151,6 +161,8 @@ class ICEBERG_EXPORT ManifestGroup : public ErrorCollector {
       const CreateTasksFunction& create_tasks);
 
  private:
+  class FilePlanningIterator;
+
   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,
diff --git a/src/iceberg/manifest/manifest_reader.cc b/src/iceberg/manifest/manifest_reader.cc
index 68760b440..6b67da116 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,8 +690,141 @@ Result<std::shared_ptr<Schema>> ProjectSchema(std::shared_ptr<Schema> schema,
   return schema;
 }
 
+template <typename T>
+class VectorIterator final : public Iterator<T> {
+ public:
+  explicit VectorIterator(std::vector<T> values) : values_(std::move(values)) {}
+
+  Result<std::optional<T>> Next() override {
+    if (next_ == values_.size()) {
+      return std::nullopt;
+    }
+    return std::optional<T>{std::move(values_[next_++])};
+  }
+
+ private:
+  std::vector<T> values_;
+  size_t next_ = 0;
+};
+
+class ManifestEntryIteratorImpl final : public Iterator<ManifestEntry> {
+ public:
+  ManifestEntryIteratorImpl(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>> Next() 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
 
+Result<std::unique_ptr<Iterator<ManifestEntry>>> ManifestReader::EntriesIterator() {
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, Entries());
+  return std::make_unique<VectorIterator<ManifestEntry>>(std::move(entries));
+}
+
+Result<std::unique_ptr<Iterator<ManifestEntry>>> ManifestReader::LiveEntriesIterator() {
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, LiveEntries());
+  return std::make_unique<VectorIterator<ManifestEntry>>(std::move(entries));
+}
+
 bool ManifestReader::ShouldDropStats(const std::vector<std::string>& columns) {
   // Make sure we only drop all stats if we had projected all stats.
   // We do not drop stats even if we had partially added some stats columns, except for
@@ -861,14 +995,26 @@ Status ManifestReaderImpl::OpenReader(std::shared_ptr<Schema> projection) {
 }
 
 Result<std::vector<ManifestEntry>> ManifestReaderImpl::Entries() {
-  return ReadEntries(/*only_live=*/false);
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, EntriesIterator());
+  return entries->ToVector();
 }
 
 Result<std::vector<ManifestEntry>> ManifestReaderImpl::LiveEntries() {
-  return ReadEntries(/*only_live=*/true);
+  ICEBERG_ASSIGN_OR_RAISE(auto entries, LiveEntriesIterator());
+  return entries->ToVector();
+}
+
+Result<std::unique_ptr<Iterator<ManifestEntry>>> ManifestReaderImpl::EntriesIterator() {
+  return MakeEntriesIterator(/*only_live=*/false);
+}
+
+Result<std::unique_ptr<Iterator<ManifestEntry>>>
+ManifestReaderImpl::LiveEntriesIterator() {
+  return MakeEntriesIterator(/*only_live=*/true);
 }
 
-Result<std::vector<ManifestEntry>> ManifestReaderImpl::ReadEntries(bool only_live) {
+Result<std::unique_ptr<Iterator<ManifestEntry>>> ManifestReaderImpl::MakeEntriesIterator(
+    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 +1040,29 @@ 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_RETURN_UNEXPECTED(GetEvaluator());
+    evaluator = std::move(evaluator_);
   }
   if (HasRowFilter()) {
-    ICEBERG_ASSIGN_OR_RAISE(metrics_evaluator, GetMetricsEvaluator());
+    ICEBERG_RETURN_UNEXPECTED(GetMetricsEvaluator());
+    metrics_evaluator = std::move(metrics_evaluator_);
   }
 
   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 iterator = std::unique_ptr<Iterator<ManifestEntry>>(new ManifestEntryIteratorImpl(
+      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 iterator;
 }
 
 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..a789e112b 100644
--- a/src/iceberg/manifest/manifest_reader.h
+++ b/src/iceberg/manifest/manifest_reader.h
@@ -33,6 +33,7 @@
 #include "iceberg/metrics/counter.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
+#include "iceberg/util/iterator.h"
 
 namespace iceberg {
 
@@ -42,13 +43,24 @@ class ICEBERG_EXPORT ManifestReader {
   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;
 
   /// \brief Read only live (non-deleted) manifest entries.
   virtual Result<std::vector<ManifestEntry>> LiveEntries() = 0;
 
+  /// \brief Lazily read manifest entries.
+  ///
+  /// The returned iterator reads and filters one underlying record batch at a time. This
+  /// bounds memory use for large manifests. The iterator owns its reader resources and
+  /// may outlive this ManifestReader.
+  virtual Result<std::unique_ptr<Iterator<ManifestEntry>>> EntriesIterator();
+
+  /// \brief Lazily read only live (non-deleted) manifest entries.
+  ///
+  /// The default implementation adapts LiveEntries() for compatibility with custom reader
+  /// implementations. Built-in readers override this with a streaming implementation.
+  virtual Result<std::unique_ptr<Iterator<ManifestEntry>>> LiveEntriesIterator();
+
   /// \brief Select specific columns of data file to read from the manifest entries.
   ///
   /// \note Column names should match the names in `DataFile` schema. Unmatched names
diff --git a/src/iceberg/manifest/manifest_reader_internal.h b/src/iceberg/manifest/manifest_reader_internal.h
index 4ad708e43..da2335484 100644
--- a/src/iceberg/manifest/manifest_reader_internal.h
+++ b/src/iceberg/manifest/manifest_reader_internal.h
@@ -66,6 +66,10 @@ class ManifestReaderImpl : public ManifestReader {
 
   Result<std::vector<ManifestEntry>> LiveEntries() override;
 
+  Result<std::unique_ptr<Iterator<ManifestEntry>>> EntriesIterator() override;
+
+  Result<std::unique_ptr<Iterator<ManifestEntry>>> LiveEntriesIterator() override;
+
   ManifestReader& Select(const std::vector<std::string>& columns) override;
 
   ManifestReader& FilterPartitions(std::shared_ptr<Expression> expr) override;
@@ -81,8 +85,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 iterator with optional live-only filtering.
+  Result<std::unique_ptr<Iterator<ManifestEntry>>> MakeEntriesIterator(bool only_live);
 
   /// \brief Lazily open the underlying Avro reader with appropriate schema projection.
   Status OpenReader(std::shared_ptr<Schema> projection);
@@ -108,7 +112,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..54a6d4794 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -20,6 +20,7 @@
 #include "iceberg/table_scan.h"
 
 #include <algorithm>
+#include <chrono>
 #include <cstdint>
 #include <utility>
 
@@ -64,6 +65,93 @@ const std::vector<std::string> kScanColumnsWithStats = [] {
   return cols;
 }();
 
+template <typename T>
+class EmptyIterator final : public Iterator<T> {
+ public:
+  Result<std::optional<T>> Next() 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 ReportingFileTaskIterator final : public Iterator<std::shared_ptr<FileScanTask>> {
+ public:
+  ReportingFileTaskIterator(
+      std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>> iterator,
+      std::shared_ptr<ScanMetrics> scan_metrics,
+      std::chrono::nanoseconds planning_duration,
+      std::shared_ptr<MetricsReporter> reporter, ScanReport report)
+      : iterator_(std::move(iterator)),
+        scan_metrics_(std::move(scan_metrics)),
+        planning_duration_(std::move(planning_duration)),
+        reporter_(std::move(reporter)),
+        report_(std::move(report)) {}
+
+  ~ReportingFileTaskIterator() override { Finalize(); }
+
+  Result<std::optional<std::shared_ptr<FileScanTask>>> Next() override {
+    auto start = std::chrono::steady_clock::now();
+    auto result = iterator_->Next();
+    planning_duration_ += std::chrono::duration_cast<std::chrono::nanoseconds>(
+        std::chrono::steady_clock::now() - start);
+    if (result.has_value() && !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_);
+  }
+
+  std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>> iterator_;
+  std::shared_ptr<ScanMetrics> scan_metrics_;
+  std::chrono::nanoseconds planning_duration_;
+  std::shared_ptr<MetricsReporter> reporter_;
+  ScanReport report_;
+  bool finalized_ = false;
+};
+
 }  // namespace
 
 namespace internal {
@@ -572,39 +660,8 @@ Status DataTableScan::ReportScan(const Snapshot& snapshot,
     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,
-  };
+  ICEBERG_ASSIGN_OR_RAISE(auto report,
+                          MakeScanReport(*this, snapshot, scan_metrics.ToResult()));
   return context_.metrics_reporter->Report(report);
 }
 
@@ -661,6 +718,71 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() co
   return tasks;
 }
 
+Result<std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>>
+DataTableScan::PlanFilesIterator() const {
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot, this->snapshot());
+  if (!snapshot) {
+    return std::make_unique<EmptyIterator<std::shared_ptr<FileScanTask>>>();
+  }
+
+  std::shared_ptr<ScanMetrics> scan_metrics;
+  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_start = std::chrono::steady_clock::now();
+  }
+
+  TableMetadataCache metadata_cache(metadata_.get());
+  ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById());
+
+  SnapshotCache snapshot_cache(snapshot.get());
+  ICEBERG_ASSIGN_OR_RAISE(auto data_manifests, snapshot_cache.DataManifests(io_));
+  ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests, snapshot_cache.DeleteManifests(io_));
+
+  if (scan_metrics) {
+    scan_metrics->total_data_manifests->Increment(
+        static_cast<int64_t>(data_manifests.size()));
+    scan_metrics->total_delete_manifests->Increment(
+        static_cast<int64_t>(delete_manifests.size()));
+  }
+
+  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()}));
+  manifest_group->CaseSensitive(context_.case_sensitive)
+      .Select(ScanColumns())
+      .FilterData(filter())
+      .IgnoreDeleted()
+      .ColumnsToKeepStats(context_.columns_to_keep_stats)
+      .PlanWith(context_.plan_executor)
+      .WithScanMetrics(scan_metrics);
+  if (context_.ignore_residuals) {
+    manifest_group->IgnoreResiduals();
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto iterator, manifest_group->PlanFilesIterator());
+  if (!planning_start.has_value()) {
+    return iterator;
+  }
+
+  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, matching PlanFiles().
+    return iterator;
+  }
+
+  return std::unique_ptr<Iterator<std::shared_ptr<FileScanTask>>>(
+      new ReportingFileTaskIterator(std::move(iterator), std::move(scan_metrics),
+                                    planning_duration, context_.metrics_reporter,
+                                    std::move(report).value()));
+}
+
 // Friend function template for IncrementalScan that implements the shared PlanFiles
 // logic. It resolves the from/to snapshot range from the scan context and delegates
 // to the two-arg virtual PlanFiles() override in the concrete subclass.
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index bee2b7d1d..6c99e4c5f 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/iterator.h"
 
 namespace iceberg {
 
@@ -463,6 +464,13 @@ class ICEBERG_EXPORT DataTableScan : public TableScan {
   /// \return A Result containing scan tasks or an error.
   Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const;
 
+  /// \brief Lazily plans scan tasks by resolving manifests and data files on demand.
+  ///
+  /// 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;
+
  private:
   Status ReportScan(const Snapshot& snapshot, const ScanMetrics& scan_metrics) const;
 
diff --git a/src/iceberg/test/manifest_reader_test.cc b/src/iceberg/test/manifest_reader_test.cc
index b57b0bc4a..8c5ff5314 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, EntriesIteratorOwnsReaderResources) {
+  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 iterator, reader->EntriesIterator());
+  reader.reset();
+
+  ICEBERG_UNWRAP_OR_FAIL(auto first, iterator->Next());
+  ASSERT_TRUE(first.has_value());
+  EXPECT_EQ(first->data_file->file_path, "/path/to/data-a.parquet");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto second, iterator->Next());
+  ASSERT_TRUE(second.has_value());
+  EXPECT_EQ(second->data_file->file_path, "/path/to/data-b.parquet");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto end, iterator->Next());
+  EXPECT_FALSE(end.has_value());
+}
+
 TEST_P(TestManifestReader, DeletedEntriesDoNotInheritFirstRowId) {
   auto version = GetParam();
   if (version < 3) {
diff --git a/src/iceberg/test/table_scan_test.cc b/src/iceberg/test/table_scan_test.cc
index 375c9aa53..5a57c9bcf 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 iterator, scan->PlanFilesIterator());
+  ICEBERG_UNWRAP_OR_FAIL(auto next, iterator->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 iterator, scan->PlanFilesIterator());
+  scan.reset();
+  ICEBERG_UNWRAP_OR_FAIL(auto streamed_tasks, iterator->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) {
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
index 0d2b17fae..0b19adaf5 100644
--- a/src/iceberg/type_fwd.h
+++ b/src/iceberg/type_fwd.h
@@ -229,6 +229,8 @@ struct SessionContext;
 
 /// \brief Task execution.
 class Executor;
+template <typename T>
+class Iterator;
 
 /// \brief Metrics reporting.
 class MetricsReporter;
diff --git a/src/iceberg/util/iterator.h b/src/iceberg/util/iterator.h
new file mode 100644
index 000000000..09c06fc91
--- /dev/null
+++ b/src/iceberg/util/iterator.h
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/util/iterator.h
+/// \brief Pull-based iterator interface for fallible, lazily produced values.
+
+#include <memory>
+#include <optional>
+#include <utility>
+#include <vector>
+
+#include "iceberg/result.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+/// \brief A pull-based iterator 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.
+///
+/// \tparam T Value returned by the iterator.
+template <typename T>
+class Iterator {
+ public:
+  virtual ~Iterator() = default;
+
+  Iterator() = default;
+  Iterator(const Iterator&) = delete;
+  Iterator& operator=(const Iterator&) = delete;
+
+  /// \brief Return the next value, or std::nullopt when the iterator is exhausted.
+  virtual Result<std::optional<T>> Next() = 0;
+
+  /// \brief Consume the remaining values into a vector.
+  Result<std::vector<T>> ToVector() {
+    std::vector<T> values;
+    while (true) {
+      ICEBERG_ASSIGN_OR_RAISE(auto value, Next());
+      if (!value.has_value()) {
+        return values;
+      }
+      values.push_back(std::move(value).value());
+    }
+  }
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/util/meson.build b/src/iceberg/util/meson.build
index f9436e7ed..831cc5888 100644
--- a/src/iceberg/util/meson.build
+++ b/src/iceberg/util/meson.build
@@ -32,6 +32,7 @@ install_headers(
         'formatter.h',
         'functional.h',
         'int128.h',
+        'iterator.h',
         'lazy.h',
         'location_util.h',
         'macros.h',