This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new c712948f refactor: move each TableScan subclass to its own file (#994)
c712948f is described below
commit c712948f8cd29e14a3d83ae7070b3724579d34ac
Author: Manu Zhang <[email protected]>
AuthorDate: Sun Oct 11 21:19:38 2026 +0800
refactor: move each TableScan subclass to its own file (#994)
Move DataTableScan, IncrementalAppendScan and IncrementalChangelogScan
into data_table_scan.{h,cc}, incremental_append_scan.{h,cc} and
incremental_changelog_scan.{h,cc}. table_scan.{h,cc} keep TableScan,
IncrementalScan, TableScanBuilder, the scan task types and
TableScanContext.
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
example/demo_example.cc | 1 +
src/iceberg/CMakeLists.txt | 3 +
src/iceberg/cpp20_compatibility_internal.h | 3 +
src/iceberg/data_table_scan.cc | 220 +++++++++++
src/iceberg/data_table_scan.h | 66 ++++
src/iceberg/incremental_append_scan.cc | 118 ++++++
src/iceberg/incremental_append_scan.h | 60 +++
src/iceberg/incremental_changelog_scan.cc | 183 +++++++++
src/iceberg/incremental_changelog_scan.h | 61 +++
src/iceberg/table.cc | 3 +
src/iceberg/table_scan.cc | 418 +--------------------
src/iceberg/table_scan.h | 76 ----
src/iceberg/test/incremental_append_scan_test.cc | 2 +
.../test/incremental_changelog_scan_test.cc | 2 +
src/iceberg/test/scan_planning_metrics_test.cc | 1 +
src/iceberg/test/table_scan_test.cc | 1 +
16 files changed, 731 insertions(+), 487 deletions(-)
diff --git a/example/demo_example.cc b/example/demo_example.cc
index 6098a98f..a965f9e8 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 7394ce82..657d487c 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 8c4a465f..b5f46e48 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 00000000..f35f454c
--- /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 00000000..a6de0020
--- /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 00000000..cf01fe89
--- /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 00000000..74922d27
--- /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 00000000..5a1dbba1
--- /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 00000000..6625b68f
--- /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 c512c11d..ca02c623 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 f6023a6f..46f235fb 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 7310d435..6ec29ad9 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 194b5115..90cdc8f5 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 dbf71537..e8b37404 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 9098ccde..000f0e59 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 c9af765e..f08440a1 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"
