feat: Impl Incremental changelog scan

Author: WZhuo

src/iceberg/table_scan.cc

@@ -24,13 +24,15 @@
 
 #include "iceberg/expression/binder.h"
 #include "iceberg/expression/expression.h"
+#include "iceberg/expression/residual_evaluator.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/manifest/manifest_group.h"
 #include "iceberg/result.h"
 #include "iceberg/schema.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_internal.h"
 #include "iceberg/util/timepoint.h"
@@ -294,6 +296,24 @@ Result<ArrowArrayStream> FileScanTask::ToArrow(
   return MakeArrowArrayStream(std::move(reader));
 }
 
+// ChangelogScanTask implementation
+
+int64_t ChangelogScanTask::size_bytes() const {
+  int64_t total_size = data_file_->file_size_in_bytes;
+  for (const auto& delete_file : delete_files_) {
+    total_size +=
+        (delete_file->IsDeletionVector() ? delete_file->content_size_in_bytes.value_or(0)
+                                         : delete_file->file_size_in_bytes);
+  }
+  return total_size;
+}
+
+int32_t ChangelogScanTask::files_count() const { return 1 + delete_files_.size(); }
+
+int64_t ChangelogScanTask::estimated_row_count() const {
+  return data_file_->record_count;
+}
+
 // Generic template implementation for Make
 template <typename ScanType>
 Result<std::unique_ptr<TableScanBuilder<ScanType>>> TableScanBuilder<ScanType>::Make(
@@ -747,11 +767,13 @@ Result<std::vector<std::shared_ptr<FileScanTask>>> IncrementalAppendScan::PlanFi
 // IncrementalChangelogScan implementation
 
 Result<std::unique_ptr<IncrementalChangelogScan>> IncrementalChangelogScan::Make(
-    [[maybe_unused]] std::shared_ptr<TableMetadata> metadata,
-    [[maybe_unused]] std::shared_ptr<Schema> schema,
-    [[maybe_unused]] std::shared_ptr<FileIO> io,
-    [[maybe_unused]] internal::TableScanContext context) {
-  return NotImplemented("IncrementalChangelogScan is not implemented");
+    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>>>
@@ -762,7 +784,126 @@ IncrementalChangelogScan::PlanFiles() const {
 Result<std::vector<std::shared_ptr<ChangelogScanTask>>>
 IncrementalChangelogScan::PlanFiles(std::optional<int64_t> from_snapshot_id_exclusive,
                                     int64_t to_snapshot_id_inclusive) const {
-  return NotImplemented("IncrementalChangelogScan::PlanFiles is not implemented");
+  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<SnapshotCache>>>
+      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_cache = std::make_unique<SnapshotCache>(snapshot.get());
+      ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests,
+                              snapshot_cache->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_cache));
+    }
+  }
+  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;
+  int32_t ordinal = 0;
+  for (const auto& snapshot : changelog_snapshots) {
+    snapshot_ids.insert(snapshot.first->snapshot_id);
+    snapshot_ordinals[snapshot.first->snapshot_id] = ordinal++;
+  }
+
+  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 == snapshot.first->snapshot_id &&
+          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), {}));
+
+  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);
+
+  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) {
+      if (!entry.snapshot_id.has_value() || entry.data_file == nullptr) {
+        continue;
+      }
+
+      int64_t commit_snapshot_id = entry.snapshot_id.value();
+      auto ordinal_it = snapshot_ordinals.find(commit_snapshot_id);
+      if (ordinal_it == snapshot_ordinals.end()) {
+        continue;
+      }
+      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

src/iceberg/table_scan.h

@@ -102,14 +102,87 @@ class ICEBERG_EXPORT FileScanTask : public ScanTask {
   std::shared_ptr<Expression> residual_filter_;
 };
 
+enum class ChangelogOperation : uint8_t {
+  kInsert,
+  kDelete,
+  kUpdateBefore,
+  kUpdateAfter,
+};
+
 /// \brief A scan task for reading changelog entries between snapshots.
 class ICEBERG_EXPORT ChangelogScanTask : public ScanTask {
  public:
+  /// \brief Construct an AddedRowsScanTask.
+  ///
+  /// \param change_ordinal Position in the changelog order (0-based).
+  /// \param commit_snapshot_id The snapshot ID that committed this change.
+  /// \param data_file The data file containing the added rows.
+  /// \param delete_files Delete files that apply to this data file.
+  /// \param residual_filter Optional residual filter to apply after reading.
+  ChangelogScanTask(int32_t change_ordinal, int64_t commit_snapshot_id,
+                    std::shared_ptr<DataFile> data_file,
+                    std::vector<std::shared_ptr<DataFile>> delete_files = {},
+                    std::shared_ptr<Expression> residual_filter = nullptr)
+      : change_ordinal_(change_ordinal),
+        commit_snapshot_id_(commit_snapshot_id),
+        data_file_(std::move(data_file)),
+        delete_files_(std::move(delete_files)),
+        residual_filter_(std::move(residual_filter)) {}
+
   Kind kind() const override { return Kind::kChangelogScanTask; }
-  // TODO(): Return actual values once member fields are implemented
-  int64_t size_bytes() const override { return 0; }
-  int32_t files_count() const override { return 0; }
-  int64_t estimated_row_count() const override { return 0; }
+
+  int64_t size_bytes() const override;
+  int32_t files_count() const override;
+  int64_t estimated_row_count() const override;
+
+  virtual ChangelogOperation operation() const = 0;
+
+  /// \brief The position of this change in the changelog order (0-based).
+  virtual int32_t change_ordinal() const { return change_ordinal_; }
+
+  /// \brief The snapshot ID that committed this change.
+  virtual int64_t commit_snapshot_id() const { return commit_snapshot_id_; }
+
+  /// \brief The data file containing the added rows.
+  const std::shared_ptr<DataFile>& data_file() const { return data_file_; }
+
+  /// \brief Delete files that apply to this data file.
+  const std::vector<std::shared_ptr<DataFile>>& delete_files() const {
+    return delete_files_;
+  }
+
+  /// \brief Residual filter to apply after reading.
+  const std::shared_ptr<Expression>& residual_filter() const { return residual_filter_; }
+
+ protected:
+  int32_t change_ordinal_;
+  int64_t commit_snapshot_id_;
+  std::shared_ptr<DataFile> data_file_;
+  std::vector<std::shared_ptr<DataFile>> delete_files_;
+  std::shared_ptr<Expression> residual_filter_;
+};
+
+/// \brief A scan task for reading rows that were added between snapshots.
+///
+/// This task represents data files that were added to the table, along with any
+/// delete files that should be applied when reading the data.
+class ICEBERG_EXPORT AddedRowsScanTask : public ChangelogScanTask {
+ public:
+  using ChangelogScanTask::ChangelogScanTask;
+
+  ChangelogOperation operation() const override { return ChangelogOperation::kInsert; }
+};
+
+/// \brief A scan task for reading data files that were deleted between snapshots.
+///
+/// This task represents data files that were removed from the table. Unlike
+/// AddedRowsScanTask, delete files are not applicable here since the entire
+/// data file was deleted.
+class ICEBERG_EXPORT DeletedDataFileScanTask : public ChangelogScanTask {
+ public:
+  using ChangelogScanTask::ChangelogScanTask;
+
+  ChangelogOperation operation() const override { return ChangelogOperation::kDelete; }
 };
 
 namespace internal {
@@ -133,8 +206,24 @@ struct TableScanContext {
 
   // Validate the context parameters to see if they have conflicts.
   [[nodiscard]] Status Validate() const;
+
+  /// \brief Returns true if this scan is a current lineage scan, which means it does not
+  /// specify from/to snapshot IDs.
+  bool IsScanCurrentLineage() const;
+
+  /// \brief Get the snapshot ID to scan up to (inclusive) based on the context.
+  Result<int64_t> ToSnapshotIdInclusive(const TableMetadata& metadata) const;
+
+  /// \brief Get the snapshot ID to scan from (exclusive) based on the context.
+  Result<std::optional<int64_t>> FromSnapshotIdExclusive(
+      const TableMetadata& metadata, int64_t to_snapshot_id_inclusive) const;
 };
 
+// Internal validation functions for IncrementalScanBuilder
+Status CheckSnapshotValid(const TableMetadata& metadata, int64_t snapshot_id);
+Result<int64_t> CheckRefValid(const TableMetadata& metadata, const std::string& ref);
+Status CheckBranchValid(const TableMetadata& metadata, const std::string& branch);
+
 }  // namespace internal
 
 // Concept to check if a type is an incremental scan

src/iceberg/test/CMakeLists.txt

@@ -176,6 +176,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    SOURCES
                    file_scan_task_test.cc
                    incremental_append_scan_test.cc
+                   incremental_changelog_scan_test.cc
                    table_scan_test.cc)
 
   add_iceberg_test(table_update_test