diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index cf465e5fd..eee720613 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -61,6 +61,8 @@ set(ICEBERG_SOURCES
     inspect/history_table.cc
     inspect/metadata_scan.cc
     inspect/metadata_table.cc
+    inspect/metadata_scan_task.cc
+    inspect/metadata_scan_task_reader.cc
     inspect/snapshots_table.cc
     inheritable_metadata.cc
     json_serde.cc
diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h
index d186e5bb0..6bcc5b742 100644
--- a/src/iceberg/cpp20_compatibility_internal.h
+++ b/src/iceberg/cpp20_compatibility_internal.h
@@ -82,6 +82,8 @@
 #include "iceberg/inheritable_metadata.h"
 #include "iceberg/inspect/history_table.h"
 #include "iceberg/inspect/metadata_scan.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_task_reader.h"
 #include "iceberg/inspect/metadata_table.h"
 #include "iceberg/inspect/snapshots_table.h"
 #include "iceberg/labels.h"
diff --git a/src/iceberg/inspect/history_table.cc b/src/iceberg/inspect/history_table.cc
index 518073ca0..a7f58e5b5 100644
--- a/src/iceberg/inspect/history_table.cc
+++ b/src/iceberg/inspect/history_table.cc
@@ -19,24 +19,67 @@
 
 #include "iceberg/inspect/history_table.h"
 
+#include <array>
 #include <memory>
+#include <unordered_map>
+#include <unordered_set>
 #include <utility>
-#include <vector>
 
+#include "iceberg/arrow_row_builder_internal.h"
+#include "iceberg/constants.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_util_internal.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
 #include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg {
+namespace {
 
-HistoryTable::HistoryTable(std::shared_ptr<Table> table)
+struct HistoryRow {
+  const SnapshotLogEntry& entry;
+  const Snapshot* snapshot;
+  bool is_current_ancestor;
+};
+
+constexpr std::array<internal::MetadataTableColumn<HistoryRow>, 4> kHistoryColumns = {
+    {{.field_id = 1,
+      .append =
+          [](ArrowArray* column, const HistoryRow& row) {
+            return AppendInt(column,
+                             UnixMsFromTimePointMs(row.entry.timestamp_ms) * 1000);
+          }},
+     {.field_id = 2,
+      .append =
+          [](ArrowArray* column, const HistoryRow& row) {
+            return AppendInt(column, row.entry.snapshot_id);
+          }},
+     {.field_id = 3,
+      .append =
+          [](ArrowArray* column, const HistoryRow& row) {
+            return row.snapshot != nullptr && row.snapshot->parent_snapshot_id
+                       ? AppendInt(column, *row.snapshot->parent_snapshot_id)
+                       : AppendNull(column);
+          }},
+     {.field_id = 4, .append = [](ArrowArray* column, const HistoryRow& row) {
+        return AppendBoolean(column, row.is_current_ancestor);
+      }}}};
+
+}  // namespace
+
+// HistoryTable implementation
+HistoryTable::HistoryTable(std::shared_ptr<const Table> table)
     : MetadataTable(std::move(table)) {}
 
 HistoryTable::~HistoryTable() = default;
 
-const std::shared_ptr<Schema>& HistoryTable::schema() const {
+const std::shared_ptr<Schema>& HistoryTable::schema() const { return CanonicalSchema(); }
+
+const std::shared_ptr<Schema>& HistoryTable::CanonicalSchema() {
   static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
       SchemaField::MakeRequired(2, "snapshot_id", int64()),
@@ -45,14 +88,81 @@ const std::shared_ptr<Schema>& HistoryTable::schema() const {
   return schema;
 }
 
-Result<std::unique_ptr<HistoryTable>> HistoryTable::Make(std::shared_ptr<Table> table) {
+Result<std::unique_ptr<HistoryTable>> HistoryTable::Make(
+    std::shared_ptr<const Table> table) {
   ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
   return std::unique_ptr<HistoryTable>(new HistoryTable(std::move(table)));
 }
 
-Result<ArrowArrayStream> HistoryTable::ScanProjection(
-    const Schema& /*projected_schema*/) const {
-  return NotSupported("Scan is not supported for the history table");
+Result<std::unique_ptr<MetadataScan>> HistoryTable::MakeScan(
+    internal::MetadataScanContext context) const {
+  return HistoryScan::Make(std::move(context));
+}
+
+Result<ArrowArrayStream> HistoryTable::ReadRows(
+    std::shared_ptr<const TableMetadata> metadata,
+    const std::shared_ptr<Schema>& projected_schema, int64_t batch_size) {
+  std::unordered_map<int64_t, const Snapshot*> snapshots_by_id;
+  for (const auto& snapshot : metadata->snapshots) {
+    if (snapshot) {
+      snapshots_by_id.emplace(snapshot->snapshot_id, snapshot.get());
+    }
+  }
+
+  // Walk the current snapshot's ancestry; the insert check stops on cycles.
+  std::unordered_set<int64_t> ancestors;
+  for (auto current = metadata->current_snapshot_id;
+       current != kInvalidSnapshotId && ancestors.insert(current).second;) {
+    auto it = snapshots_by_id.find(current);
+    if (it == snapshots_by_id.end() || !it->second->parent_snapshot_id) {
+      break;
+    }
+    current = *it->second->parent_snapshot_id;
+  }
+
+  const size_t num_rows = metadata->snapshot_log.size();
+  return internal::MakeMetadataBatchStream<HistoryRow>(
+      CanonicalSchema(), projected_schema, kHistoryColumns, num_rows, batch_size,
+      [metadata = std::move(metadata), snapshots_by_id = std::move(snapshots_by_id),
+       ancestors = std::move(ancestors)](size_t index) -> HistoryRow {
+        const auto& entry = metadata->snapshot_log[index];
+        auto it = snapshots_by_id.find(entry.snapshot_id);
+        return {.entry = entry,
+                .snapshot = it != snapshots_by_id.end() ? it->second : nullptr,
+                .is_current_ancestor = ancestors.contains(entry.snapshot_id)};
+      });
+}
+
+// HistoryScan implementation
+HistoryScan::HistoryScan(internal::MetadataScanContext context,
+                         std::shared_ptr<MetadataScanTask> task)
+    : MetadataScan(std::move(context)), task_(std::move(task)) {}
+
+HistoryScan::~HistoryScan() = default;
+
+Result<std::unique_ptr<HistoryScan>> HistoryScan::Make(
+    internal::MetadataScanContext context) {
+  ICEBERG_PRECHECK(context.kind == MetadataTableKind::kHistory,
+                   "Unexpected metadata table kind for history scan");
+  if (context.snapshot_id.has_value() || !context.ref_name.empty() ||
+      context.as_of_time_in_millis.has_value()) {
+    return NotSupported("Metadata table does not support time travel");
+  }
+  ICEBERG_PRECHECK(context.table_metadata != nullptr, "Table metadata cannot be null");
+  ICEBERG_PRECHECK(context.schema != nullptr, "Metadata table schema cannot be null");
+  ICEBERG_PRECHECK(context.projected_schema != nullptr,
+                   "Projected schema cannot be null");
+  ICEBERG_RETURN_UNEXPECTED(internal::SelectAppendFns<HistoryRow>(
+      *HistoryTable::CanonicalSchema(), *context.projected_schema, kHistoryColumns));
+  ICEBERG_ASSIGN_OR_RAISE(auto task,
+                          StaticMetadataTask::Make(context.kind, context.table_metadata,
+                                                   context.projected_schema));
+  return std::unique_ptr<HistoryScan>(
+      new HistoryScan(std::move(context), std::move(task)));
+}
+
+Result<MetadataScanTaskStreamPtr> HistoryScan::PlanTasksStream() const {
+  return MetadataScanTaskStream::FromVector({task_});
 }
 
 }  // namespace iceberg
diff --git a/src/iceberg/inspect/history_table.h b/src/iceberg/inspect/history_table.h
index 6f07bcd08..ec18e36a4 100644
--- a/src/iceberg/inspect/history_table.h
+++ b/src/iceberg/inspect/history_table.h
@@ -22,9 +22,12 @@
 /// \file iceberg/inspect/history_table.h
 /// \brief Define the history metadata table.
 
+#include <cstdint>
 #include <memory>
 
+#include "iceberg/arrow_c_data.h"
 #include "iceberg/iceberg_export.h"
+#include "iceberg/inspect/metadata_scan.h"
 #include "iceberg/inspect/metadata_table.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -34,19 +37,46 @@ namespace iceberg {
 /// \brief History metadata table.
 class ICEBERG_EXPORT HistoryTable : public MetadataTable {
  public:
-  static Result<std::unique_ptr<HistoryTable>> Make(std::shared_ptr<Table> table);
+  static Result<std::unique_ptr<HistoryTable>> Make(std::shared_ptr<const Table> table);
+
+  /// \brief Return the canonical schema of the history metadata table.
+  static const std::shared_ptr<Schema>& CanonicalSchema();
 
   ~HistoryTable() override;
 
-  Kind kind() const noexcept override { return Kind::kHistory; }
+  MetadataTableKind kind() const noexcept override { return MetadataTableKind::kHistory; }
 
   const std::shared_ptr<Schema>& schema() const override;
 
  protected:
-  Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema) const override;
+  Result<std::unique_ptr<MetadataScan>> MakeScan(
+      internal::MetadataScanContext context) const override;
 
  private:
-  explicit HistoryTable(std::shared_ptr<Table> table);
+  friend class MetadataScanTaskReader;
+
+  static Result<ArrowArrayStream> ReadRows(
+      std::shared_ptr<const TableMetadata> metadata,
+      const std::shared_ptr<Schema>& projected_schema, int64_t batch_size);
+
+  explicit HistoryTable(std::shared_ptr<const Table> table);
+};
+
+/// \brief Scan the history metadata captured from a table.
+class ICEBERG_EXPORT HistoryScan final : public MetadataScan {
+ public:
+  /// \brief Create a scan from captured metadata table settings.
+  static Result<std::unique_ptr<HistoryScan>> Make(internal::MetadataScanContext context);
+
+  ~HistoryScan() override;
+
+  Result<MetadataScanTaskStreamPtr> PlanTasksStream() const override;
+
+ private:
+  HistoryScan(internal::MetadataScanContext context,
+              std::shared_ptr<MetadataScanTask> task);
+
+  const std::shared_ptr<MetadataScanTask> task_;
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan.cc b/src/iceberg/inspect/metadata_scan.cc
index 6117c28c6..e961023c1 100644
--- a/src/iceberg/inspect/metadata_scan.cc
+++ b/src/iceberg/inspect/metadata_scan.cc
@@ -19,25 +19,113 @@
 
 #include "iceberg/inspect/metadata_scan.h"
 
-#include <memory>
+#include <optional>
 #include <utility>
 
+#include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/inspect/metadata_scan_task_reader.h"
 #include "iceberg/inspect/metadata_table.h"
 #include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/table.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/stream.h"
 
 namespace iceberg {
+namespace {
+
+class TaskArrowStreamSource {
+ public:
+  TaskArrowStreamSource(std::vector<std::shared_ptr<MetadataScanTask>> tasks,
+                        std::unique_ptr<MetadataScanTaskReader> reader,
+                        std::shared_ptr<Schema> schema)
+      : tasks_(std::move(tasks)),
+        reader_(std::move(reader)),
+        schema_(std::move(schema)) {}
+
+  ~TaskArrowStreamSource() { std::ignore = Close(); }
+
+  Status Close() {
+    if (current_.release != nullptr) {
+      current_.release(&current_);
+    }
+    return {};
+  }
+
+  Result<ArrowSchema> Schema() {
+    ArrowSchema schema{};
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*schema_, &schema));
+    return schema;
+  }
+
+  Result<std::optional<ArrowArray>> Next() {
+    if (error_) {
+      return iceberg::unexpected(*error_);
+    }
+    auto result = NextBatch();
+    if (!result) {
+      error_ = result.error();
+    }
+    return result;
+  }
+
+  Result<std::optional<ArrowArray>> NextBatch() {
+    while (true) {
+      if (current_.release == nullptr) {
+        if (next_task_ == tasks_.size()) {
+          return std::nullopt;
+        }
+        ICEBERG_ASSIGN_OR_RAISE(current_, reader_->Open(*tasks_[next_task_++]));
+      }
+
+      ArrowArray array{};
+      const int error = current_.get_next(&current_, &array);
+      if (error != 0) {
+        std::string message = current_.get_last_error != nullptr &&
+                                      current_.get_last_error(&current_) != nullptr
+                                  ? current_.get_last_error(&current_)
+                                  : "Metadata task stream failed";
+        current_.release(&current_);
+        return IOError("{}", message);
+      }
+      if (array.release != nullptr) {
+        return array;
+      }
+      current_.release(&current_);
+    }
+  }
+
+  const std::vector<std::shared_ptr<MetadataScanTask>> tasks_;
+  const std::unique_ptr<MetadataScanTaskReader> reader_;
+  const std::shared_ptr<iceberg::Schema> schema_;
+  size_t next_task_ = 0;
+  ArrowArrayStream current_{};
+  std::optional<Error> error_;
+};
+
+}  // namespace
 
 MetadataScanBuilder::MetadataScanBuilder(const MetadataTable& table)
-    : table_(&table), projected_schema_(table.schema()) {}
+    : table_(&table),
+      context_{.kind = table.kind(),
+               .table_metadata = table.source_table()->metadata(),
+               .schema = table.schema(),
+               .projected_schema = table.schema()} {}
 
 MetadataScanBuilder::~MetadataScanBuilder() = default;
 
+Result<std::unique_ptr<MetadataScanBuilder>> MetadataScanBuilder::Make(
+    const MetadataTable& table) {
+  ICEBERG_PRECHECK(table.source_table() != nullptr, "Source table cannot be null");
+  ICEBERG_PRECHECK(table.schema() != nullptr, "Metadata table schema cannot be null");
+  return std::unique_ptr<MetadataScanBuilder>(new MetadataScanBuilder(table));
+}
+
 MetadataScanBuilder& MetadataScanBuilder::Select(
     const std::vector<std::string>& columns) {
-  auto projected = table_->schema()->Select(columns, case_sensitive_);
+  auto projected = context_.schema->Select(columns, context_.case_sensitive);
   if (projected) {
-    projected_schema_ = std::move(projected).value();
+    context_.projected_schema = std::move(projected).value();
   } else {
     AddError(projected.error());
   }
@@ -45,44 +133,61 @@ MetadataScanBuilder& MetadataScanBuilder::Select(
 }
 
 MetadataScanBuilder& MetadataScanBuilder::Project(std::shared_ptr<Schema> schema) {
-  projected_schema_ = std::move(schema);
+  context_.projected_schema = std::move(schema);
   return *this;
 }
 
 MetadataScanBuilder& MetadataScanBuilder::CaseSensitive(bool value) {
-  case_sensitive_ = value;
+  context_.case_sensitive = value;
   return *this;
 }
 
 MetadataScanBuilder& MetadataScanBuilder::UseSnapshot(int64_t snapshot_id) {
   ICEBERG_BUILDER_CHECK(
-      std::holds_alternative<std::monostate>(snapshot_selection_.snapshot),
+      !context_.snapshot_id.has_value() && !context_.as_of_time_in_millis.has_value(),
       "Cannot override snapshot, a snapshot or timestamp is already set");
-  snapshot_selection_.snapshot = snapshot_id;
+  context_.snapshot_id = snapshot_id;
   return *this;
 }
 
 MetadataScanBuilder& MetadataScanBuilder::UseRef(const std::string& ref) {
-  ICEBERG_BUILDER_CHECK(snapshot_selection_.ref_name.empty(),
-                        "Cannot override ref, already set ref={}",
-                        snapshot_selection_.ref_name);
+  ICEBERG_BUILDER_CHECK(context_.ref_name.empty(),
+                        "Cannot override ref, already set ref={}", context_.ref_name);
   ICEBERG_BUILDER_CHECK(!ref.empty(), "Ref name cannot be empty");
-  snapshot_selection_.ref_name = ref;
+  context_.ref_name = ref;
   return *this;
 }
 
-MetadataScanBuilder& MetadataScanBuilder::AsOfTime(int64_t timestamp_millis) {
+MetadataScanBuilder& MetadataScanBuilder::AsOfTime(int64_t as_of_time_in_millis) {
   ICEBERG_BUILDER_CHECK(
-      std::holds_alternative<std::monostate>(snapshot_selection_.snapshot),
+      !context_.snapshot_id.has_value() && !context_.as_of_time_in_millis.has_value(),
       "Cannot override snapshot, a snapshot or timestamp is already set");
-  snapshot_selection_.snapshot = TimePointMsFromUnixMs(timestamp_millis);
+  context_.as_of_time_in_millis = as_of_time_in_millis;
   return *this;
 }
 
-Result<ArrowArrayStream> MetadataScanBuilder::Build() {
+Result<std::unique_ptr<MetadataScan>> MetadataScanBuilder::Build() {
   ICEBERG_RETURN_UNEXPECTED(CheckErrors());
-  ICEBERG_PRECHECK(projected_schema_ != nullptr, "Projected schema cannot be null");
-  return table_->ScanSnapshot(snapshot_selection_, *projected_schema_);
+  ICEBERG_PRECHECK(context_.projected_schema != nullptr,
+                   "Projected schema cannot be null");
+  return table_->MakeScan(context_);
+}
+
+MetadataScan::MetadataScan(internal::MetadataScanContext context)
+    : context_(std::move(context)) {}
+
+MetadataScan::~MetadataScan() = default;
+
+Result<std::vector<std::shared_ptr<MetadataScanTask>>> MetadataScan::PlanTasks() const {
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, PlanTasksStream());
+  return stream->ToVector();
+}
+
+Result<ArrowArrayStream> MetadataScan::ToArrowArrayStream() const {
+  ICEBERG_ASSIGN_OR_RAISE(auto tasks, PlanTasks());
+  ICEBERG_ASSIGN_OR_RAISE(auto reader, MetadataScanTaskReader::Make({}));
+  return MakeArrowArrayStream(std::make_unique<TaskArrowStreamSource>(
+      std::move(tasks), std::move(reader), context_.projected_schema));
 }
 
 }  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan.h b/src/iceberg/inspect/metadata_scan.h
index f487bce4e..5df4cc034 100644
--- a/src/iceberg/inspect/metadata_scan.h
+++ b/src/iceberg/inspect/metadata_scan.h
@@ -20,12 +20,12 @@
 #pragma once
 
 /// \file iceberg/inspect/metadata_scan.h
-/// \brief Scan configuration for Iceberg metadata tables.
+/// \brief Configure metadata scans, plan reusable tasks, and read Arrow batches.
 
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <string>
-#include <variant>
 #include <vector>
 
 #include "iceberg/arrow_c_data.h"
@@ -33,104 +33,101 @@
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
 #include "iceberg/util/error_collector.h"
-#include "iceberg/util/timepoint.h"
+#include "iceberg/util/stream.h"
 
 namespace iceberg {
 
-/// \brief Snapshot selection parameters for a time-travel scan.
-struct SnapshotSelection {
-  /// \brief Select the current snapshot, a snapshot ID, or an as-of timestamp.
-  ///
-  /// std::monostate selects the current snapshot.
-  std::variant<std::monostate, int64_t, TimePointMs> snapshot;
+using MetadataScanTaskStream = Stream<std::shared_ptr<MetadataScanTask>>;
+using MetadataScanTaskStreamPtr = std::unique_ptr<MetadataScanTaskStream>;
 
-  /// \brief Resolve the snapshot relative to this branch or tag.
-  ///
-  /// An empty string uses the main branch.
+namespace internal {
+
+/// \brief Captured metadata and logical settings shared by a builder and its scans.
+struct MetadataScanContext {
+  MetadataTableKind kind;
+  std::shared_ptr<const TableMetadata> table_metadata;
+  std::shared_ptr<Schema> schema;
+  std::shared_ptr<Schema> projected_schema;
+  bool case_sensitive = true;
+  std::optional<int64_t> snapshot_id;
   std::string ref_name;
+  std::optional<int64_t> as_of_time_in_millis;
 };
 
-/// \brief Configure a scan over a metadata table.
-///
-/// Builders are created by MetadataTable::NewScan(), and the metadata table must
-/// outlive the builder.
-///
-/// \code
-///   ICEBERG_ASSIGN_OR_RAISE(auto builder, snapshots_table->NewScan());
-///   ICEBERG_ASSIGN_OR_RAISE(auto stream, builder->Select({"snapshot_id"}).Build());
-/// \endcode
+}  // namespace internal
+
+/// \brief Configure a scan over captured metadata; the metadata table must outlive it.
 class ICEBERG_EXPORT MetadataScanBuilder : public ErrorCollector {
  public:
   ~MetadataScanBuilder() override;
 
-  /// \brief Select columns by name; nested columns use dot-separated paths.
-  ///
-  /// Columns are returned in schema order, and names that match no column are
-  /// ignored. An empty list selects no columns; `*` selects all columns. Names are
-  /// resolved immediately, using the current CaseSensitive() setting. Selecting a
-  /// map's value returns the whole map; selecting only its key is an error.
+  /// \brief Capture the source metadata and schema of a metadata table.
+  static Result<std::unique_ptr<MetadataScanBuilder>> Make(const MetadataTable& table);
+
+  /// \brief Select names using the current CaseSensitive() setting.
   ///
-  /// \param columns Column names from the metadata table's schema.
+  /// Columns are returned in schema order; unknown names are ignored. An empty
+  /// list selects no columns and `*` selects all columns. Nested names use dots;
+  /// selecting a map value selects the whole map, while selecting only its key fails.
   MetadataScanBuilder& Select(const std::vector<std::string>& columns);
 
-  /// \brief Set the output schema to a projection of the metadata table schema.
-  ///
-  /// Each field must select a whole column of the metadata table by field id, with
-  /// the same type and without tightening its nullability. Fields may be reordered.
+  /// \brief Project whole columns by field ID, optionally reordering them.
   ///
-  /// \param schema The output schema.
+  /// Types must match and nullable columns cannot become required.
   MetadataScanBuilder& Project(std::shared_ptr<Schema> schema);
 
-  /// \brief Use case-sensitive column names in Select(); enabled by default.
-  ///
-  /// \param value Whether column names are matched case-sensitively.
+  /// \brief Match column names case-sensitively; enabled by default.
   MetadataScanBuilder& CaseSensitive(bool value);
 
-  /// \brief Request this scan to use the given snapshot by ID.
-  ///
-  /// Only metadata tables that support time travel accept a snapshot; Build()
-  /// returns NotSupported otherwise. The table resolves the snapshot when the scan
-  /// is built.
-  ///
-  /// \param snapshot_id A snapshot ID.
+  /// \brief Select a snapshot ID; Build rejects tables without time travel.
   MetadataScanBuilder& UseSnapshot(int64_t snapshot_id);
 
-  /// \brief Request this scan to resolve the snapshot on the given branch or tag.
-  ///
-  /// Only metadata tables that support time travel accept a reference; Build()
-  /// returns NotSupported otherwise. The table resolves the reference when the scan
-  /// is built.
-  ///
-  /// \param ref A branch or tag name.
+  /// \brief Select a branch or tag; Build rejects tables without time travel.
   MetadataScanBuilder& UseRef(const std::string& ref);
 
-  /// \brief Request this scan to use the most recent snapshot as of the given time.
-  ///
-  /// The snapshot is resolved on the branch set by UseRef(), or main if no branch
-  /// is set. Only metadata tables that support time travel accept a timestamp;
-  /// Build() returns NotSupported otherwise.
+  /// \brief Select the latest snapshot at or before this timestamp on the selected ref.
   ///
-  /// \param timestamp_millis A timestamp in milliseconds since the Unix epoch.
-  MetadataScanBuilder& AsOfTime(int64_t timestamp_millis);
+  /// The timestamp is milliseconds since the Unix epoch. Build rejects tables
+  /// without time travel.
+  MetadataScanBuilder& AsOfTime(int64_t as_of_time_in_millis);
 
-  /// \brief Validate the settings and scan the metadata table.
+  /// \brief Validate the settings and build a scan with its own copy of them.
   ///
-  /// Each call returns a new stream with the output schema. The caller owns the
-  /// returned stream and must invoke its `release` callback when the stream is no
-  /// longer needed.
-  ///
-  /// \return An Arrow stream containing the selected columns, or an error.
-  Result<ArrowArrayStream> Build();
+  /// Later builder changes and normal source table refreshes do not affect it.
+  Result<std::unique_ptr<MetadataScan>> Build();
 
  private:
-  friend class MetadataTable;
-
   explicit MetadataScanBuilder(const MetadataTable& table);
 
   const MetadataTable* table_;
-  std::shared_ptr<Schema> projected_schema_;
-  bool case_sensitive_ = true;
-  SnapshotSelection snapshot_selection_;
+  internal::MetadataScanContext context_;
+};
+
+/// \brief A metadata scan that plans tasks or reads all of them as Arrow batches.
+class ICEBERG_EXPORT MetadataScan {
+ public:
+  virtual ~MetadataScan();
+
+  /// \brief Return a fresh task stream; consuming it may report planning errors.
+  virtual Result<MetadataScanTaskStreamPtr> PlanTasksStream() const = 0;
+
+  /// \brief Collect every task from PlanTasksStream(), preserving planning order.
+  virtual Result<std::vector<std::shared_ptr<MetadataScanTask>>> PlanTasks() const;
+
+  /// \brief Read every task in planning order as one Arrow stream.
+  ///
+  /// The stream owns its read state and can outlive this scan. Release the stream
+  /// and each returned schema or array independently. To set batch size, open the
+  /// planned tasks with MetadataScanTaskReader; each task retains its projection.
+  virtual Result<ArrowArrayStream> ToArrowArrayStream() const;
+
+ protected:
+  explicit MetadataScan(internal::MetadataScanContext context);
+
+  const internal::MetadataScanContext& context() const { return context_; }
+
+ private:
+  const internal::MetadataScanContext context_;
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task.cc b/src/iceberg/inspect/metadata_scan_task.cc
new file mode 100644
index 000000000..afb3eb106
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task.cc
@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+
+#include "iceberg/inspect/metadata_scan_task.h"
+
+#include <utility>
+
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+// StaticMetadataTask
+
+Result<std::shared_ptr<StaticMetadataTask>> StaticMetadataTask::Make(
+    MetadataTableKind table_kind, std::shared_ptr<const TableMetadata> metadata,
+    std::shared_ptr<Schema> projected_schema) {
+  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+  ICEBERG_PRECHECK(projected_schema != nullptr, "Projected schema cannot be null");
+  if (table_kind != MetadataTableKind::kSnapshots &&
+      table_kind != MetadataTableKind::kHistory) {
+    return NotImplemented("Static metadata table kind is not implemented");
+  }
+  return std::shared_ptr<StaticMetadataTask>(new StaticMetadataTask(
+      table_kind, std::move(metadata), std::move(projected_schema)));
+}
+
+StaticMetadataTask::StaticMetadataTask(MetadataTableKind table_kind,
+                                       std::shared_ptr<const TableMetadata> metadata,
+                                       std::shared_ptr<Schema> projected_schema)
+    : table_kind_(table_kind),
+      metadata_(std::move(metadata)),
+      projected_schema_(std::move(projected_schema)) {}
+
+}  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task.h b/src/iceberg/inspect/metadata_scan_task.h
new file mode 100644
index 000000000..b3511e949
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task.h
@@ -0,0 +1,88 @@
+/*
+ * 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
+
+#include <memory>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Base class for metadata table scan tasks.
+///
+/// Metadata tables expose table metadata (snapshots, history, etc.) as queryable
+/// Arrow tables. A task is a unit of work opened by MetadataScanTaskReader.
+class ICEBERG_EXPORT MetadataScanTask {
+ public:
+  /// \brief Task category, determines how the reader opens the task.
+  enum class TaskKind {
+    /// StaticMetadataTask reads in-memory table metadata (snapshots, history).
+    kStatic,
+  };
+
+  virtual ~MetadataScanTask() = default;
+
+  /// \brief Return how this task is read.
+  virtual TaskKind task_kind() const = 0;
+
+  /// \brief Return the metadata table whose rows this task produces.
+  virtual MetadataTableKind table_kind() const = 0;
+};
+
+/// \brief Task for static metadata tables (snapshots, history).
+///
+/// Holds the metadata captured for a table scan; readers produce Arrow batches
+/// on demand. The metadata and its nested objects must not be modified while
+/// this task or its streams are in use.
+class ICEBERG_EXPORT StaticMetadataTask : public MetadataScanTask {
+ public:
+  /// \brief Create a static metadata task.
+  ///
+  /// \param table_kind The metadata table kind (kSnapshots or kHistory)
+  /// \param metadata The table metadata to scan
+  /// \param projected_schema The planned output schema
+  /// \return A task, or an error for null metadata or an unsupported table kind.
+  static Result<std::shared_ptr<StaticMetadataTask>> Make(
+      MetadataTableKind table_kind, std::shared_ptr<const TableMetadata> metadata,
+      std::shared_ptr<Schema> projected_schema);
+
+  TaskKind task_kind() const override { return TaskKind::kStatic; }
+
+  MetadataTableKind table_kind() const override { return table_kind_; }
+
+  /// \brief Return the captured metadata and its shared ownership.
+  const std::shared_ptr<const TableMetadata>& metadata() const { return metadata_; }
+
+  /// \brief Return the output schema captured when this task was planned.
+  const std::shared_ptr<Schema>& projected_schema() const { return projected_schema_; }
+
+ private:
+  StaticMetadataTask(MetadataTableKind table_kind,
+                     std::shared_ptr<const TableMetadata> metadata,
+                     std::shared_ptr<Schema> projected_schema);
+
+  const MetadataTableKind table_kind_;
+  const std::shared_ptr<const TableMetadata> metadata_;
+  const std::shared_ptr<Schema> projected_schema_;
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task_reader.cc b/src/iceberg/inspect/metadata_scan_task_reader.cc
new file mode 100644
index 000000000..1ce5b3c1a
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task_reader.cc
@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+
+#include "iceberg/inspect/metadata_scan_task_reader.h"
+
+#include <memory>
+#include <utility>
+
+#include "iceberg/inspect/history_table.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/snapshots_table.h"
+#include "iceberg/util/checked_cast.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+MetadataScanTaskReader::MetadataScanTaskReader(Options options)
+    : options_(std::move(options)) {}
+
+MetadataScanTaskReader::~MetadataScanTaskReader() = default;
+
+Result<std::unique_ptr<MetadataScanTaskReader>> MetadataScanTaskReader::Make(
+    Options options) {
+  ICEBERG_PRECHECK(options.batch_size > 0, "batch size must be positive");
+  return std::unique_ptr<MetadataScanTaskReader>(
+      new MetadataScanTaskReader(std::move(options)));
+}
+
+Result<ArrowArrayStream> MetadataScanTaskReader::Open(
+    const MetadataScanTask& task) const {
+  switch (task.task_kind()) {
+    case MetadataScanTask::TaskKind::kStatic: {
+      const auto& static_task = internal::checked_cast<const StaticMetadataTask&>(task);
+      switch (static_task.table_kind()) {
+        case MetadataTableKind::kSnapshots:
+          return SnapshotsTable::ReadRows(static_task.metadata(),
+                                          static_task.projected_schema(),
+                                          options_.batch_size);
+        case MetadataTableKind::kHistory:
+          return HistoryTable::ReadRows(static_task.metadata(),
+                                        static_task.projected_schema(),
+                                        options_.batch_size);
+        default:
+          return NotImplemented("Static metadata table kind is not implemented");
+      }
+    }
+  }
+  return NotImplemented("Metadata scan task kind is not implemented");
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_scan_task_reader.h b/src/iceberg/inspect/metadata_scan_task_reader.h
new file mode 100644
index 000000000..8003276ba
--- /dev/null
+++ b/src/iceberg/inspect/metadata_scan_task_reader.h
@@ -0,0 +1,61 @@
+/*
+ * 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
+
+#include <cstdint>
+#include <memory>
+
+#include "iceberg/arrow_c_data.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief Opens metadata scan tasks as Arrow streams.
+class ICEBERG_EXPORT MetadataScanTaskReader {
+ public:
+  /// \brief Options shared by all tasks opened by this reader.
+  struct Options {
+    /// Positive maximum number of rows per Arrow batch.
+    int64_t batch_size = 1024;
+  };
+
+  /// \brief Create a reusable task reader from shared read options.
+  static Result<std::unique_ptr<MetadataScanTaskReader>> Make(Options options);
+
+  ~MetadataScanTaskReader();
+
+  /// \brief Open a task and return an Arrow C stream for its projected rows.
+  ///
+  /// Batches are built on demand. The stream retains its input data and can
+  /// outlive this reader and the task; each call starts an independent read.
+  Result<ArrowArrayStream> Open(const MetadataScanTask& task) const;
+
+  MetadataScanTaskReader(const MetadataScanTaskReader&) = delete;
+  MetadataScanTaskReader& operator=(const MetadataScanTaskReader&) = delete;
+
+ private:
+  explicit MetadataScanTaskReader(Options options);
+
+  const Options options_;
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_table_columns_internal.h b/src/iceberg/inspect/metadata_scan_util_internal.h
similarity index 56%
rename from src/iceberg/inspect/metadata_table_columns_internal.h
rename to src/iceberg/inspect/metadata_scan_util_internal.h
index 596ffa2f6..5cd2058f9 100644
--- a/src/iceberg/inspect/metadata_table_columns_internal.h
+++ b/src/iceberg/inspect/metadata_scan_util_internal.h
@@ -19,20 +19,25 @@
 
 #pragma once
 
-/// \file iceberg/inspect/metadata_table_columns_internal.h
-/// \brief Build only the selected columns of metadata table rows.
+/// \file iceberg/inspect/metadata_scan_util_internal.h
+/// \brief Read in-memory metadata as Arrow batches produced on demand.
 
 #include <algorithm>
-#include <cstddef>
 #include <cstdint>
+#include <memory>
+#include <optional>
 #include <span>
+#include <type_traits>
+#include <utility>
 #include <vector>
 
 #include "iceberg/arrow_c_data.h"
+#include "iceberg/arrow_c_data_util_internal.h"
 #include "iceberg/arrow_row_builder_internal.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
+#include "iceberg/schema_internal.h"
 #include "iceberg/type.h"
 #include "iceberg/util/macros.h"
 
@@ -100,4 +105,80 @@ Status AppendRow(ArrowRowBuilder& builder,
   return builder.FinishRow();
 }
 
+/// \brief Build batches lazily, retaining the row producer and its input owners.
+template <typename Row, typename GetRow>
+class MetadataBatchSource {
+ public:
+  MetadataBatchSource(std::shared_ptr<iceberg::Schema> schema,
+                      std::vector<AppendColumnFn<Row>> append_fns, size_t num_rows,
+                      int64_t batch_size, GetRow get_row)
+      : schema_(std::move(schema)),
+        append_fns_(std::move(append_fns)),
+        num_rows_(num_rows),
+        batch_size_(static_cast<size_t>(batch_size)),
+        get_row_(std::move(get_row)) {}
+
+  Status Close() { return {}; }
+
+  Result<ArrowSchema> Schema() {
+    ArrowSchema schema{};
+    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*schema_, &schema));
+    return schema;
+  }
+
+  Result<std::optional<ArrowArray>> Next() {
+    if (error_) {
+      return unexpected(*error_);
+    }
+    auto result = NextBatch();
+    if (!result) {
+      error_ = result.error();
+    }
+    return result;
+  }
+
+ private:
+  Result<std::optional<ArrowArray>> NextBatch() {
+    if (index_ == num_rows_) {
+      return std::nullopt;
+    }
+    const size_t end = index_ + std::min(batch_size_, num_rows_ - index_);
+    ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(*schema_));
+    while (index_ < end) {
+      ICEBERG_RETURN_UNEXPECTED(AppendRow<Row>(builder, append_fns_, get_row_(index_)));
+      ++index_;
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto batch, std::move(builder).Finish());
+    return batch;
+  }
+
+  const std::shared_ptr<iceberg::Schema> schema_;
+  const std::vector<AppendColumnFn<Row>> append_fns_;
+  const size_t num_rows_;
+  const size_t batch_size_;
+  GetRow get_row_;
+  size_t index_ = 0;
+  std::optional<Error> error_;
+};
+
+/// \brief Read only projected columns, constructing at most batch_size rows per batch.
+///
+/// get_row must retain shared ownership of its inputs. Each stream has independent
+/// read state, and rows are accessed only when a batch is requested.
+template <typename Row, typename GetRow>
+Result<ArrowArrayStream> MakeMetadataBatchStream(
+    const std::shared_ptr<Schema>& canonical_schema,
+    const std::shared_ptr<Schema>& projected_schema,
+    std::span<const MetadataTableColumn<Row>> columns, size_t num_rows,
+    int64_t batch_size, GetRow&& get_row) {
+  ICEBERG_PRECHECK(batch_size > 0, "Batch size must be positive");
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto append_fns,
+      SelectAppendFns<Row>(*canonical_schema, *projected_schema, columns));
+  using Source = MetadataBatchSource<Row, std::decay_t<GetRow>>;
+  return MakeArrowArrayStream(
+      std::make_unique<Source>(projected_schema, std::move(append_fns), num_rows,
+                               batch_size, std::forward<GetRow>(get_row)));
+}
+
 }  // namespace iceberg::internal
diff --git a/src/iceberg/inspect/metadata_table.cc b/src/iceberg/inspect/metadata_table.cc
index 3e21688ff..6bd17644f 100644
--- a/src/iceberg/inspect/metadata_table.cc
+++ b/src/iceberg/inspect/metadata_table.cc
@@ -21,46 +21,37 @@
 
 #include <memory>
 #include <utility>
-#include <variant>
 
+#include "iceberg/inspect/history_table.h"
 #include "iceberg/inspect/metadata_scan.h"
+#include "iceberg/inspect/snapshots_table.h"
 
 namespace iceberg {
 
-MetadataTable::MetadataTable(std::shared_ptr<Table> source_table)
+MetadataTable::MetadataTable(std::shared_ptr<const Table> source_table)
     : source_table_(std::move(source_table)) {}
 
 MetadataTable::~MetadataTable() = default;
 
-bool MetadataTable::supports_time_travel() const noexcept { return false; }
+Result<std::unique_ptr<MetadataTable>> MetadataTable::Make(
+    std::shared_ptr<const Table> table, MetadataTableKind kind) {
+  switch (kind) {
+    case MetadataTableKind::kSnapshots:
+      return SnapshotsTable::Make(std::move(table));
+    case MetadataTableKind::kHistory:
+      return HistoryTable::Make(std::move(table));
+    default:
+      break;
+  }
+  return NotImplemented("Metadata table kind is not implemented");
+}
 
-const std::shared_ptr<Table>& MetadataTable::source_table() const {
+const std::shared_ptr<const Table>& MetadataTable::source_table() const {
   return source_table_;
 }
 
 Result<std::unique_ptr<MetadataScanBuilder>> MetadataTable::NewScan() const {
-  return std::unique_ptr<MetadataScanBuilder>(new MetadataScanBuilder(*this));
-}
-
-Result<ArrowArrayStream> MetadataTable::ScanSnapshot(
-    const SnapshotSelection& snapshot_selection, const Schema& projected_schema) const {
-  if (!std::holds_alternative<std::monostate>(snapshot_selection.snapshot) ||
-      !snapshot_selection.ref_name.empty()) {
-    return NotSupported("Metadata table does not support time travel");
-  }
-  return ScanProjection(projected_schema);
-}
-
-TimeTravelMetadataTable::TimeTravelMetadataTable(std::shared_ptr<Table> source_table)
-    : MetadataTable(std::move(source_table)) {}
-
-TimeTravelMetadataTable::~TimeTravelMetadataTable() = default;
-
-bool TimeTravelMetadataTable::supports_time_travel() const noexcept { return true; }
-
-Result<ArrowArrayStream> TimeTravelMetadataTable::ScanProjection(
-    const Schema& projected_schema) const {
-  return ScanSnapshot(SnapshotSelection{}, projected_schema);
+  return MetadataScanBuilder::Make(*this);
 }
 
 }  // namespace iceberg
diff --git a/src/iceberg/inspect/metadata_table.h b/src/iceberg/inspect/metadata_table.h
index 7717ce58a..83f44f5c1 100644
--- a/src/iceberg/inspect/metadata_table.h
+++ b/src/iceberg/inspect/metadata_table.h
@@ -20,40 +20,56 @@
 #pragma once
 
 /// \file iceberg/inspect/metadata_table.h
-/// \brief Base APIs for inspecting Iceberg metadata tables.
+/// \brief Metadata table factories, schemas, and scan entry points.
 
 #include <concepts>
+#include <cstdint>
 #include <memory>
 #include <utility>
 
-#include "iceberg/arrow_c_data.h"
 #include "iceberg/iceberg_export.h"
-#include "iceberg/inspect/metadata_scan.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
 
 namespace iceberg {
+namespace internal {
+struct MetadataScanContext;
+}  // namespace internal
+
+/// \brief Iceberg metadata tables exposed by Table::Inspect().
+enum class MetadataTableKind : uint8_t {
+  kEntries,
+  kFiles,
+  kDataFiles,
+  kDeleteFiles,
+  kHistory,
+  kMetadataLogEntries,
+  kSnapshots,
+  kRefs,
+  kManifests,
+  kPartitions,
+  kPositionDeletes,
+  kAllEntries,
+  kAllFiles,
+  kAllDataFiles,
+  kAllDeleteFiles,
+  kAllManifests,
+};
 
-/// \brief Base interface for an Iceberg metadata table.
+/// \brief A metadata table backed by a source Iceberg table.
 class ICEBERG_EXPORT MetadataTable {
  public:
-  /// \brief Supported metadata table kinds.
-  enum class Kind {
-    kSnapshots,
-    kHistory,
-  };
+  using Kind = MetadataTableKind;
 
-  /// \brief Maximum number of rows emitted in each Arrow batch.
-  static constexpr int64_t kBatchSize = 1024;
+  /// \brief Create a metadata table; unsupported kinds return NotImplemented.
+  static Result<std::unique_ptr<MetadataTable>> Make(std::shared_ptr<const Table> table,
+                                                     Kind kind);
 
-  /// \brief Create a metadata table of the requested concrete type.
-  ///
-  /// \tparam MetadataTableType Concrete class derived from MetadataTable.
-  /// \param table Source table whose metadata will be exposed.
-  /// \return The constructed metadata table, or an error.
+  /// \brief Create a specific metadata table from its concrete class.
   template <typename MetadataTableType>
     requires std::derived_from<MetadataTableType, MetadataTable>
-  static Result<std::unique_ptr<MetadataTableType>> Make(std::shared_ptr<Table> table) {
+  static Result<std::unique_ptr<MetadataTableType>> Make(
+      std::shared_ptr<const Table> table) {
     return MetadataTableType::Make(std::move(table));
   }
 
@@ -66,70 +82,25 @@ class ICEBERG_EXPORT MetadataTable {
   virtual const std::shared_ptr<Schema>& schema() const = 0;
 
   /// \brief Return the source table whose metadata is exposed.
-  const std::shared_ptr<Table>& source_table() const;
+  const std::shared_ptr<const Table>& source_table() const;
 
-  /// \brief Return whether this metadata table supports time travel.
-  virtual bool supports_time_travel() const noexcept;
-
-  /// \brief Create a builder for configuring a metadata table scan.
+  /// \brief Capture the source metadata and create a scan builder.
   ///
-  /// A builder with no options scans all columns of the current snapshot on the main
-  /// branch. This table must outlive the builder.
+  /// This metadata table must outlive the builder. Built scans retain their own
+  /// inputs and can outlive both tables.
   virtual Result<std::unique_ptr<MetadataScanBuilder>> NewScan() const;
 
  protected:
-  explicit MetadataTable(std::shared_ptr<Table> source_table);
-
-  /// \brief Implement a scan that returns the columns of a projected schema.
-  ///
-  /// \param projected_schema Projection of schema() with the columns to return. It
-  /// is only valid during this call.
-  virtual Result<ArrowArrayStream> ScanProjection(
-      const Schema& projected_schema) const = 0;
+  explicit MetadataTable(std::shared_ptr<const Table> source_table);
 
-  /// \brief Implement a scan for the requested snapshot selection.
-  ///
-  /// The default implementation scans the current snapshot on the main branch with
-  /// ScanProjection() and returns NotSupported for any other selection.
-  /// TimeTravelMetadataTable requires an implementation.
-  ///
-  /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection.
-  /// \param projected_schema Projection of schema() with the columns to return. It
-  /// is only valid during this call.
-  virtual Result<ArrowArrayStream> ScanSnapshot(
-      const SnapshotSelection& snapshot_selection, const Schema& projected_schema) const;
+  /// \brief Plan a scan using the captured metadata, projection, and snapshot selection.
+  virtual Result<std::unique_ptr<MetadataScan>> MakeScan(
+      internal::MetadataScanContext context) const = 0;
 
  private:
   friend class MetadataScanBuilder;
 
-  std::shared_ptr<Table> source_table_;
-};
-
-/// \brief Base interface for metadata tables that support time travel.
-///
-/// Scans of these tables accept MetadataScanBuilder::UseSnapshot(), UseRef(), and
-/// AsOfTime(), which MetadataScanBuilder::Build() rejects for other tables.
-class ICEBERG_EXPORT TimeTravelMetadataTable : public MetadataTable {
- public:
-  ~TimeTravelMetadataTable() override;
-
-  /// \brief Return true because this interface supports time travel.
-  bool supports_time_travel() const noexcept final;
-
- protected:
-  explicit TimeTravelMetadataTable(std::shared_ptr<Table> source_table);
-
-  /// \brief Scan the current snapshot on the main branch.
-  Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema) const final;
-
-  /// \brief Implement a scan for the requested snapshot selection.
-  ///
-  /// \param snapshot_selection Snapshot ID, timestamp, and optional ref selection.
-  /// \param projected_schema Projection of schema() with the columns to return. It
-  /// is only valid during this call.
-  Result<ArrowArrayStream> ScanSnapshot(const SnapshotSelection& snapshot_selection,
-                                        const Schema& projected_schema) const override =
-      0;
+  const std::shared_ptr<const Table> source_table_;
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/inspect/snapshots_table.cc b/src/iceberg/inspect/snapshots_table.cc
index 0dede12c3..7eb204b43 100644
--- a/src/iceberg/inspect/snapshots_table.cc
+++ b/src/iceberg/inspect/snapshots_table.cc
@@ -20,22 +20,15 @@
 #include "iceberg/inspect/snapshots_table.h"
 
 #include <array>
-#include <chrono>
-#include <cstddef>
 #include <memory>
-#include <optional>
 #include <utility>
 #include <vector>
 
-#include <nanoarrow/nanoarrow.h>
-
-#include "iceberg/arrow_c_data_util_internal.h"
 #include "iceberg/arrow_row_builder_internal.h"
-#include "iceberg/inspect/metadata_table_columns_internal.h"
-#include "iceberg/nanoarrow_status_internal.h"
+#include "iceberg/inspect/metadata_scan_task.h"
+#include "iceberg/inspect/metadata_scan_util_internal.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
-#include "iceberg/schema_internal.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table.h"
 #include "iceberg/table_metadata.h"
@@ -45,165 +38,127 @@
 namespace iceberg {
 namespace {
 
-Status AppendCommittedAt(ArrowArray* column, const Snapshot& snapshot) {
-  return AppendInt(column, std::chrono::duration_cast<std::chrono::microseconds>(
-                               snapshot.timestamp_ms.time_since_epoch())
-                               .count());
-}
-
-Status AppendSnapshotId(ArrowArray* column, const Snapshot& snapshot) {
-  return AppendInt(column, snapshot.snapshot_id);
-}
-
-Status AppendParentId(ArrowArray* column, const Snapshot& snapshot) {
-  if (snapshot.parent_snapshot_id.has_value()) {
-    return AppendInt(column, *snapshot.parent_snapshot_id);
-  }
-  return AppendNull(column);
-}
-
-Status AppendOperation(ArrowArray* column, const Snapshot& snapshot) {
-  auto operation = snapshot.Operation();
-  if (operation.has_value()) {
-    return AppendString(column, *operation);
-  }
-  return AppendNull(column);
-}
-
-Status AppendManifestList(ArrowArray* column, const Snapshot& snapshot) {
-  return AppendString(column, snapshot.manifest_list);
-}
-
-Status AppendSummary(ArrowArray* column, const Snapshot& snapshot) {
-  if (snapshot.summary.empty()) {
-    return AppendNull(column);
-  }
-  if (snapshot.summary.contains(SnapshotSummaryFields::kOperation)) {
-    auto summary = snapshot.summary;
-    summary.erase(SnapshotSummaryFields::kOperation);
-    return AppendStringMap(column, summary);
-  }
-  return AppendStringMap(column, snapshot.summary);
-}
-
-constexpr std::array<internal::MetadataTableColumn<Snapshot>, 6> kSnapshotColumns{{
-    {.field_id = 1, .append = AppendCommittedAt},
-    {.field_id = 2, .append = AppendSnapshotId},
-    {.field_id = 3, .append = AppendParentId},
-    {.field_id = 4, .append = AppendOperation},
-    {.field_id = 5, .append = AppendManifestList},
-    {.field_id = 6, .append = AppendSummary},
-}};
-
-class SnapshotsTableStream {
- public:
-  static Result<std::unique_ptr<SnapshotsTableStream>> Make(
-      std::shared_ptr<const TableMetadata> metadata,
-      const iceberg::Schema& projected_schema,
-      std::vector<internal::AppendColumnFn<Snapshot>> append_fns) {
-    ArrowSchema arrow_schema{};
-    ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(projected_schema, &arrow_schema));
-    return std::unique_ptr<SnapshotsTableStream>(new SnapshotsTableStream(
-        std::move(metadata), std::move(append_fns), std::move(arrow_schema)));
-  }
-
-  ~SnapshotsTableStream() = default;
-
-  Status Close() {
-    metadata_.reset();
-    if (arrow_schema_.release != nullptr) {
-      ArrowSchemaRelease(&arrow_schema_);
-    }
-    return {};
-  }
-
-  Result<std::optional<ArrowArray>> Next() {
-    if (metadata_ == nullptr) [[unlikely]] {
-      return InvalidArgument("Cannot read from a closed snapshots table stream");
-    }
-    const auto& snapshots = metadata_->snapshots;
-    if (next_snapshot_ == snapshots.size()) {
-      return std::nullopt;
-    }
-
-    ICEBERG_ASSIGN_OR_RAISE(auto builder, ArrowRowBuilder::Make(&arrow_schema_));
-    while (next_snapshot_ < snapshots.size() &&
-           builder.num_rows() < MetadataTable::kBatchSize) {
-      const auto& snapshot = snapshots[next_snapshot_++];
-      if (snapshot == nullptr) [[unlikely]] {
-        continue;
-      }
-      ICEBERG_RETURN_UNEXPECTED(
-          internal::AppendRow<Snapshot>(builder, append_fns_, *snapshot));
-    }
-    if (builder.num_rows() == 0) {
-      return std::nullopt;
-    }
-
-    ICEBERG_ASSIGN_OR_RAISE(auto array, std::move(builder).Finish());
-    return array;
-  }
-
-  Result<ArrowSchema> Schema() {
-    if (arrow_schema_.release == nullptr) [[unlikely]] {
-      return InvalidArgument("Cannot read schema from a closed snapshots table stream");
-    }
-    ArrowSchema schema_copy{};
-    ICEBERG_NANOARROW_RETURN_UNEXPECTED(
-        ArrowSchemaDeepCopy(&arrow_schema_, &schema_copy));
-    return schema_copy;
-  }
-
- private:
-  SnapshotsTableStream(std::shared_ptr<const TableMetadata> metadata,
-                       std::vector<internal::AppendColumnFn<Snapshot>> append_fns,
-                       ArrowSchema arrow_schema)
-      : metadata_(std::move(metadata)),
-        append_fns_(std::move(append_fns)),
-        arrow_schema_(std::move(arrow_schema)) {}
-
-  std::shared_ptr<const TableMetadata> metadata_;
-  std::vector<internal::AppendColumnFn<Snapshot>> append_fns_;
-  ArrowSchema arrow_schema_{};
-  size_t next_snapshot_ = 0;
-};
+constexpr std::array<internal::MetadataTableColumn<Snapshot>, 6> kSnapshotColumns = {
+    {{.field_id = 1,
+      .append =
+          [](ArrowArray* column, const Snapshot& row) {
+            return AppendInt(column, UnixMsFromTimePointMs(row.timestamp_ms) * 1000);
+          }},
+     {.field_id = 2,
+      .append = [](ArrowArray* column,
+                   const Snapshot& row) { return AppendInt(column, row.snapshot_id); }},
+     {.field_id = 3,
+      .append =
+          [](ArrowArray* column, const Snapshot& row) {
+            return row.parent_snapshot_id ? AppendInt(column, *row.parent_snapshot_id)
+                                          : AppendNull(column);
+          }},
+     {.field_id = 4,
+      .append =
+          [](ArrowArray* column, const Snapshot& row) {
+            auto operation = row.Operation();
+            return operation ? AppendString(column, *operation) : AppendNull(column);
+          }},
+     {.field_id = 5,
+      .append =
+          [](ArrowArray* column, const Snapshot& row) {
+            return row.manifest_list.empty() ? AppendNull(column)
+                                             : AppendString(column, row.manifest_list);
+          }},
+     {.field_id = 6, .append = [](ArrowArray* column, const Snapshot& row) {
+        if (row.summary.empty()) {
+          return AppendNull(column);
+        }
+        auto summary = row.summary;
+        summary.erase(SnapshotSummaryFields::kOperation);
+        return AppendStringMap(column, summary);
+      }}}};
 
 }  // namespace
 
-SnapshotsTable::SnapshotsTable(std::shared_ptr<Table> table)
+// SnapshotsTable implementation
+SnapshotsTable::SnapshotsTable(std::shared_ptr<const Table> table)
     : MetadataTable(std::move(table)) {}
 
 SnapshotsTable::~SnapshotsTable() = default;
 
 const std::shared_ptr<Schema>& SnapshotsTable::schema() const {
+  return CanonicalSchema();
+}
+
+const std::shared_ptr<Schema>& SnapshotsTable::CanonicalSchema() {
   static const auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "committed_at", timestamp_tz()),
       SchemaField::MakeRequired(2, "snapshot_id", int64()),
       SchemaField::MakeOptional(3, "parent_id", int64()),
       SchemaField::MakeOptional(4, "operation", string()),
       SchemaField::MakeOptional(5, "manifest_list", string()),
-      SchemaField::MakeOptional(6, "summary",
-                                std::make_shared<iceberg::MapType>(
-                                    SchemaField::MakeRequired(7, "key", string()),
+      SchemaField::MakeOptional(
+          6, "summary",
+          std::make_shared<MapType>(SchemaField::MakeRequired(7, "key", string()),
                                     SchemaField::MakeRequired(8, "value", string())))});
   return schema;
 }
 
 Result<std::unique_ptr<SnapshotsTable>> SnapshotsTable::Make(
-    std::shared_ptr<Table> table) {
+    std::shared_ptr<const Table> table) {
   ICEBERG_PRECHECK(table != nullptr, "Table cannot be null");
   return std::unique_ptr<SnapshotsTable>(new SnapshotsTable(std::move(table)));
 }
 
-Result<ArrowArrayStream> SnapshotsTable::ScanProjection(
-    const Schema& projected_schema) const {
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto append_fns,
-      internal::SelectAppendFns<Snapshot>(*schema(), projected_schema, kSnapshotColumns));
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto stream, SnapshotsTableStream::Make(source_table()->metadata(),
-                                              projected_schema, std::move(append_fns)));
-  return MakeArrowArrayStream(std::move(stream));
+Result<std::unique_ptr<MetadataScan>> SnapshotsTable::MakeScan(
+    internal::MetadataScanContext context) const {
+  return SnapshotsScan::Make(std::move(context));
+}
+
+Result<ArrowArrayStream> SnapshotsTable::ReadRows(
+    std::shared_ptr<const TableMetadata> metadata,
+    const std::shared_ptr<Schema>& projected_schema, int64_t batch_size) {
+  std::vector<size_t> snapshot_indices;
+  for (size_t i = 0; i < metadata->snapshots.size(); ++i) {
+    if (metadata->snapshots[i]) {
+      snapshot_indices.push_back(i);
+    }
+  }
+  const size_t num_rows = snapshot_indices.size();
+  return internal::MakeMetadataBatchStream<Snapshot>(
+      CanonicalSchema(), projected_schema, kSnapshotColumns, num_rows, batch_size,
+      [metadata = std::move(metadata),
+       snapshot_indices = std::move(snapshot_indices)](size_t index) -> const Snapshot& {
+        return *metadata->snapshots[snapshot_indices[index]];
+      });
+}
+
+// SnapshotsScan implementation
+SnapshotsScan::SnapshotsScan(internal::MetadataScanContext context,
+                             std::shared_ptr<MetadataScanTask> task)
+    : MetadataScan(std::move(context)), task_(std::move(task)) {}
+
+SnapshotsScan::~SnapshotsScan() = default;
+
+Result<std::unique_ptr<SnapshotsScan>> SnapshotsScan::Make(
+    internal::MetadataScanContext context) {
+  ICEBERG_PRECHECK(context.kind == MetadataTableKind::kSnapshots,
+                   "Unexpected metadata table kind for snapshots scan");
+  if (context.snapshot_id.has_value() || !context.ref_name.empty() ||
+      context.as_of_time_in_millis.has_value()) {
+    return NotSupported("Metadata table does not support time travel");
+  }
+  ICEBERG_PRECHECK(context.table_metadata != nullptr, "Table metadata cannot be null");
+  ICEBERG_PRECHECK(context.schema != nullptr, "Metadata table schema cannot be null");
+  ICEBERG_PRECHECK(context.projected_schema != nullptr,
+                   "Projected schema cannot be null");
+  ICEBERG_RETURN_UNEXPECTED(internal::SelectAppendFns<Snapshot>(
+      *SnapshotsTable::CanonicalSchema(), *context.projected_schema, kSnapshotColumns));
+  ICEBERG_ASSIGN_OR_RAISE(auto task,
+                          StaticMetadataTask::Make(context.kind, context.table_metadata,
+                                                   context.projected_schema));
+  return std::unique_ptr<SnapshotsScan>(
+      new SnapshotsScan(std::move(context), std::move(task)));
+}
+
+Result<MetadataScanTaskStreamPtr> SnapshotsScan::PlanTasksStream() const {
+  return MetadataScanTaskStream::FromVector({task_});
 }
 
 }  // namespace iceberg
diff --git a/src/iceberg/inspect/snapshots_table.h b/src/iceberg/inspect/snapshots_table.h
index ae1434a5e..4bfaa554f 100644
--- a/src/iceberg/inspect/snapshots_table.h
+++ b/src/iceberg/inspect/snapshots_table.h
@@ -22,9 +22,12 @@
 /// \file iceberg/inspect/snapshots_table.h
 /// \brief Define the snapshots metadata table.
 
+#include <cstdint>
 #include <memory>
 
+#include "iceberg/arrow_c_data.h"
 #include "iceberg/iceberg_export.h"
+#include "iceberg/inspect/metadata_scan.h"
 #include "iceberg/inspect/metadata_table.h"
 #include "iceberg/result.h"
 #include "iceberg/type_fwd.h"
@@ -34,22 +37,49 @@ namespace iceberg {
 /// \brief Snapshots metadata table.
 class ICEBERG_EXPORT SnapshotsTable : public MetadataTable {
  public:
-  static Result<std::unique_ptr<SnapshotsTable>> Make(std::shared_ptr<Table> table);
+  static Result<std::unique_ptr<SnapshotsTable>> Make(std::shared_ptr<const Table> table);
+
+  /// \brief Return the canonical schema of the snapshots metadata table.
+  static const std::shared_ptr<Schema>& CanonicalSchema();
 
   ~SnapshotsTable() override;
 
-  Kind kind() const noexcept override { return Kind::kSnapshots; }
+  MetadataTableKind kind() const noexcept override {
+    return MetadataTableKind::kSnapshots;
+  }
 
   const std::shared_ptr<Schema>& schema() const override;
 
  protected:
-  /// \brief Scan all snapshots as rows.
-  ///
-  /// The snapshots table always returns every known snapshot.
-  Result<ArrowArrayStream> ScanProjection(const Schema& projected_schema) const override;
+  Result<std::unique_ptr<MetadataScan>> MakeScan(
+      internal::MetadataScanContext context) const override;
 
  private:
-  explicit SnapshotsTable(std::shared_ptr<Table> table);
+  friend class MetadataScanTaskReader;
+
+  static Result<ArrowArrayStream> ReadRows(
+      std::shared_ptr<const TableMetadata> metadata,
+      const std::shared_ptr<Schema>& projected_schema, int64_t batch_size);
+
+  explicit SnapshotsTable(std::shared_ptr<const Table> table);
+};
+
+/// \brief Scan the snapshots metadata captured from a table.
+class ICEBERG_EXPORT SnapshotsScan final : public MetadataScan {
+ public:
+  /// \brief Create a scan from captured metadata table settings.
+  static Result<std::unique_ptr<SnapshotsScan>> Make(
+      internal::MetadataScanContext context);
+
+  ~SnapshotsScan() override;
+
+  Result<MetadataScanTaskStreamPtr> PlanTasksStream() const override;
+
+ private:
+  SnapshotsScan(internal::MetadataScanContext context,
+                std::shared_ptr<MetadataScanTask> task);
+
+  const std::shared_ptr<MetadataScanTask> task_;
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/table.cc b/src/iceberg/table.cc
index e56746e3b..0fa6bf405 100644
--- a/src/iceberg/table.cc
+++ b/src/iceberg/table.cc
@@ -25,6 +25,7 @@
 #include "iceberg/data_table_scan.h"
 #include "iceberg/incremental_append_scan.h"
 #include "iceberg/incremental_changelog_scan.h"
+#include "iceberg/inspect/metadata_table.h"
 #include "iceberg/labels.h"
 #include "iceberg/location_provider.h"
 #include "iceberg/partition_spec.h"
@@ -294,6 +295,10 @@ Result<std::shared_ptr<SnapshotManager>> Table::NewSnapshotManager() {
   return SnapshotManager::Make(shared_from_this());
 }
 
+Result<std::unique_ptr<MetadataTable>> Table::Inspect(MetadataTableKind kind) const {
+  return MetadataTable::Make(shared_from_this(), kind);
+}
+
 Result<std::shared_ptr<StagedTable>> StagedTable::Make(
     TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
     std::string metadata_location, std::shared_ptr<FileIO> io,
diff --git a/src/iceberg/table.h b/src/iceberg/table.h
index 97b1a43a9..50e6aa8fd 100644
--- a/src/iceberg/table.h
+++ b/src/iceberg/table.h
@@ -221,6 +221,9 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
   virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
 
+  /// \brief Construct a metadata table inspector for this table.
+  Result<std::unique_ptr<MetadataTable>> Inspect(MetadataTableKind kind) const;
+
  protected:
   Table(TableIdentifier identifier, std::shared_ptr<TableMetadata> metadata,
         std::string metadata_location, std::shared_ptr<FileIO> io,
diff --git a/src/iceberg/test/arrow_test_util.h b/src/iceberg/test/arrow_test_util.h
new file mode 100644
index 000000000..3f3d58c37
--- /dev/null
+++ b/src/iceberg/test/arrow_test_util.h
@@ -0,0 +1,87 @@
+/*
+ * 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
+
+#include <memory>
+#include <string_view>
+#include <utility>
+
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/json/from_string.h>
+#include <arrow/record_batch.h>
+#include <arrow/table.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/arrow_c_data.h"
+#include "iceberg/result.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+/// \brief Consume an Arrow C stream into a table, retaining the stream's schema.
+inline Result<std::shared_ptr<::arrow::Table>> ReadArrowTable(ArrowArrayStream&& stream) {
+  auto reader = ::arrow::ImportRecordBatchReader(&stream);
+  if (!reader.ok()) {
+    if (stream.release != nullptr) {
+      stream.release(&stream);
+    }
+    return InvalidArrowData(reader.status().ToString());
+  }
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto table, reader.ValueUnsafe()->ToTable());
+  return table;
+}
+
+/// \brief Build an expected Arrow table from an Iceberg schema and JSON rows.
+inline Result<std::shared_ptr<::arrow::Table>> TableFromJson(const Schema& schema,
+                                                             std::string_view rows) {
+  ArrowSchema c_schema{};
+  ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(schema, &c_schema));
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(auto arrow_schema, ::arrow::ImportSchema(&c_schema));
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(
+      auto array,
+      ::arrow::json::ArrayFromJSONString(::arrow::struct_(arrow_schema->fields()), rows));
+  const auto& struct_array = static_cast<const ::arrow::StructArray&>(*array);
+  auto batch =
+      ::arrow::RecordBatch::Make(arrow_schema, array->length(), struct_array.fields());
+  ICEBERG_ARROW_ASSIGN_OR_RETURN(
+      auto table, ::arrow::Table::FromRecordBatches(arrow_schema, {std::move(batch)}));
+  return table;
+}
+
+/// \brief Compare schemas and values independently of Arrow chunk boundaries.
+inline ::testing::AssertionResult TablesEqual(const ::arrow::Table& expected,
+                                              const ::arrow::Table& actual) {
+  const auto status = actual.ValidateFull();
+  if (!status.ok()) {
+    return ::testing::AssertionFailure() << status.ToString();
+  }
+  if (expected.Equals(actual, /*check_metadata=*/true)) {
+    return ::testing::AssertionSuccess();
+  }
+  return ::testing::AssertionFailure() << "Expected:\n"
+                                       << expected.ToString() << "\nActual:\n"
+                                       << actual.ToString();
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/test/history_table_test.cc b/src/iceberg/test/history_table_test.cc
index 8da311c02..810efa1ff 100644
--- a/src/iceberg/test/history_table_test.cc
+++ b/src/iceberg/test/history_table_test.cc
@@ -17,39 +17,142 @@
  * under the License.
  */
 
-/// \file history_table_test.cc
-/// Unit tests for HistoryTable.
-
 #include "iceberg/inspect/history_table.h"
 
-#include <gmock/gmock.h>
+#include <cstdint>
+#include <memory>
+#include <utility>
+#include <vector>
+
 #include <gtest/gtest.h>
 
-#include "iceberg/inspect/metadata_table.h"
+#include "iceberg/inspect/metadata_scan_task_reader.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/test/arrow_test_util.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/metadata_table_test_base.h"
 #include "iceberg/type.h"
+#include "iceberg/util/timepoint.h"
 
 namespace iceberg {
-namespace {
-
-std::shared_ptr<Schema> MakeHistorySchema() {
-  return std::make_shared<Schema>(std::vector<SchemaField>{
-      SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
-      SchemaField::MakeRequired(2, "snapshot_id", int64()),
-      SchemaField::MakeOptional(3, "parent_id", int64()),
-      SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+
+class HistoryTableTest : public MetadataTableTestBase {
+ protected:
+  void SetUp() override {
+    MetadataTableTestBase::SetUp();
+    metadata_->snapshots = {
+        std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1}),
+        std::make_shared<Snapshot>(Snapshot{.snapshot_id = 2, .parent_snapshot_id = 1}),
+        std::make_shared<Snapshot>(Snapshot{.snapshot_id = 3, .parent_snapshot_id = 1})};
+    metadata_->current_snapshot_id = 3;
+    // Roll back from 2 to 1, include an expired snapshot (4), then move to 3.
+    metadata_->snapshot_log = {
+        {.timestamp_ms = TimePointMsFromUnixMs(1000), .snapshot_id = 1},
+        {.timestamp_ms = TimePointMsFromUnixMs(2000), .snapshot_id = 2},
+        {.timestamp_ms = TimePointMsFromUnixMs(2500), .snapshot_id = 1},
+        {.timestamp_ms = TimePointMsFromUnixMs(2700), .snapshot_id = 4},
+        {.timestamp_ms = TimePointMsFromUnixMs(3000), .snapshot_id = 3}};
+    ICEBERG_UNWRAP_OR_FAIL(metadata_table_, table_->Inspect(MetadataTableKind::kHistory));
+  }
+
+  static std::shared_ptr<Schema> ExpectedSchema() {
+    return std::make_shared<Schema>(std::vector<SchemaField>{
+        SchemaField::MakeRequired(1, "made_current_at", timestamp_tz()),
+        SchemaField::MakeRequired(2, "snapshot_id", int64()),
+        SchemaField::MakeOptional(3, "parent_id", int64()),
+        SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+  }
+
+  std::unique_ptr<MetadataTable> metadata_table_;
+};
+
+TEST_F(HistoryTableTest, CanonicalSchema) {
+  const auto expected = ExpectedSchema();
+  EXPECT_EQ(*HistoryTable::CanonicalSchema(), *expected);
+  EXPECT_EQ(*metadata_table_->schema(), *expected);
 }
 
-}  // namespace
+TEST_F(HistoryTableTest, Read) {
+  ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table_->NewScan());
+  ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanTasks());
+  ASSERT_EQ(tasks.size(), 1);
+  ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(*ExpectedSchema(), R"([
+    {"made_current_at":1000000,"snapshot_id":1,"parent_id":null,"is_current_ancestor":true},
+    {"made_current_at":2000000,"snapshot_id":2,"parent_id":1,"is_current_ancestor":false},
+    {"made_current_at":2500000,"snapshot_id":1,"parent_id":null,"is_current_ancestor":true},
+    {"made_current_at":2700000,"snapshot_id":4,"parent_id":null,"is_current_ancestor":false},
+    {"made_current_at":3000000,"snapshot_id":3,"parent_id":1,"is_current_ancestor":true}
+  ])"));
+  for (int64_t batch_size : {1, 2}) {
+    SCOPED_TRACE(batch_size);
+    ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                           MetadataScanTaskReader::Make({.batch_size = batch_size}));
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(*tasks.front()));
+    ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+    EXPECT_TRUE(TablesEqual(*expected, *actual));
+  }
+}
+
+TEST_F(HistoryTableTest, Project) {
+  ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table_->NewScan());
+  builder->Select({"snapshot_id", "is_current_ancestor"});
+  ICEBERG_UNWRAP_OR_FAIL(auto selected_scan, builder->Build());
+
+  auto projection = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeOptional(3, "parent_id", int64()),
+                               SchemaField::MakeRequired(2, "snapshot_id", int64())});
+  builder->Project(projection);
+  ICEBERG_UNWRAP_OR_FAIL(auto projected_scan, builder->Build());
 
-class HistoryTableTest : public MetadataTableTestBase {};
+  Schema selection({SchemaField::MakeRequired(2, "snapshot_id", int64()),
+                    SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+  ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(selection, R"([
+    {"snapshot_id":1,"is_current_ancestor":true},
+    {"snapshot_id":2,"is_current_ancestor":false},
+    {"snapshot_id":1,"is_current_ancestor":true},
+    {"snapshot_id":4,"is_current_ancestor":false},
+    {"snapshot_id":3,"is_current_ancestor":true}
+  ])"));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, selected_scan->ToArrowArrayStream());
+  ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+  EXPECT_TRUE(TablesEqual(*expected, *actual));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto reordered, TableFromJson(*projection, R"([
+    {"parent_id":null,"snapshot_id":1},
+    {"parent_id":1,"snapshot_id":2},
+    {"parent_id":null,"snapshot_id":1},
+    {"parent_id":null,"snapshot_id":4},
+    {"parent_id":1,"snapshot_id":3}
+  ])"));
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, projected_scan->PlanTasks());
+  ASSERT_EQ(tasks.size(), 1);
+  ICEBERG_UNWRAP_OR_FAIL(auto reader, MetadataScanTaskReader::Make({.batch_size = 2}));
+  ICEBERG_UNWRAP_OR_FAIL(auto projected_stream, reader->Open(*tasks.front()));
+  ICEBERG_UNWRAP_OR_FAIL(auto projected, ReadArrowTable(std::move(projected_stream)));
+  EXPECT_TRUE(TablesEqual(*reordered, *projected));
+}
 
-TEST_F(HistoryTableTest, SchemaMatchesIcebergSchema) {
-  ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make<HistoryTable>(table_));
-  EXPECT_TRUE(*history_table->schema() == *MakeHistorySchema());
+TEST_F(HistoryTableTest, CyclicAncestryTerminates) {
+  metadata_->snapshots.front()->parent_snapshot_id = 3;
+  ICEBERG_UNWRAP_OR_FAIL(auto builder, metadata_table_->NewScan());
+  builder->Select({"snapshot_id", "is_current_ancestor"});
+  ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, scan->ToArrowArrayStream());
+  ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+  Schema schema({SchemaField::MakeRequired(2, "snapshot_id", int64()),
+                 SchemaField::MakeRequired(4, "is_current_ancestor", boolean())});
+  ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(schema, R"([
+    {"snapshot_id":1,"is_current_ancestor":true},
+    {"snapshot_id":2,"is_current_ancestor":false},
+    {"snapshot_id":1,"is_current_ancestor":true},
+    {"snapshot_id":4,"is_current_ancestor":false},
+    {"snapshot_id":3,"is_current_ancestor":true}
+  ])"));
+  EXPECT_TRUE(TablesEqual(*expected, *actual));
 }
 
 }  // namespace iceberg
diff --git a/src/iceberg/test/metadata_table_test.cc b/src/iceberg/test/metadata_table_test.cc
index 187646c9b..0f93ba068 100644
--- a/src/iceberg/test/metadata_table_test.cc
+++ b/src/iceberg/test/metadata_table_test.cc
@@ -23,7 +23,6 @@
 #include <memory>
 #include <string>
 #include <utility>
-#include <variant>
 #include <vector>
 
 #include <arrow/c/bridge.h>
@@ -34,53 +33,35 @@
 
 #include "iceberg/constants.h"
 #include "iceberg/inspect/history_table.h"
+#include "iceberg/inspect/metadata_scan.h"
+#include "iceberg/inspect/metadata_scan_task.h"
 #include "iceberg/inspect/snapshots_table.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
 #include "iceberg/table.h"
 #include "iceberg/table_identifier.h"
 #include "iceberg/table_metadata.h"
+#include "iceberg/test/arrow_test_util.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/mock_catalog.h"
 #include "iceberg/test/mock_io.h"
 #include "iceberg/type.h"
-#include "iceberg/util/timepoint.h"
 
 namespace iceberg {
 namespace {
 
-/// \brief Time-travel table that records the arguments of each scan.
-class RecordingTimeTravelTable : public TimeTravelMetadataTable {
+class PlannedMetadataScan : public MetadataScan {
  public:
-  explicit RecordingTimeTravelTable(std::shared_ptr<Table> table)
-      : TimeTravelMetadataTable(std::move(table)) {}
+  PlannedMetadataScan(internal::MetadataScanContext context,
+                      std::vector<std::shared_ptr<MetadataScanTask>> tasks)
+      : MetadataScan(std::move(context)), tasks_(std::move(tasks)) {}
 
-  Kind kind() const noexcept override { return Kind::kSnapshots; }
-
-  const std::shared_ptr<Schema>& schema() const override {
-    static const auto schema = std::make_shared<Schema>(
-        std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int64()),
-                                 SchemaField::MakeOptional(2, "name", string())});
-    return schema;
-  }
-
-  const SnapshotSelection& scanned_snapshot() const { return scanned_snapshot_; }
-  const std::vector<std::string>& scanned_columns() const { return scanned_columns_; }
-
- protected:
-  Result<ArrowArrayStream> ScanSnapshot(const SnapshotSelection& snapshot_selection,
-                                        const Schema& projected_schema) const override {
-    scanned_snapshot_ = snapshot_selection;
-    scanned_columns_.clear();
-    for (const auto& field : projected_schema.fields()) {
-      scanned_columns_.emplace_back(field.name());
-    }
-    return NotSupported("Recording table does not return rows");
+  Result<MetadataScanTaskStreamPtr> PlanTasksStream() const override {
+    return MetadataScanTaskStream::FromVector(tasks_);
   }
 
  private:
-  mutable SnapshotSelection scanned_snapshot_;
-  mutable std::vector<std::string> scanned_columns_;
+  const std::vector<std::shared_ptr<MetadataScanTask>> tasks_;
 };
 
 }  // namespace
@@ -104,6 +85,11 @@ class MetadataTableTest : public ::testing::Test {
                                                std::make_shared<MockCatalog>()));
   }
 
+  static Result<ArrowArrayStream> BuildStream(MetadataScanBuilder& builder) {
+    ICEBERG_ASSIGN_OR_RAISE(auto scan, builder.Build());
+    return scan->ToArrowArrayStream();
+  }
+
   /// \brief Import the schema of a stream and release the stream.
   static Result<std::shared_ptr<::arrow::Schema>> ReadSchema(ArrowArrayStream&& stream) {
     auto reader_result = ::arrow::ImportRecordBatchReader(&stream);
@@ -131,15 +117,15 @@ TEST_F(MetadataTableTest, ScanBuilderSelectResolvesNamesImmediately) {
 
   // Select() uses the case sensitivity in effect when it is called.
   ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table->NewScan());
-  ICEBERG_UNWRAP_OR_FAIL(auto stream,
-                         builder->Select({"SNAPSHOT_ID"}).CaseSensitive(false).Build());
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto stream, BuildStream(builder->Select({"SNAPSHOT_ID"}).CaseSensitive(false)));
   ICEBERG_UNWRAP_OR_FAIL(auto schema, ReadSchema(std::move(stream)));
   EXPECT_EQ(schema->num_fields(), 0);
 
   ICEBERG_UNWRAP_OR_FAIL(auto insensitive_builder, snapshots_table->NewScan());
   ICEBERG_UNWRAP_OR_FAIL(
       auto insensitive_stream,
-      insensitive_builder->CaseSensitive(false).Select({"SNAPSHOT_ID"}).Build());
+      BuildStream(insensitive_builder->CaseSensitive(false).Select({"SNAPSHOT_ID"})));
   ICEBERG_UNWRAP_OR_FAIL(auto insensitive_schema,
                          ReadSchema(std::move(insensitive_stream)));
   EXPECT_THAT(insensitive_schema->field_names(), ::testing::ElementsAre("snapshot_id"));
@@ -160,8 +146,8 @@ TEST_F(MetadataTableTest, ScanBuilderProject) {
 
   // Project() replaces an earlier selection.
   ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table->NewScan());
-  ICEBERG_UNWRAP_OR_FAIL(auto stream,
-                         builder->Select({"snapshot_id"}).Project(projection).Build());
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto stream, BuildStream(builder->Select({"snapshot_id"}).Project(projection)));
   ICEBERG_UNWRAP_OR_FAIL(auto schema, ReadSchema(std::move(stream)));
   EXPECT_THAT(schema->field_names(), ::testing::ElementsAre("operation", "snapshot_id"));
 
@@ -170,73 +156,30 @@ TEST_F(MetadataTableTest, ScanBuilderProject) {
               IsError(ErrorKind::kInvalidArgument));
 }
 
-TEST_F(MetadataTableTest, SupportsTimeTravel) {
-  ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
-                         MetadataTable::Make<SnapshotsTable>(table_));
-  EXPECT_FALSE(snapshots_table->supports_time_travel());
-
-  ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make<HistoryTable>(table_));
-  EXPECT_FALSE(history_table->supports_time_travel());
-
-  RecordingTimeTravelTable time_travel_table(table_);
-  EXPECT_TRUE(time_travel_table.supports_time_travel());
-}
-
-TEST_F(MetadataTableTest, TimeTravelScanDefaults) {
-  RecordingTimeTravelTable table(table_);
-  ICEBERG_UNWRAP_OR_FAIL(auto builder, table.NewScan());
-  EXPECT_THAT(builder->Build(), IsError(ErrorKind::kNotSupported));
-  EXPECT_TRUE(std::holds_alternative<std::monostate>(table.scanned_snapshot().snapshot));
-  EXPECT_TRUE(table.scanned_snapshot().ref_name.empty());
-  EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("id", "name"));
-}
-
-TEST_F(MetadataTableTest, TimeTravelScanSelectsColumnsAndSnapshot) {
-  RecordingTimeTravelTable table(table_);
-  ICEBERG_UNWRAP_OR_FAIL(auto builder, table.NewScan());
-  EXPECT_THAT(builder->Select({"name"}).UseSnapshot(1).UseRef("dev").Build(),
-              IsError(ErrorKind::kNotSupported));
-  ASSERT_TRUE(std::holds_alternative<int64_t>(table.scanned_snapshot().snapshot));
-  EXPECT_EQ(std::get<int64_t>(table.scanned_snapshot().snapshot), 1);
-  EXPECT_EQ(table.scanned_snapshot().ref_name, "dev");
-  EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("name"));
-}
-
-TEST_F(MetadataTableTest, TimeTravelScanAsOfTime) {
-  RecordingTimeTravelTable table(table_);
-  ICEBERG_UNWRAP_OR_FAIL(auto builder, table.NewScan());
-  EXPECT_THAT(builder->AsOfTime(1234567890000).Build(),
-              IsError(ErrorKind::kNotSupported));
-  ASSERT_TRUE(std::holds_alternative<TimePointMs>(table.scanned_snapshot().snapshot));
-  EXPECT_EQ(std::get<TimePointMs>(table.scanned_snapshot().snapshot),
-            TimePointMsFromUnixMs(1234567890000));
-  EXPECT_TRUE(table.scanned_snapshot().ref_name.empty());
-  EXPECT_THAT(table.scanned_columns(), ::testing::ElementsAre("id", "name"));
-}
-
 TEST_F(MetadataTableTest, ScanBuilderRejectsConflictingSnapshotOptions) {
-  RecordingTimeTravelTable table(table_);
+  ICEBERG_UNWRAP_OR_FAIL(auto table, MetadataTable::Make<SnapshotsTable>(table_));
 
-  ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, table.NewScan());
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, table->NewScan());
   auto snapshot_result = snapshot_builder->UseSnapshot(1).AsOfTime(2).Build();
   EXPECT_THAT(snapshot_result, IsError(ErrorKind::kValidationFailed));
   EXPECT_THAT(snapshot_result, HasErrorMessage("Cannot override snapshot"));
 
-  ICEBERG_UNWRAP_OR_FAIL(auto ref_builder, table.NewScan());
+  ICEBERG_UNWRAP_OR_FAIL(auto ref_builder, table->NewScan());
   auto ref_result = ref_builder->UseRef("dev").UseRef("main").Build();
   EXPECT_THAT(ref_result, IsError(ErrorKind::kValidationFailed));
   EXPECT_THAT(ref_result, HasErrorMessage("Cannot override ref"));
 
-  ICEBERG_UNWRAP_OR_FAIL(auto empty_ref_builder, table.NewScan());
+  ICEBERG_UNWRAP_OR_FAIL(auto empty_ref_builder, table->NewScan());
   EXPECT_THAT(empty_ref_builder->UseRef("").Build(),
               IsError(ErrorKind::kValidationFailed));
 }
 
-TEST_F(MetadataTableTest, HistoryTableScanIsNotSupported) {
+TEST_F(MetadataTableTest, HistoryTableScanAndTimeTravelSelection) {
   ICEBERG_UNWRAP_OR_FAIL(auto history_table, MetadataTable::Make<HistoryTable>(table_));
   ICEBERG_UNWRAP_OR_FAIL(auto builder, history_table->NewScan());
-  EXPECT_THAT(builder->Select({"snapshot_id"}).Build(),
-              IsError(ErrorKind::kNotSupported));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, BuildStream(builder->Select({"snapshot_id"})));
+  ICEBERG_UNWRAP_OR_FAIL(auto schema, ReadSchema(std::move(stream)));
+  EXPECT_THAT(schema->field_names(), ::testing::ElementsAre("snapshot_id"));
 
   // Tables without time travel reject snapshot selections.
   ICEBERG_UNWRAP_OR_FAIL(auto snapshot_builder, history_table->NewScan());
@@ -245,4 +188,33 @@ TEST_F(MetadataTableTest, HistoryTableScanIsNotSupported) {
   EXPECT_THAT(snapshot_result, HasErrorMessage("does not support time travel"));
 }
 
+TEST_F(MetadataTableTest, ScanReadsEveryTaskInPlanningOrder) {
+  auto projection = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(2, "snapshot_id", int64())});
+  auto metadata = std::make_shared<TableMetadata>();
+  metadata->snapshots = {std::make_shared<Snapshot>(Snapshot{.snapshot_id = 2})};
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto first,
+      StaticMetadataTask::Make(MetadataTableKind::kSnapshots, metadata, projection));
+  auto next_metadata = std::make_shared<TableMetadata>();
+  next_metadata->snapshots = {std::make_shared<Snapshot>(Snapshot{.snapshot_id = 1})};
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto second,
+      StaticMetadataTask::Make(MetadataTableKind::kSnapshots, next_metadata, projection));
+  ICEBERG_UNWRAP_OR_FAIL(auto empty, StaticMetadataTask::Make(
+                                         MetadataTableKind::kSnapshots,
+                                         std::make_shared<TableMetadata>(), projection));
+  PlannedMetadataScan scan({.kind = MetadataTableKind::kSnapshots,
+                            .table_metadata = metadata,
+                            .schema = SnapshotsTable::CanonicalSchema(),
+                            .projected_schema = projection},
+                           {empty, first, empty, second});
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, scan.ToArrowArrayStream());
+  ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(stream)));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto expected,
+      TableFromJson(*projection, R"([{"snapshot_id":2},{"snapshot_id":1}])"));
+  EXPECT_TRUE(TablesEqual(*expected, *actual));
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/metadata_table_test_base.h b/src/iceberg/test/metadata_table_test_base.h
index cefdb5cbe..de44c1d1d 100644
--- a/src/iceberg/test/metadata_table_test_base.h
+++ b/src/iceberg/test/metadata_table_test_base.h
@@ -39,6 +39,7 @@
 #include <gtest/gtest.h>
 
 #include "iceberg/constants.h"
+#include "iceberg/inspect/metadata_scan.h"
 #include "iceberg/inspect/metadata_table.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_field.h"
@@ -81,10 +82,16 @@ class MetadataTableTestBase : public ::testing::Test {
                                                "s3://bucket/meta.json", io_, catalog_));
   }
 
+  /// \brief Build a scan and open its projected rows.
+  static Result<ArrowArrayStream> BuildStream(MetadataScanBuilder& builder) {
+    ICEBERG_ASSIGN_OR_RAISE(auto scan, builder.Build());
+    return scan->ToArrowArrayStream();
+  }
+
   /// \brief Scan all columns of a metadata table.
   static Result<ArrowArrayStream> Scan(MetadataTable& table) {
     ICEBERG_ASSIGN_OR_RAISE(auto builder, table.NewScan());
-    return builder->Build();
+    return BuildStream(*builder);
   }
 
   /// \brief Scan the selected columns of a metadata table.
@@ -92,7 +99,7 @@ class MetadataTableTestBase : public ::testing::Test {
                                               const std::vector<std::string>& columns,
                                               bool case_sensitive = true) {
     ICEBERG_ASSIGN_OR_RAISE(auto builder, table.NewScan());
-    return builder->CaseSensitive(case_sensitive).Select(columns).Build();
+    return BuildStream(builder->CaseSensitive(case_sensitive).Select(columns));
   }
 
   /// \brief Import and consume a scan-produced ArrowArrayStream.
diff --git a/src/iceberg/test/snapshots_table_test.cc b/src/iceberg/test/snapshots_table_test.cc
index 9a62c6b63..dea10a01c 100644
--- a/src/iceberg/test/snapshots_table_test.cc
+++ b/src/iceberg/test/snapshots_table_test.cc
@@ -32,7 +32,9 @@
 #include <gtest/gtest.h>
 
 #include "iceberg/constants.h"
+#include "iceberg/inspect/metadata_scan_task_reader.h"
 #include "iceberg/inspect/metadata_table.h"
+#include "iceberg/test/arrow_test_util.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/test/metadata_table_test_base.h"
 
@@ -194,7 +196,7 @@ TEST_F(SnapshotsTableTest, ScanBuildsRepeatedly) {
   ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
   builder->Select({"snapshot_id"});
   for (int read = 0; read < 2; ++read) {
-    ICEBERG_UNWRAP_OR_FAIL(auto stream, builder->Build());
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, BuildStream(*builder));
     ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
     ASSERT_EQ(batches.size(), 1);
     EXPECT_EQ(batches.front()->num_rows(), 2);
@@ -223,7 +225,7 @@ TEST_F(SnapshotsTableTest, ScanProjectedColumns) {
                                     SchemaField::MakeRequired(8, "value", string()))),
       SchemaField::MakeRequired(2, "snapshot_id", int64())});
   ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
-  ICEBERG_UNWRAP_OR_FAIL(auto stream, builder->Project(projection).Build());
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, BuildStream(builder->Project(projection)));
   ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
   ASSERT_EQ(batches.size(), 1);
   const auto& batch = batches.front();
@@ -320,6 +322,7 @@ TEST_F(SnapshotsTableTest, ScanSkipsNullSnapshots) {
 TEST_F(SnapshotsTableTest, ScanEmptySummary) {
   auto [missing_summary, operation_only_summary] = MakeTestSnapshots();
   missing_summary->summary.clear();
+  missing_summary->manifest_list.clear();
   operation_only_summary->summary = {
       {SnapshotSummaryFields::kOperation, DataOperation::kAppend}};
   ICEBERG_UNWRAP_OR_FAIL(auto table,
@@ -334,14 +337,18 @@ TEST_F(SnapshotsTableTest, ScanEmptySummary) {
   auto summaries =
       std::static_pointer_cast<::arrow::MapArray>(batches.front()->column(5));
   EXPECT_TRUE(summaries->IsNull(0));
+  auto manifests =
+      std::static_pointer_cast<::arrow::StringArray>(batches.front()->column(4));
+  EXPECT_TRUE(manifests->IsNull(0));
+  EXPECT_FALSE(manifests->IsNull(1));
   EXPECT_FALSE(summaries->IsNull(1));
   EXPECT_EQ(summaries->value_length(1), 0);
 }
 
 TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) {
   auto snapshot = MakeTestSnapshots().first;
-  std::vector<std::shared_ptr<Snapshot>> snapshots(MetadataTable::kBatchSize + 1,
-                                                   snapshot);
+  std::vector<std::shared_ptr<Snapshot>> snapshots(
+      MetadataScanTaskReader::Options{}.batch_size + 1, snapshot);
   ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots(std::move(snapshots),
                                                             /*current_snapshot_id=*/1));
   ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
@@ -350,14 +357,14 @@ TEST_F(SnapshotsTableTest, ScanReturnsMultipleBatches) {
   ICEBERG_UNWRAP_OR_FAIL(auto stream, Scan(*snapshots_table));
   ICEBERG_UNWRAP_OR_FAIL(auto batches, ReadAllBatches(std::move(stream)));
   ASSERT_EQ(batches.size(), 2);
-  EXPECT_EQ(batches[0]->num_rows(), MetadataTable::kBatchSize);
+  EXPECT_EQ(batches[0]->num_rows(), MetadataScanTaskReader::Options{}.batch_size);
   EXPECT_EQ(batches[1]->num_rows(), 1);
 }
 
 TEST_F(SnapshotsTableTest, ScanUsesMetadataCapturedAtCreation) {
   auto snapshot = MakeTestSnapshots().first;
-  std::vector<std::shared_ptr<Snapshot>> snapshots(MetadataTable::kBatchSize + 1,
-                                                   snapshot);
+  std::vector<std::shared_ptr<Snapshot>> snapshots(
+      MetadataScanTaskReader::Options{}.batch_size + 1, snapshot);
   ICEBERG_UNWRAP_OR_FAIL(auto table, MakeTableWithSnapshots(std::move(snapshots),
                                                             /*current_snapshot_id=*/1));
   ICEBERG_UNWRAP_OR_FAIL(auto snapshots_table,
@@ -371,7 +378,8 @@ TEST_F(SnapshotsTableTest, ScanUsesMetadataCapturedAtCreation) {
   auto first_batch_result = reader->Next();
   ASSERT_TRUE(first_batch_result.ok()) << first_batch_result.status().ToString();
   ASSERT_NE(*first_batch_result, nullptr);
-  EXPECT_EQ((*first_batch_result)->num_rows(), MetadataTable::kBatchSize);
+  EXPECT_EQ((*first_batch_result)->num_rows(),
+            MetadataScanTaskReader::Options{}.batch_size);
 
   ICEBERG_UNWRAP_OR_FAIL(
       auto refreshed_table,
@@ -387,4 +395,50 @@ TEST_F(SnapshotsTableTest, ScanUsesMetadataCapturedAtCreation) {
   EXPECT_EQ((*second_batch_result)->num_rows(), 1);
 }
 
+TEST_F(SnapshotsTableTest, CanonicalSchema) {
+  auto expected = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "committed_at", timestamp_tz()),
+      SchemaField::MakeRequired(2, "snapshot_id", int64()),
+      SchemaField::MakeOptional(3, "parent_id", int64()),
+      SchemaField::MakeOptional(4, "operation", string()),
+      SchemaField::MakeOptional(5, "manifest_list", string()),
+      SchemaField::MakeOptional(
+          6, "summary",
+          std::make_shared<MapType>(SchemaField::MakeRequired(7, "key", string()),
+                                    SchemaField::MakeRequired(8, "value", string())))});
+  EXPECT_EQ(*SnapshotsTable::CanonicalSchema(), *expected);
+  EXPECT_EQ(*snapshots_table_->schema(), *expected);
+}
+
+TEST_F(SnapshotsTableTest, PlannedTaskRetainsProjectionAndIndependentReadState) {
+  ICEBERG_UNWRAP_OR_FAIL(auto builder, snapshots_table_->NewScan());
+  builder->Select({"snapshot_id", "operation"});
+  ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build());
+  ICEBERG_UNWRAP_OR_FAIL(auto tasks, scan->PlanTasks());
+  ASSERT_EQ(tasks.size(), 1);
+  ICEBERG_UNWRAP_OR_FAIL(auto reader, MetadataScanTaskReader::Make({.batch_size = 1}));
+  ICEBERG_UNWRAP_OR_FAIL(auto first, reader->Open(*tasks.front()));
+  ICEBERG_UNWRAP_OR_FAIL(auto second, reader->Open(*tasks.front()));
+  // Streams retain their own input and cursors after the planning objects are gone.
+  tasks.clear();
+  scan.reset();
+  reader.reset();
+  builder.reset();
+  snapshots_table_.reset();
+  table_.reset();
+  Schema selected({SchemaField::MakeRequired(2, "snapshot_id", int64()),
+                   SchemaField::MakeOptional(4, "operation", string())});
+  ICEBERG_UNWRAP_OR_FAIL(auto expected, TableFromJson(selected, R"([
+    {"snapshot_id":1,"operation":"append"},
+    {"snapshot_id":2,"operation":"append"}
+  ])"));
+  ICEBERG_UNWRAP_OR_FAIL(auto actual, ReadArrowTable(std::move(first)));
+  EXPECT_TRUE(TablesEqual(*expected, *actual));
+  ASSERT_EQ(actual->column(0)->num_chunks(), 2);
+  EXPECT_EQ(actual->column(0)->chunk(0)->length(), 1);
+  EXPECT_EQ(actual->column(0)->chunk(1)->length(), 1);
+  ICEBERG_UNWRAP_OR_FAIL(auto repeated, ReadArrowTable(std::move(second)));
+  EXPECT_TRUE(TablesEqual(*expected, *repeated));
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/stream_test.cc b/src/iceberg/test/stream_test.cc
index 91b352167..dbb9248ab 100644
--- a/src/iceberg/test/stream_test.cc
+++ b/src/iceberg/test/stream_test.cc
@@ -104,6 +104,43 @@ 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(StreamTest, FromVectorPreservesOrderAndEndOfStream) {
+  auto stream = Stream<int>::FromVector({3, 1, 2});
+  ICEBERG_UNWRAP_OR_FAIL(auto first, stream->Next());
+  ASSERT_TRUE(first.has_value());
+  EXPECT_EQ(*first, 3);
+  ICEBERG_UNWRAP_OR_FAIL(auto remaining, stream->ToVector());
+  EXPECT_EQ(remaining, (std::vector<int>{1, 2}));
+  for (int i = 0; i < 2; ++i) {
+    ICEBERG_UNWRAP_OR_FAIL(auto end, stream->Next());
+    EXPECT_FALSE(end.has_value());
+  }
+  auto empty = Stream<int>::FromVector({});
+  ICEBERG_UNWRAP_OR_FAIL(auto end, empty->Next());
+  EXPECT_FALSE(end.has_value());
+}
+
+TEST(StreamTest, FromVectorSupportsBooleans) {
+  auto stream = Stream<bool>::FromVector({true, false, true});
+  ICEBERG_UNWRAP_OR_FAIL(auto values, stream->ToVector());
+  EXPECT_EQ(values, (std::vector<bool>{true, false, true}));
+}
+
+TEST(StreamTest, FromVectorSupportsMoveAndCopyOnlyValues) {
+  std::vector<std::unique_ptr<int>> values;
+  values.push_back(std::make_unique<int>(42));
+  auto stream = Stream<std::unique_ptr<int>>::FromVector(std::move(values));
+  ICEBERG_UNWRAP_OR_FAIL(auto result, stream->ToVector());
+  ASSERT_EQ(result.size(), 1);
+  EXPECT_EQ(*result.front(), 42);
+
+  std::vector<CopyOnly> copy_values(1, CopyOnly(42));
+  auto copy_stream = Stream<CopyOnly>::FromVector(std::move(copy_values));
+  ICEBERG_UNWRAP_OR_FAIL(auto copied, copy_stream->ToVector());
+  ASSERT_EQ(copied.size(), 1);
+  EXPECT_EQ(copied.front().value(), 42);
+}
+
 TEST(StreamTest, SupportsIncompleteSharedPointerValue) {
   class IncompleteType;
   std::unique_ptr<Stream<std::shared_ptr<IncompleteType>>> stream;
diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h
index 9386c9dd6..c0230feb2 100644
--- a/src/iceberg/type_fwd.h
+++ b/src/iceberg/type_fwd.h
@@ -19,6 +19,8 @@
 
 #pragma once
 
+#include <cstdint>
+
 /// \file iceberg/type_fwd.h
 /// Forward declarations and enum definitions.  When writing your own headers,
 /// you can include this instead of the "full" headers to help reduce compile
@@ -280,9 +282,14 @@ class DeleteLoader;
 class PositionDeleteIndex;
 
 /// \brief Metadata tables.
+enum class MetadataTableKind : uint8_t;
+
 class HistoryTable;
 class MetadataScanBuilder;
 class MetadataTable;
+class MetadataScan;
+class MetadataScanTask;
+class MetadataScanTaskReader;
 class SnapshotsTable;
 
 /// \brief Table encryption
diff --git a/src/iceberg/util/stream.h b/src/iceberg/util/stream.h
index fc01e98c0..1dc3055a1 100644
--- a/src/iceberg/util/stream.h
+++ b/src/iceberg/util/stream.h
@@ -22,7 +22,9 @@
 /// \file iceberg/util/stream.h
 /// \brief Pull-based stream interface for fallible, lazily produced values.
 
+#include <cstddef>
 #include <deque>
+#include <memory>
 #include <optional>
 #include <type_traits>
 #include <utility>
@@ -62,6 +64,9 @@ class Stream {
   /// \brief Move-assign a stream and its terminal state.
   Stream& operator=(Stream&&) noexcept = default;
 
+  /// \brief Create a stream that owns and consumes the given values in order.
+  static std::unique_ptr<Stream<T>> FromVector(std::vector<T> values);
+
   /// \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
@@ -136,4 +141,27 @@ class Stream {
   std::optional<Error> error_;
 };
 
+template <typename T>
+std::unique_ptr<Stream<T>> Stream<T>::FromVector(std::vector<T> values) {
+  class VectorStream final : public Stream<T> {
+   public:
+    explicit VectorStream(std::vector<T> values) : values_(std::move(values)) {}
+
+   protected:
+    Result<std::optional<T>> NextImpl() override {
+      if (index_ == values_.size()) {
+        return std::nullopt;
+      }
+      auto&& value = values_[index_++];
+      return Result<std::optional<T>>(std::in_place, std::in_place,
+                                      std::move_if_noexcept(value));
+    }
+
+   private:
+    std::vector<T> values_;
+    size_t index_ = 0;
+  };
+  return std::make_unique<VectorStream>(std::move(values));
+}
+
 }  // namespace iceberg