diff --git a/example/demo_example.cc b/example/demo_example.cc
index 6098a98ff..a965f9e89 100644
--- a/example/demo_example.cc
+++ b/example/demo_example.cc
@@ -23,6 +23,7 @@
 #include "iceberg/arrow/arrow_io_util.h"
 #include "iceberg/avro/avro_register.h"
 #include "iceberg/catalog/memory/in_memory_catalog.h"
+#include "iceberg/data_table_scan.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/parquet/parquet_register.h"
 #include "iceberg/table.h"
diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index b5fb4874e..1542442cb 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -26,6 +26,7 @@ set(ICEBERG_SOURCES
     catalog/memory/in_memory_catalog.cc
     catalog/session_catalog.cc
     catalog/session_context.cc
+    data_table_scan.cc
     delete_file_index.cc
     deletes/dv_util.cc
     deletes/dv_writer.cc
@@ -53,6 +54,8 @@ set(ICEBERG_SOURCES
     file_reader.cc
     file_writer.cc
     geospatial.cc
+    incremental_append_scan.cc
+    incremental_changelog_scan.cc
     inspect/history_table.cc
     inspect/metadata_table.cc
     inspect/snapshots_table.cc
diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h
index 5c54b5441..2ca87e450 100644
--- a/src/iceberg/cpp20_compatibility_internal.h
+++ b/src/iceberg/cpp20_compatibility_internal.h
@@ -42,6 +42,7 @@
 #include "iceberg/data/file_scan_task_reader.h"
 #include "iceberg/data/position_delete_writer.h"
 #include "iceberg/data/writer.h"
+#include "iceberg/data_table_scan.h"
 #include "iceberg/delete_file_index.h"
 #include "iceberg/deletes/dv_writer.h"
 #include "iceberg/deletes/position_delete_index.h"
@@ -74,6 +75,8 @@
 #include "iceberg/iceberg_bundle_export.h"
 #include "iceberg/iceberg_data_export.h"
 #include "iceberg/iceberg_export.h"
+#include "iceberg/incremental_append_scan.h"
+#include "iceberg/incremental_changelog_scan.h"
 #include "iceberg/inheritable_metadata.h"
 #include "iceberg/inspect/history_table.h"
 #include "iceberg/inspect/metadata_table.h"
diff --git a/src/iceberg/data_table_scan.cc b/src/iceberg/data_table_scan.cc
new file mode 100644
index 000000000..f35f454c0
--- /dev/null
+++ b/src/iceberg/data_table_scan.cc
@@ -0,0 +1,220 @@
+/*
+ * 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/data_table_scan.h"
+
+#include <algorithm>
+#include <chrono>
+#include <iterator>
+#include <optional>
+#include <utility>
+
+#include "iceberg/expression/sanitize_expression.h"
+#include "iceberg/manifest/manifest_group.h"
+#include "iceberg/metrics/metrics_context.h"
+#include "iceberg/metrics/metrics_reporter.h"
+#include "iceberg/metrics/scan_report.h"
+#include "iceberg/schema.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/type_util.h"
+
+namespace iceberg {
+
+namespace {
+
+template <typename T>
+class EmptyStream final : public Stream<T> {
+ public:
+  Result<std::optional<T>> NextImpl() override { return std::nullopt; }
+};
+
+Result<ScanReport> MakeScanReport(const DataTableScan& scan, const Snapshot& snapshot,
+                                  ScanMetricsResult scan_metrics) {
+  ICEBERG_ASSIGN_OR_RAISE(auto schema_ptr, scan.schema());
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto projected_id_set,
+      GetProjectedIdsVisitor::GetProjectedIds(*schema_ptr, /*include_struct_ids=*/true));
+  std::vector<int32_t> projected_field_ids(projected_id_set.begin(),
+                                           projected_id_set.end());
+  std::ranges::sort(projected_field_ids);
+
+  std::vector<std::string> projected_field_names;
+  projected_field_names.reserve(projected_field_ids.size());
+  for (int32_t field_id : projected_field_ids) {
+    ICEBERG_ASSIGN_OR_RAISE(auto field_name, schema_ptr->FindColumnNameById(field_id));
+    ICEBERG_CHECK(field_name.has_value(), "Projected field {} not found in schema",
+                  field_id);
+    projected_field_names.emplace_back(*field_name);
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto sanitized_filter,
+                          SanitizeExpression::Sanitize(*schema_ptr, scan.filter(),
+                                                       scan.context().case_sensitive));
+
+  return ScanReport{
+      .table_name = scan.context().table_name,
+      .snapshot_id = snapshot.snapshot_id,
+      .filter = std::move(sanitized_filter),
+      .schema_id = schema_ptr->schema_id(),
+      .projected_field_ids = std::move(projected_field_ids),
+      .projected_field_names = std::move(projected_field_names),
+      .scan_metrics = std::move(scan_metrics),
+      .metadata = scan.context().options,
+  };
+}
+
+class ReportingFileTaskStream final : public FileScanTaskStream {
+ public:
+  ReportingFileTaskStream(FileScanTaskStreamPtr stream,
+                          std::shared_ptr<ScanMetrics> scan_metrics,
+                          std::chrono::nanoseconds planning_duration,
+                          std::shared_ptr<MetricsReporter> reporter, ScanReport report)
+      : stream_(std::move(stream)),
+        scan_metrics_(std::move(scan_metrics)),
+        planning_duration_(std::move(planning_duration)),
+        reporter_(std::move(reporter)),
+        report_(std::move(report)) {}
+
+  ~ReportingFileTaskStream() override { Finalize(); }
+
+  Result<std::optional<std::shared_ptr<FileScanTask>>> NextImpl() override {
+    auto start = std::chrono::steady_clock::now();
+    auto result = stream_->Next();
+    planning_duration_ += std::chrono::duration_cast<std::chrono::nanoseconds>(
+        std::chrono::steady_clock::now() - start);
+    if (!result.has_value()) {
+      // Failed planning does not emit a successful scan report.
+      finalized_ = true;
+    } else if (!result.value().has_value()) {
+      Finalize();
+    }
+    return result;
+  }
+
+ private:
+  void Finalize() {
+    if (finalized_) {
+      return;
+    }
+    finalized_ = true;
+    scan_metrics_->total_planning_duration->Record(planning_duration_);
+    report_.scan_metrics = scan_metrics_->ToResult();
+    std::ignore = reporter_->Report(report_);
+  }
+
+  FileScanTaskStreamPtr stream_;
+  std::shared_ptr<ScanMetrics> scan_metrics_;
+  std::chrono::nanoseconds planning_duration_;
+  std::shared_ptr<MetricsReporter> reporter_;
+  ScanReport report_;
+  bool finalized_ = false;
+};
+
+}  // namespace
+
+Result<std::unique_ptr<DataTableScan>> DataTableScan::Make(
+    std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+    std::shared_ptr<FileIO> io, internal::TableScanContext context) {
+  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+  ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
+  ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
+  return std::unique_ptr<DataTableScan>(new DataTableScan(
+      std::move(metadata), std::move(schema), std::move(io), std::move(context)));
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() const {
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, PlanFilesStream());
+  return stream->ToVector();
+}
+
+Result<FileScanTaskStreamPtr> DataTableScan::PlanFilesStream() const {
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot, this->snapshot());
+  if (!snapshot) {
+    return std::make_unique<EmptyStream<std::shared_ptr<FileScanTask>>>();
+  }
+
+  std::shared_ptr<ScanMetrics> scan_metrics;
+  std::optional<std::chrono::steady_clock::time_point> planning_start;
+  if (context_.metrics_reporter) {
+    auto metrics_context = MetricsContext::Default();
+    scan_metrics = ScanMetrics::Make(*metrics_context);
+    planning_start = std::chrono::steady_clock::now();
+  }
+
+  TableMetadataCache metadata_cache(metadata_.get());
+  ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById());
+
+  SnapshotReader snapshot_reader(snapshot.get());
+  ICEBERG_ASSIGN_OR_RAISE(auto data_manifests, snapshot_reader.DataManifests(io_));
+  ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests, snapshot_reader.DeleteManifests(io_));
+
+  if (scan_metrics) {
+    scan_metrics->total_data_manifests->Increment(
+        static_cast<int64_t>(data_manifests.size()));
+    scan_metrics->total_delete_manifests->Increment(
+        static_cast<int64_t>(delete_manifests.size()));
+  }
+
+  std::vector<ManifestFile> owned_data_manifests(
+      std::make_move_iterator(data_manifests.begin()),
+      std::make_move_iterator(data_manifests.end()));
+  std::vector<ManifestFile> owned_delete_manifests(
+      std::make_move_iterator(delete_manifests.begin()),
+      std::make_move_iterator(delete_manifests.end()));
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto manifest_group,
+      ManifestGroup::Make(io_, schema_, specs_by_id, std::move(owned_data_manifests),
+                          std::move(owned_delete_manifests)));
+  manifest_group->CaseSensitive(context_.case_sensitive)
+      .Select(ScanColumns())
+      .FilterData(filter())
+      .IgnoreDeleted()
+      .ColumnsToKeepStats(context_.columns_to_keep_stats)
+      .WithScanMetrics(scan_metrics);
+  if (data_manifests.size() > 1 || delete_manifests.size() > 1) {
+    manifest_group->PlanWith(context_.plan_executor);
+  }
+  if (context_.ignore_residuals) {
+    manifest_group->IgnoreResiduals();
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto stream, std::move(*manifest_group).PlanFilesStream());
+  if (!planning_start.has_value()) {
+    return stream;
+  }
+
+  auto planning_duration = std::chrono::duration_cast<std::chrono::nanoseconds>(
+      std::chrono::steady_clock::now() - planning_start.value());
+
+  auto report = MakeScanReport(*this, *snapshot, ScanMetricsResult{});
+  if (!report.has_value()) {
+    // Scan reporting is best effort.
+    return stream;
+  }
+
+  return std::make_unique<ReportingFileTaskStream>(
+      std::move(stream), std::move(scan_metrics), planning_duration,
+      context_.metrics_reporter, std::move(report).value());
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/data_table_scan.h b/src/iceberg/data_table_scan.h
new file mode 100644
index 000000000..a6de0020c
--- /dev/null
+++ b/src/iceberg/data_table_scan.h
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/data_table_scan.h
+/// \brief Scan that reads the data files of a table snapshot.
+
+#include <memory>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/table_scan.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief A scan that reads data files and applies delete files to filter rows.
+class ICEBERG_EXPORT DataTableScan : public TableScan {
+ public:
+  ~DataTableScan() override = default;
+
+  /// \brief Constructs a DataTableScan instance.
+  static Result<std::unique_ptr<DataTableScan>> Make(
+      std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+      std::shared_ptr<FileIO> io, internal::TableScanContext context);
+
+  /// \brief Plans the scan tasks by resolving manifests and data files.
+  ///
+  /// Collects PlanFilesStream() into a vector.
+  /// \return A Result containing scan tasks or an error.
+  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const;
+
+  /// \brief Lazily plans scan tasks by resolving manifests and data files on demand.
+  ///
+  /// The returned fallible, single-pass stream owns its planning resources and
+  /// can outlive this scan. An executor configured through PlanWith() is borrowed and
+  /// must remain alive until the stream is destroyed, as later Next() calls may submit
+  /// work to it.
+  Result<FileScanTaskStreamPtr> PlanFilesStream() const;
+
+ protected:
+  using TableScan::TableScan;
+};
+
+extern template class ICEBERG_EXTERN_TEMPLATE_CLASS_EXPORT
+    TableScanBuilder<DataTableScan>;
+
+}  // namespace iceberg
diff --git a/src/iceberg/incremental_append_scan.cc b/src/iceberg/incremental_append_scan.cc
new file mode 100644
index 000000000..cf01fe894
--- /dev/null
+++ b/src/iceberg/incremental_append_scan.cc
@@ -0,0 +1,118 @@
+/*
+ * 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/incremental_append_scan.h"
+
+#include <algorithm>
+#include <iterator>
+#include <unordered_set>
+#include <utility>
+
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/manifest/manifest_group.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/snapshot_util.h"
+
+namespace iceberg {
+
+Result<std::unique_ptr<IncrementalAppendScan>> IncrementalAppendScan::Make(
+    std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+    std::shared_ptr<FileIO> io, internal::TableScanContext context) {
+  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+  ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
+  ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
+  return std::unique_ptr<IncrementalAppendScan>(new IncrementalAppendScan(
+      std::move(metadata), std::move(schema), std::move(io), std::move(context)));
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFiles()
+    const {
+  return ResolvePlanFiles<FileScanTask>(*this);
+}
+
+Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFiles(
+    std::optional<int64_t> from_snapshot_id_exclusive,
+    int64_t to_snapshot_id_inclusive) const {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto ancestors_snapshots,
+      SnapshotUtil::AncestorsBetween(*metadata_, to_snapshot_id_inclusive,
+                                     from_snapshot_id_exclusive));
+
+  std::vector<std::shared_ptr<Snapshot>> append_snapshots;
+  std::ranges::copy_if(ancestors_snapshots, std::back_inserter(append_snapshots),
+                       [](const auto& snapshot) {
+                         return snapshot != nullptr &&
+                                snapshot->Operation().has_value() &&
+                                snapshot->Operation().value() == DataOperation::kAppend;
+                       });
+  if (append_snapshots.empty()) {
+    return std::vector<std::shared_ptr<FileScanTask>>{};
+  }
+
+  std::unordered_set<int64_t> snapshot_ids;
+  std::ranges::transform(append_snapshots,
+                         std::inserter(snapshot_ids, snapshot_ids.end()),
+                         [](const auto& snapshot) { return snapshot->snapshot_id; });
+
+  std::unordered_set<ManifestFile> data_manifests;
+  for (const auto& snapshot : append_snapshots) {
+    SnapshotReader snapshot_reader(snapshot.get());
+    ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.DataManifests(io_));
+    std::ranges::copy_if(
+        manifests, std::inserter(data_manifests, data_manifests.end()),
+        [&snapshot_ids](const ManifestFile& manifest) {
+          return manifest.added_snapshot_id.has_value() &&
+                 snapshot_ids.contains(manifest.added_snapshot_id.value());
+        });
+  }
+  if (data_manifests.empty()) {
+    return std::vector<std::shared_ptr<FileScanTask>>{};
+  }
+
+  TableMetadataCache metadata_cache(metadata_.get());
+  ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById());
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto manifest_group,
+      ManifestGroup::Make(
+          io_, schema_, specs_by_id,
+          std::vector<ManifestFile>(data_manifests.begin(), data_manifests.end()), {}));
+
+  manifest_group->CaseSensitive(context_.case_sensitive)
+      .Select(ScanColumns())
+      .FilterData(filter())
+      .FilterManifestEntries([&snapshot_ids](const ManifestEntry& entry) {
+        return entry.snapshot_id.has_value() &&
+               snapshot_ids.contains(entry.snapshot_id.value()) &&
+               entry.status == ManifestStatus::kAdded;
+      })
+      .IgnoreDeleted()
+      .ColumnsToKeepStats(context_.columns_to_keep_stats)
+      .PlanWith(context_.plan_executor);
+
+  if (context_.ignore_residuals) {
+    manifest_group->IgnoreResiduals();
+  }
+
+  return std::move(*manifest_group).PlanFiles();
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/incremental_append_scan.h b/src/iceberg/incremental_append_scan.h
new file mode 100644
index 000000000..74922d277
--- /dev/null
+++ b/src/iceberg/incremental_append_scan.h
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+/// \file iceberg/incremental_append_scan.h
+/// \brief Scan that reads the data files appended between snapshots.
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/table_scan.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief A scan that reads data files added between snapshots (incremental appends).
+class ICEBERG_EXPORT IncrementalAppendScan : public IncrementalScan<FileScanTask> {
+ public:
+  /// \brief Constructs an IncrementalAppendScan instance.
+  static Result<std::unique_ptr<IncrementalAppendScan>> Make(
+      std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+      std::shared_ptr<FileIO> io, internal::TableScanContext context);
+
+  ~IncrementalAppendScan() override = default;
+
+  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const override;
+
+ protected:
+  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles(
+      std::optional<int64_t> from_snapshot_id_exclusive,
+      int64_t to_snapshot_id_inclusive) const override;
+
+  using IncrementalScan::IncrementalScan;
+};
+
+extern template class ICEBERG_EXTERN_TEMPLATE_CLASS_EXPORT
+    TableScanBuilder<IncrementalAppendScan>;
+
+}  // namespace iceberg
diff --git a/src/iceberg/incremental_changelog_scan.cc b/src/iceberg/incremental_changelog_scan.cc
new file mode 100644
index 000000000..5a1dbba11
--- /dev/null
+++ b/src/iceberg/incremental_changelog_scan.cc
@@ -0,0 +1,183 @@
+/*
+ * 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/incremental_changelog_scan.h"
+
+#include <limits>
+#include <ranges>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+#include "iceberg/expression/residual_evaluator.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/manifest/manifest_group.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/util/content_file_util.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/snapshot_util.h"
+
+namespace iceberg {
+
+Result<std::unique_ptr<IncrementalChangelogScan>> IncrementalChangelogScan::Make(
+    std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+    std::shared_ptr<FileIO> io, internal::TableScanContext context) {
+  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
+  ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
+  ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
+  return std::unique_ptr<IncrementalChangelogScan>(new IncrementalChangelogScan(
+      std::move(metadata), std::move(schema), std::move(io), std::move(context)));
+}
+
+Result<std::vector<std::shared_ptr<ChangelogScanTask>>>
+IncrementalChangelogScan::PlanFiles() const {
+  return ResolvePlanFiles<ChangelogScanTask>(*this);
+}
+
+Result<std::vector<std::shared_ptr<ChangelogScanTask>>>
+IncrementalChangelogScan::PlanFiles(std::optional<int64_t> from_snapshot_id_exclusive,
+                                    int64_t to_snapshot_id_inclusive) const {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto ancestors_snapshots,
+      SnapshotUtil::AncestorsBetween(*metadata_, to_snapshot_id_inclusive,
+                                     from_snapshot_id_exclusive));
+
+  std::vector<std::pair<std::shared_ptr<Snapshot>, std::unique_ptr<SnapshotReader>>>
+      changelog_snapshots;
+
+  for (const auto& snapshot : std::ranges::reverse_view(ancestors_snapshots)) {
+    auto operation = snapshot->Operation();
+    if (!operation.has_value() || operation.value() != DataOperation::kReplace) {
+      auto snapshot_reader = std::make_unique<SnapshotReader>(snapshot.get());
+      ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests,
+                              snapshot_reader->DeleteManifests(io_));
+      if (!delete_manifests.empty()) {
+        return NotSupported(
+            "Delete files are currently not supported in changelog scans");
+      }
+      changelog_snapshots.emplace_back(snapshot, std::move(snapshot_reader));
+    }
+  }
+  if (changelog_snapshots.empty()) {
+    return std::vector<std::shared_ptr<ChangelogScanTask>>{};
+  }
+
+  std::unordered_set<int64_t> snapshot_ids;
+  std::unordered_map<int64_t, int32_t> snapshot_ordinals;
+  for (const auto& snapshot : changelog_snapshots) {
+    ICEBERG_PRECHECK(
+        std::cmp_less_equal(snapshot_ids.size(), std::numeric_limits<int32_t>::max()),
+        "Number of snapshots in changelog scan exceeds maximum supported");
+    snapshot_ids.insert(snapshot.first->snapshot_id);
+    snapshot_ordinals.try_emplace(snapshot.first->snapshot_id,
+                                  static_cast<int32_t>(snapshot_ordinals.size()));
+  }
+
+  std::vector<ManifestFile> data_manifests;
+  std::unordered_set<std::string> seen_manifest_paths;
+  for (const auto& snapshot : changelog_snapshots) {
+    ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot.second->DataManifests(io_));
+    for (auto& manifest : manifests) {
+      if (manifest.added_snapshot_id.has_value() &&
+          snapshot_ids.contains(manifest.added_snapshot_id.value()) &&
+          seen_manifest_paths.insert(manifest.manifest_path).second) {
+        data_manifests.push_back(manifest);
+      }
+    }
+  }
+  if (data_manifests.empty()) {
+    return std::vector<std::shared_ptr<ChangelogScanTask>>{};
+  }
+
+  TableMetadataCache metadata_cache(metadata_.get());
+  ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById());
+
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto manifest_group,
+      ManifestGroup::Make(io_, schema_, specs_by_id, std::move(data_manifests),
+                          /*delete_manifests=*/{}));
+
+  manifest_group->CaseSensitive(context_.case_sensitive)
+      .Select(ScanColumns())
+      .FilterData(filter())
+      .FilterManifestEntries([&snapshot_ids](const ManifestEntry& entry) {
+        return entry.snapshot_id.has_value() &&
+               snapshot_ids.contains(entry.snapshot_id.value());
+      })
+      .IgnoreExisting()
+      .ColumnsToKeepStats(context_.columns_to_keep_stats)
+      .PlanWith(context_.plan_executor);
+
+  if (context_.ignore_residuals) {
+    manifest_group->IgnoreResiduals();
+  }
+
+  auto create_tasks_func =
+      [&snapshot_ordinals](
+          std::vector<ManifestEntry>&& entries,
+          const TaskContext& ctx) -> Result<std::vector<std::shared_ptr<ScanTask>>> {
+    std::vector<std::shared_ptr<ScanTask>> tasks;
+    tasks.reserve(entries.size());
+
+    for (auto& entry : entries) {
+      ICEBERG_PRECHECK(entry.snapshot_id.has_value() && entry.data_file,
+                       "Invalid manifest entry with missing snapshot id or data file");
+
+      int64_t commit_snapshot_id = entry.snapshot_id.value();
+      auto ordinal_it = snapshot_ordinals.find(commit_snapshot_id);
+      ICEBERG_PRECHECK(ordinal_it != snapshot_ordinals.end(),
+                       "Invalid manifest entry with missing snapshot ordinal");
+
+      int32_t change_ordinal = ordinal_it->second;
+
+      if (ctx.drop_stats) {
+        ContentFileUtil::DropAllStats(*entry.data_file);
+      } else if (!ctx.columns_to_keep_stats.empty()) {
+        ContentFileUtil::DropUnselectedStats(*entry.data_file, ctx.columns_to_keep_stats);
+      }
+
+      ICEBERG_ASSIGN_OR_RAISE(auto residual,
+                              ctx.residuals->ResidualFor(entry.data_file->partition));
+      switch (entry.status) {
+        case ManifestStatus::kAdded:
+          tasks.push_back(std::make_shared<AddedRowsScanTask>(
+              change_ordinal, commit_snapshot_id, std::move(entry.data_file),
+              std::vector<std::shared_ptr<DataFile>>{}, std::move(residual)));
+          break;
+        case ManifestStatus::kDeleted:
+          tasks.push_back(std::make_shared<DeletedDataFileScanTask>(
+              change_ordinal, commit_snapshot_id, std::move(entry.data_file),
+              std::vector<std::shared_ptr<DataFile>>{}, std::move(residual)));
+          break;
+        case ManifestStatus::kExisting:
+          return InvalidArgument("Unexpected entry status: EXISTING");
+      }
+    }
+    return tasks;
+  };
+
+  ICEBERG_ASSIGN_OR_RAISE(auto tasks, manifest_group->Plan(create_tasks_func));
+  return tasks | std::views::transform([](const auto& task) {
+           return std::static_pointer_cast<ChangelogScanTask>(task);
+         }) |
+         std::ranges::to<std::vector>();
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/incremental_changelog_scan.h b/src/iceberg/incremental_changelog_scan.h
new file mode 100644
index 000000000..6625b68f2
--- /dev/null
+++ b/src/iceberg/incremental_changelog_scan.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
+
+/// \file iceberg/incremental_changelog_scan.h
+/// \brief Scan that reads the changelog between snapshots.
+
+#include <cstdint>
+#include <memory>
+#include <optional>
+#include <vector>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+#include "iceberg/table_scan.h"
+#include "iceberg/type_fwd.h"
+
+namespace iceberg {
+
+/// \brief A scan that reads changelog entries between snapshots.
+class ICEBERG_EXPORT IncrementalChangelogScan
+    : public IncrementalScan<ChangelogScanTask> {
+ public:
+  /// \brief Constructs an IncrementalChangelogScan instance.
+  static Result<std::unique_ptr<IncrementalChangelogScan>> Make(
+      std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
+      std::shared_ptr<FileIO> io, internal::TableScanContext context);
+
+  ~IncrementalChangelogScan() override = default;
+
+  Result<std::vector<std::shared_ptr<ChangelogScanTask>>> PlanFiles() const override;
+
+ protected:
+  Result<std::vector<std::shared_ptr<ChangelogScanTask>>> PlanFiles(
+      std::optional<int64_t> from_snapshot_id_exclusive,
+      int64_t to_snapshot_id_inclusive) const override;
+
+  using IncrementalScan::IncrementalScan;
+};
+
+extern template class ICEBERG_EXTERN_TEMPLATE_CLASS_EXPORT
+    TableScanBuilder<IncrementalChangelogScan>;
+
+}  // namespace iceberg
diff --git a/src/iceberg/table.cc b/src/iceberg/table.cc
index c512c11df..ca02c6231 100644
--- a/src/iceberg/table.cc
+++ b/src/iceberg/table.cc
@@ -22,6 +22,9 @@
 #include <memory>
 
 #include "iceberg/catalog.h"
+#include "iceberg/data_table_scan.h"
+#include "iceberg/incremental_append_scan.h"
+#include "iceberg/incremental_changelog_scan.h"
 #include "iceberg/location_provider.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc
index f6023a6fc..46f235fbe 100644
--- a/src/iceberg/table_scan.cc
+++ b/src/iceberg/table_scan.cc
@@ -20,27 +20,23 @@
 #include "iceberg/table_scan.h"
 
 #include <algorithm>
-#include <chrono>
 #include <cstdint>
 #include <iterator>
 #include <ranges>
 #include <utility>
 
+#include "iceberg/data_table_scan.h"
 #include "iceberg/expression/binder.h"
 #include "iceberg/expression/expression.h"
-#include "iceberg/expression/residual_evaluator.h"
-#include "iceberg/expression/sanitize_expression.h"
+#include "iceberg/incremental_append_scan.h"
+#include "iceberg/incremental_changelog_scan.h"
 #include "iceberg/manifest/manifest_entry.h"
-#include "iceberg/manifest/manifest_group.h"
-#include "iceberg/metrics/metrics_context.h"
 #include "iceberg/metrics/metrics_reporters.h"
-#include "iceberg/metrics/scan_report.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table.h"
 #include "iceberg/table_metadata.h"
-#include "iceberg/util/content_file_util.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/snapshot_util.h"
 #include "iceberg/util/timepoint.h"
@@ -67,95 +63,6 @@ const std::vector<std::string> kScanColumnsWithStats = [] {
   return cols;
 }();
 
-template <typename T>
-class EmptyStream final : public Stream<T> {
- public:
-  Result<std::optional<T>> NextImpl() override { return std::nullopt; }
-};
-
-Result<ScanReport> MakeScanReport(const DataTableScan& scan, const Snapshot& snapshot,
-                                  ScanMetricsResult scan_metrics) {
-  ICEBERG_ASSIGN_OR_RAISE(auto schema_ptr, scan.schema());
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto projected_id_set,
-      GetProjectedIdsVisitor::GetProjectedIds(*schema_ptr, /*include_struct_ids=*/true));
-  std::vector<int32_t> projected_field_ids(projected_id_set.begin(),
-                                           projected_id_set.end());
-  std::ranges::sort(projected_field_ids);
-
-  std::vector<std::string> projected_field_names;
-  projected_field_names.reserve(projected_field_ids.size());
-  for (int32_t field_id : projected_field_ids) {
-    ICEBERG_ASSIGN_OR_RAISE(auto field_name, schema_ptr->FindColumnNameById(field_id));
-    ICEBERG_CHECK(field_name.has_value(), "Projected field {} not found in schema",
-                  field_id);
-    projected_field_names.emplace_back(*field_name);
-  }
-
-  ICEBERG_ASSIGN_OR_RAISE(auto sanitized_filter,
-                          SanitizeExpression::Sanitize(*schema_ptr, scan.filter(),
-                                                       scan.context().case_sensitive));
-
-  return ScanReport{
-      .table_name = scan.context().table_name,
-      .snapshot_id = snapshot.snapshot_id,
-      .filter = std::move(sanitized_filter),
-      .schema_id = schema_ptr->schema_id(),
-      .projected_field_ids = std::move(projected_field_ids),
-      .projected_field_names = std::move(projected_field_names),
-      .scan_metrics = std::move(scan_metrics),
-      .metadata = scan.context().options,
-  };
-}
-
-class ReportingFileTaskStream final : public FileScanTaskStream {
- public:
-  ReportingFileTaskStream(FileScanTaskStreamPtr stream,
-                          std::shared_ptr<ScanMetrics> scan_metrics,
-                          std::chrono::nanoseconds planning_duration,
-                          std::shared_ptr<MetricsReporter> reporter, ScanReport report)
-      : stream_(std::move(stream)),
-        scan_metrics_(std::move(scan_metrics)),
-        planning_duration_(std::move(planning_duration)),
-        reporter_(std::move(reporter)),
-        report_(std::move(report)) {}
-
-  ~ReportingFileTaskStream() override { Finalize(); }
-
-  Result<std::optional<std::shared_ptr<FileScanTask>>> NextImpl() override {
-    auto start = std::chrono::steady_clock::now();
-    auto result = stream_->Next();
-    planning_duration_ += std::chrono::duration_cast<std::chrono::nanoseconds>(
-        std::chrono::steady_clock::now() - start);
-    if (!result.has_value()) {
-      // Failed planning does not emit a successful scan report.
-      finalized_ = true;
-    } else if (!result.value().has_value()) {
-      Finalize();
-    }
-    return result;
-  }
-
- private:
-  void Finalize() {
-    if (finalized_) {
-      return;
-    }
-    finalized_ = true;
-    scan_metrics_->total_planning_duration->Record(planning_duration_);
-    report_.scan_metrics = scan_metrics_->ToResult();
-    std::ignore = reporter_->Report(report_);
-  }
-
-  FileScanTaskStreamPtr stream_;
-  std::shared_ptr<ScanMetrics> scan_metrics_;
-  std::chrono::nanoseconds planning_duration_;
-  std::shared_ptr<MetricsReporter> reporter_;
-  ScanReport report_;
-  bool finalized_ = false;
-};
-
 }  // namespace
 
 namespace internal {
@@ -650,92 +557,6 @@ const std::vector<std::string>& TableScan::ScanColumns() const {
   return context_.return_column_stats ? kScanColumnsWithStats : kScanColumns;
 }
 
-Result<std::unique_ptr<DataTableScan>> DataTableScan::Make(
-    std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
-    std::shared_ptr<FileIO> io, internal::TableScanContext context) {
-  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
-  ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
-  ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
-  return std::unique_ptr<DataTableScan>(new DataTableScan(
-      std::move(metadata), std::move(schema), std::move(io), std::move(context)));
-}
-
-Result<std::vector<std::shared_ptr<FileScanTask>>> DataTableScan::PlanFiles() const {
-  ICEBERG_ASSIGN_OR_RAISE(auto stream, PlanFilesStream());
-  return stream->ToVector();
-}
-
-Result<FileScanTaskStreamPtr> DataTableScan::PlanFilesStream() const {
-  ICEBERG_ASSIGN_OR_RAISE(auto snapshot, this->snapshot());
-  if (!snapshot) {
-    return std::make_unique<EmptyStream<std::shared_ptr<FileScanTask>>>();
-  }
-
-  std::shared_ptr<ScanMetrics> scan_metrics;
-  std::optional<std::chrono::steady_clock::time_point> planning_start;
-  if (context_.metrics_reporter) {
-    auto metrics_context = MetricsContext::Default();
-    scan_metrics = ScanMetrics::Make(*metrics_context);
-    planning_start = std::chrono::steady_clock::now();
-  }
-
-  TableMetadataCache metadata_cache(metadata_.get());
-  ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById());
-
-  SnapshotReader snapshot_reader(snapshot.get());
-  ICEBERG_ASSIGN_OR_RAISE(auto data_manifests, snapshot_reader.DataManifests(io_));
-  ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests, snapshot_reader.DeleteManifests(io_));
-
-  if (scan_metrics) {
-    scan_metrics->total_data_manifests->Increment(
-        static_cast<int64_t>(data_manifests.size()));
-    scan_metrics->total_delete_manifests->Increment(
-        static_cast<int64_t>(delete_manifests.size()));
-  }
-
-  std::vector<ManifestFile> owned_data_manifests(
-      std::make_move_iterator(data_manifests.begin()),
-      std::make_move_iterator(data_manifests.end()));
-  std::vector<ManifestFile> owned_delete_manifests(
-      std::make_move_iterator(delete_manifests.begin()),
-      std::make_move_iterator(delete_manifests.end()));
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto manifest_group,
-      ManifestGroup::Make(io_, schema_, specs_by_id, std::move(owned_data_manifests),
-                          std::move(owned_delete_manifests)));
-  manifest_group->CaseSensitive(context_.case_sensitive)
-      .Select(ScanColumns())
-      .FilterData(filter())
-      .IgnoreDeleted()
-      .ColumnsToKeepStats(context_.columns_to_keep_stats)
-      .WithScanMetrics(scan_metrics);
-  if (data_manifests.size() > 1 || delete_manifests.size() > 1) {
-    manifest_group->PlanWith(context_.plan_executor);
-  }
-  if (context_.ignore_residuals) {
-    manifest_group->IgnoreResiduals();
-  }
-
-  ICEBERG_ASSIGN_OR_RAISE(auto stream, std::move(*manifest_group).PlanFilesStream());
-  if (!planning_start.has_value()) {
-    return stream;
-  }
-
-  auto planning_duration = std::chrono::duration_cast<std::chrono::nanoseconds>(
-      std::chrono::steady_clock::now() - planning_start.value());
-
-  auto report = MakeScanReport(*this, *snapshot, ScanMetricsResult{});
-  if (!report.has_value()) {
-    // Scan reporting is best effort.
-    return stream;
-  }
-
-  return std::make_unique<ReportingFileTaskStream>(
-      std::move(stream), std::move(scan_metrics), planning_duration,
-      context_.metrics_reporter, std::move(report).value());
-}
-
 // Friend function template for IncrementalScan that implements the shared PlanFiles
 // logic. It resolves the from/to snapshot range from the scan context and delegates
 // to the two-arg virtual PlanFiles() override in the concrete subclass.
@@ -760,234 +581,9 @@ Result<std::vector<std::shared_ptr<ScanTaskType>>> ResolvePlanFiles(
   return scan.PlanFiles(from_snapshot_id_exclusive, to_snapshot_id_inclusive);
 }
 
-// IncrementalAppendScan implementation
-
-Result<std::unique_ptr<IncrementalAppendScan>> IncrementalAppendScan::Make(
-    std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
-    std::shared_ptr<FileIO> io, internal::TableScanContext context) {
-  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
-  ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
-  ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
-  return std::unique_ptr<IncrementalAppendScan>(new IncrementalAppendScan(
-      std::move(metadata), std::move(schema), std::move(io), std::move(context)));
-}
-
-Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFiles()
-    const {
-  return ResolvePlanFiles<FileScanTask>(*this);
-}
-
-Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFiles(
-    std::optional<int64_t> from_snapshot_id_exclusive,
-    int64_t to_snapshot_id_inclusive) const {
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto ancestors_snapshots,
-      SnapshotUtil::AncestorsBetween(*metadata_, to_snapshot_id_inclusive,
-                                     from_snapshot_id_exclusive));
-
-  std::vector<std::shared_ptr<Snapshot>> append_snapshots;
-  std::ranges::copy_if(ancestors_snapshots, std::back_inserter(append_snapshots),
-                       [](const auto& snapshot) {
-                         return snapshot != nullptr &&
-                                snapshot->Operation().has_value() &&
-                                snapshot->Operation().value() == DataOperation::kAppend;
-                       });
-  if (append_snapshots.empty()) {
-    return std::vector<std::shared_ptr<FileScanTask>>{};
-  }
-
-  std::unordered_set<int64_t> snapshot_ids;
-  std::ranges::transform(append_snapshots,
-                         std::inserter(snapshot_ids, snapshot_ids.end()),
-                         [](const auto& snapshot) { return snapshot->snapshot_id; });
-
-  std::unordered_set<ManifestFile> data_manifests;
-  for (const auto& snapshot : append_snapshots) {
-    SnapshotReader snapshot_reader(snapshot.get());
-    ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_reader.DataManifests(io_));
-    std::ranges::copy_if(
-        manifests, std::inserter(data_manifests, data_manifests.end()),
-        [&snapshot_ids](const ManifestFile& manifest) {
-          return manifest.added_snapshot_id.has_value() &&
-                 snapshot_ids.contains(manifest.added_snapshot_id.value());
-        });
-  }
-  if (data_manifests.empty()) {
-    return std::vector<std::shared_ptr<FileScanTask>>{};
-  }
-
-  TableMetadataCache metadata_cache(metadata_.get());
-  ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById());
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto manifest_group,
-      ManifestGroup::Make(
-          io_, schema_, specs_by_id,
-          std::vector<ManifestFile>(data_manifests.begin(), data_manifests.end()), {}));
-
-  manifest_group->CaseSensitive(context_.case_sensitive)
-      .Select(ScanColumns())
-      .FilterData(filter())
-      .FilterManifestEntries([&snapshot_ids](const ManifestEntry& entry) {
-        return entry.snapshot_id.has_value() &&
-               snapshot_ids.contains(entry.snapshot_id.value()) &&
-               entry.status == ManifestStatus::kAdded;
-      })
-      .IgnoreDeleted()
-      .ColumnsToKeepStats(context_.columns_to_keep_stats)
-      .PlanWith(context_.plan_executor);
-
-  if (context_.ignore_residuals) {
-    manifest_group->IgnoreResiduals();
-  }
-
-  return std::move(*manifest_group).PlanFiles();
-}
-
-// IncrementalChangelogScan implementation
-
-Result<std::unique_ptr<IncrementalChangelogScan>> IncrementalChangelogScan::Make(
-    std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
-    std::shared_ptr<FileIO> io, internal::TableScanContext context) {
-  ICEBERG_PRECHECK(metadata != nullptr, "Table metadata cannot be null");
-  ICEBERG_PRECHECK(schema != nullptr, "Schema cannot be null");
-  ICEBERG_PRECHECK(io != nullptr, "FileIO cannot be null");
-  return std::unique_ptr<IncrementalChangelogScan>(new IncrementalChangelogScan(
-      std::move(metadata), std::move(schema), std::move(io), std::move(context)));
-}
-
-Result<std::vector<std::shared_ptr<ChangelogScanTask>>>
-IncrementalChangelogScan::PlanFiles() const {
-  return ResolvePlanFiles<ChangelogScanTask>(*this);
-}
-
-Result<std::vector<std::shared_ptr<ChangelogScanTask>>>
-IncrementalChangelogScan::PlanFiles(std::optional<int64_t> from_snapshot_id_exclusive,
-                                    int64_t to_snapshot_id_inclusive) const {
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto ancestors_snapshots,
-      SnapshotUtil::AncestorsBetween(*metadata_, to_snapshot_id_inclusive,
-                                     from_snapshot_id_exclusive));
-
-  std::vector<std::pair<std::shared_ptr<Snapshot>, std::unique_ptr<SnapshotReader>>>
-      changelog_snapshots;
-
-  for (const auto& snapshot : std::ranges::reverse_view(ancestors_snapshots)) {
-    auto operation = snapshot->Operation();
-    if (!operation.has_value() || operation.value() != DataOperation::kReplace) {
-      auto snapshot_reader = std::make_unique<SnapshotReader>(snapshot.get());
-      ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests,
-                              snapshot_reader->DeleteManifests(io_));
-      if (!delete_manifests.empty()) {
-        return NotSupported(
-            "Delete files are currently not supported in changelog scans");
-      }
-      changelog_snapshots.emplace_back(snapshot, std::move(snapshot_reader));
-    }
-  }
-  if (changelog_snapshots.empty()) {
-    return std::vector<std::shared_ptr<ChangelogScanTask>>{};
-  }
-
-  std::unordered_set<int64_t> snapshot_ids;
-  std::unordered_map<int64_t, int32_t> snapshot_ordinals;
-  for (const auto& snapshot : changelog_snapshots) {
-    ICEBERG_PRECHECK(
-        std::cmp_less_equal(snapshot_ids.size(), std::numeric_limits<int32_t>::max()),
-        "Number of snapshots in changelog scan exceeds maximum supported");
-    snapshot_ids.insert(snapshot.first->snapshot_id);
-    snapshot_ordinals.try_emplace(snapshot.first->snapshot_id,
-                                  static_cast<int32_t>(snapshot_ordinals.size()));
-  }
-
-  std::vector<ManifestFile> data_manifests;
-  std::unordered_set<std::string> seen_manifest_paths;
-  for (const auto& snapshot : changelog_snapshots) {
-    ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot.second->DataManifests(io_));
-    for (auto& manifest : manifests) {
-      if (manifest.added_snapshot_id.has_value() &&
-          snapshot_ids.contains(manifest.added_snapshot_id.value()) &&
-          seen_manifest_paths.insert(manifest.manifest_path).second) {
-        data_manifests.push_back(manifest);
-      }
-    }
-  }
-  if (data_manifests.empty()) {
-    return std::vector<std::shared_ptr<ChangelogScanTask>>{};
-  }
-
-  TableMetadataCache metadata_cache(metadata_.get());
-  ICEBERG_ASSIGN_OR_RAISE(auto specs_by_id, metadata_cache.GetPartitionSpecsById());
-
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto manifest_group,
-      ManifestGroup::Make(io_, schema_, specs_by_id, std::move(data_manifests),
-                          /*delete_manifests=*/{}));
-
-  manifest_group->CaseSensitive(context_.case_sensitive)
-      .Select(ScanColumns())
-      .FilterData(filter())
-      .FilterManifestEntries([&snapshot_ids](const ManifestEntry& entry) {
-        return entry.snapshot_id.has_value() &&
-               snapshot_ids.contains(entry.snapshot_id.value());
-      })
-      .IgnoreExisting()
-      .ColumnsToKeepStats(context_.columns_to_keep_stats)
-      .PlanWith(context_.plan_executor);
-
-  if (context_.ignore_residuals) {
-    manifest_group->IgnoreResiduals();
-  }
-
-  auto create_tasks_func =
-      [&snapshot_ordinals](
-          std::vector<ManifestEntry>&& entries,
-          const TaskContext& ctx) -> Result<std::vector<std::shared_ptr<ScanTask>>> {
-    std::vector<std::shared_ptr<ScanTask>> tasks;
-    tasks.reserve(entries.size());
-
-    for (auto& entry : entries) {
-      ICEBERG_PRECHECK(entry.snapshot_id.has_value() && entry.data_file,
-                       "Invalid manifest entry with missing snapshot id or data file");
-
-      int64_t commit_snapshot_id = entry.snapshot_id.value();
-      auto ordinal_it = snapshot_ordinals.find(commit_snapshot_id);
-      ICEBERG_PRECHECK(ordinal_it != snapshot_ordinals.end(),
-                       "Invalid manifest entry with missing snapshot ordinal");
-
-      int32_t change_ordinal = ordinal_it->second;
-
-      if (ctx.drop_stats) {
-        ContentFileUtil::DropAllStats(*entry.data_file);
-      } else if (!ctx.columns_to_keep_stats.empty()) {
-        ContentFileUtil::DropUnselectedStats(*entry.data_file, ctx.columns_to_keep_stats);
-      }
-
-      ICEBERG_ASSIGN_OR_RAISE(auto residual,
-                              ctx.residuals->ResidualFor(entry.data_file->partition));
-      switch (entry.status) {
-        case ManifestStatus::kAdded:
-          tasks.push_back(std::make_shared<AddedRowsScanTask>(
-              change_ordinal, commit_snapshot_id, std::move(entry.data_file),
-              std::vector<std::shared_ptr<DataFile>>{}, std::move(residual)));
-          break;
-        case ManifestStatus::kDeleted:
-          tasks.push_back(std::make_shared<DeletedDataFileScanTask>(
-              change_ordinal, commit_snapshot_id, std::move(entry.data_file),
-              std::vector<std::shared_ptr<DataFile>>{}, std::move(residual)));
-          break;
-        case ManifestStatus::kExisting:
-          return InvalidArgument("Unexpected entry status: EXISTING");
-      }
-    }
-    return tasks;
-  };
-
-  ICEBERG_ASSIGN_OR_RAISE(auto tasks, manifest_group->Plan(create_tasks_func));
-  return tasks | std::views::transform([](const auto& task) {
-           return std::static_pointer_cast<ChangelogScanTask>(task);
-         }) |
-         std::ranges::to<std::vector>();
-}
+template Result<std::vector<std::shared_ptr<FileScanTask>>> ResolvePlanFiles(
+    const IncrementalScan<FileScanTask>& scan);
+template Result<std::vector<std::shared_ptr<ChangelogScanTask>>> ResolvePlanFiles(
+    const IncrementalScan<ChangelogScanTask>& scan);
 
 }  // namespace iceberg
diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h
index 7310d435b..6ec29ad9f 100644
--- a/src/iceberg/table_scan.h
+++ b/src/iceberg/table_scan.h
@@ -462,34 +462,6 @@ class ICEBERG_EXPORT TableScan {
   mutable std::shared_ptr<Schema> projected_schema_;
 };
 
-/// \brief A scan that reads data files and applies delete files to filter rows.
-class ICEBERG_EXPORT DataTableScan : public TableScan {
- public:
-  ~DataTableScan() override = default;
-
-  /// \brief Constructs a DataTableScan instance.
-  static Result<std::unique_ptr<DataTableScan>> Make(
-      std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
-      std::shared_ptr<FileIO> io, internal::TableScanContext context);
-
-  /// \brief Plans the scan tasks by resolving manifests and data files.
-  ///
-  /// Collects PlanFilesStream() into a vector.
-  /// \return A Result containing scan tasks or an error.
-  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const;
-
-  /// \brief Lazily plans scan tasks by resolving manifests and data files on demand.
-  ///
-  /// The returned fallible, single-pass stream owns its planning resources and
-  /// can outlive this scan. An executor configured through PlanWith() is borrowed and
-  /// must remain alive until the stream is destroyed, as later Next() calls may submit
-  /// work to it.
-  Result<FileScanTaskStreamPtr> PlanFilesStream() const;
-
- protected:
-  using TableScan::TableScan;
-};
-
 /// \brief A base template class for incremental scans that read changes between
 /// snapshots, and return scan tasks of the specified type.
 template <typename ScanTaskType>
@@ -512,52 +484,4 @@ class IncrementalScan : public TableScan {
       const IncrementalScan<T>& scan);
 };
 
-/// \brief A scan that reads data files added between snapshots (incremental appends).
-class ICEBERG_EXPORT IncrementalAppendScan : public IncrementalScan<FileScanTask> {
- public:
-  /// \brief Constructs an IncrementalAppendScan instance.
-  static Result<std::unique_ptr<IncrementalAppendScan>> Make(
-      std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
-      std::shared_ptr<FileIO> io, internal::TableScanContext context);
-
-  ~IncrementalAppendScan() override = default;
-
-  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles() const override;
-
- protected:
-  Result<std::vector<std::shared_ptr<FileScanTask>>> PlanFiles(
-      std::optional<int64_t> from_snapshot_id_exclusive,
-      int64_t to_snapshot_id_inclusive) const override;
-
-  using IncrementalScan::IncrementalScan;
-};
-
-/// \brief A scan that reads changelog entries between snapshots.
-class ICEBERG_EXPORT IncrementalChangelogScan
-    : public IncrementalScan<ChangelogScanTask> {
- public:
-  /// \brief Constructs an IncrementalChangelogScan instance.
-  static Result<std::unique_ptr<IncrementalChangelogScan>> Make(
-      std::shared_ptr<TableMetadata> metadata, std::shared_ptr<Schema> schema,
-      std::shared_ptr<FileIO> io, internal::TableScanContext context);
-
-  ~IncrementalChangelogScan() override = default;
-
-  Result<std::vector<std::shared_ptr<ChangelogScanTask>>> PlanFiles() const override;
-
- protected:
-  Result<std::vector<std::shared_ptr<ChangelogScanTask>>> PlanFiles(
-      std::optional<int64_t> from_snapshot_id_exclusive,
-      int64_t to_snapshot_id_inclusive) const override;
-
-  using IncrementalScan::IncrementalScan;
-};
-
-extern template class ICEBERG_EXTERN_TEMPLATE_CLASS_EXPORT
-    TableScanBuilder<DataTableScan>;
-extern template class ICEBERG_EXTERN_TEMPLATE_CLASS_EXPORT
-    TableScanBuilder<IncrementalAppendScan>;
-extern template class ICEBERG_EXTERN_TEMPLATE_CLASS_EXPORT
-    TableScanBuilder<IncrementalChangelogScan>;
-
 }  // namespace iceberg
diff --git a/src/iceberg/test/incremental_append_scan_test.cc b/src/iceberg/test/incremental_append_scan_test.cc
index 194b51159..90cdc8f5a 100644
--- a/src/iceberg/test/incremental_append_scan_test.cc
+++ b/src/iceberg/test/incremental_append_scan_test.cc
@@ -17,6 +17,8 @@
  * under the License.
  */
 
+#include "iceberg/incremental_append_scan.h"
+
 #include <memory>
 #include <optional>
 #include <string>
diff --git a/src/iceberg/test/incremental_changelog_scan_test.cc b/src/iceberg/test/incremental_changelog_scan_test.cc
index dbf715375..e8b374044 100644
--- a/src/iceberg/test/incremental_changelog_scan_test.cc
+++ b/src/iceberg/test/incremental_changelog_scan_test.cc
@@ -17,6 +17,8 @@
  * under the License.
  */
 
+#include "iceberg/incremental_changelog_scan.h"
+
 #include <memory>
 #include <optional>
 #include <string>
diff --git a/src/iceberg/test/scan_planning_metrics_test.cc b/src/iceberg/test/scan_planning_metrics_test.cc
index 9098ccded..000f0e590 100644
--- a/src/iceberg/test/scan_planning_metrics_test.cc
+++ b/src/iceberg/test/scan_planning_metrics_test.cc
@@ -26,6 +26,7 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include "iceberg/data_table_scan.h"
 #include "iceberg/expression/expressions.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/metrics/metrics_reporter.h"
diff --git a/src/iceberg/test/table_scan_test.cc b/src/iceberg/test/table_scan_test.cc
index c9af765e9..f08440a10 100644
--- a/src/iceberg/test/table_scan_test.cc
+++ b/src/iceberg/test/table_scan_test.cc
@@ -27,6 +27,7 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
+#include "iceberg/data_table_scan.h"
 #include "iceberg/expression/expressions.h"
 #include "iceberg/snapshot.h"
 #include "iceberg/table_metadata.h"