WZhuo commented on code in PR #611:
URL: https://github.com/apache/iceberg-cpp/pull/611#discussion_r3051186365
##########
src/iceberg/table_scan.cc:
##########
@@ -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 (snapshot_ids.contains(manifest.added_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));
Review Comment:
It's also set for `FileScanTask`.
##########
src/iceberg/table_scan.h:
##########
@@ -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
Review Comment:
It's a dead code perhaps be left over rebase, remove it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
