From 357de935f3b59c842d29ec350551694de7642df2 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Wed, 23 Sep 2026 11:16:53 +0800 Subject: [PATCH 1/2] feat(scan): plan changelog tasks from deletion vectors IncrementalChangelogScan rejected every snapshot that carried a delete manifest, so tables using row-level deletes could not be scanned for changes. Format version 3 stores row-level deletes as deletion vectors, where each data file has at most one live vector and a new vector replaces the previous one. That makes the rows a snapshot deleted the difference between the vector it added and the vector it removed. Add DeletionVectorChangelogPlanner, which reads the deletion vectors added and removed by each changelog snapshot and plans the delete side: a vector committed with its data file joins the AddedRowsScanTask, a vector removed with its data file becomes the existing delete of the DeletedDataFileScanTask, and a vector for a data file that already existed produces the new DeletedRowsScanTask with the added vector and the replaced vector. IncrementalChangelogScan uses the planner only for format version 3 tables. Delete files on format version 1 and 2 tables, position delete files and equality delete files remain unsupported and are rejected with NotSupported. The planner reads every delete manifest of the changelog snapshots, not only those written in the range, so a position or equality delete file carried in an older manifest is rejected as well unless it was removed before the range and can no longer apply to any row. Co-Authored-By: Claude Fable 5.1 --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/changelog_dv_planner_internal.cc | 228 +++++++ src/iceberg/changelog_dv_planner_internal.h | 121 ++++ src/iceberg/incremental_changelog_scan.cc | 212 +++++-- src/iceberg/incremental_changelog_scan.h | 9 + src/iceberg/table_scan.cc | 35 ++ src/iceberg/table_scan.h | 58 ++ .../test/incremental_changelog_scan_test.cc | 593 ++++++++++++++++-- 8 files changed, 1135 insertions(+), 122 deletions(-) create mode 100644 src/iceberg/changelog_dv_planner_internal.cc create mode 100644 src/iceberg/changelog_dv_planner_internal.h diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index cf465e5fdc..e2035033d3 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 + changelog_dv_planner_internal.cc data_table_scan.cc delete_file_index.cc deletes/dv_util.cc diff --git a/src/iceberg/changelog_dv_planner_internal.cc b/src/iceberg/changelog_dv_planner_internal.cc new file mode 100644 index 0000000000..6496595e27 --- /dev/null +++ b/src/iceberg/changelog_dv_planner_internal.cc @@ -0,0 +1,228 @@ +/* + * 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/changelog_dv_planner_internal.h" + +#include +#include + +#include "iceberg/expression/residual_evaluator.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_group.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/util/content_file_util.h" +#include "iceberg/util/executor_util_internal.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +namespace { + +Status ValidateChangelogDeleteFile(const DataFile& file) { + if (file.content == DataFile::Content::kEqualityDeletes) { + return NotSupported("Equality delete files are not supported in changelog scans: {}", + file.file_path); + } + if (!file.IsDeletionVector()) { + return NotSupported("Position delete files are not supported in changelog scans: {}", + file.file_path); + } + ICEBERG_PRECHECK(file.referenced_data_file.has_value(), + "Deletion vector {} does not reference a data file", file.file_path); + return {}; +} + +std::vector> FindDeletionVector( + const std::unordered_map>& dvs_by_path, + const std::string& data_file_path) { + auto it = dvs_by_path.find(data_file_path); + if (it == dvs_by_path.end()) { + return {}; + } + return {it->second}; +} + +} // namespace + +DeletionVectorChangelogPlanner::DeletionVectorChangelogPlanner( + std::shared_ptr io, std::shared_ptr schema, + std::unordered_map> specs_by_id, + internal::TableScanContext context, std::vector scan_columns, + std::shared_ptr filter, + std::unordered_map dvs_by_snapshot) + : io_(std::move(io)), + schema_(std::move(schema)), + specs_by_id_(std::move(specs_by_id)), + context_(std::move(context)), + scan_columns_(std::move(scan_columns)), + filter_(std::move(filter)), + dvs_by_snapshot_(std::move(dvs_by_snapshot)) {} + +Result> +DeletionVectorChangelogPlanner::Make( + std::shared_ptr io, std::shared_ptr schema, + std::unordered_map> specs_by_id, + const internal::TableScanContext& context, std::vector scan_columns, + std::shared_ptr filter, const std::vector& delete_manifests, + const std::unordered_set& snapshot_ids) { + ICEBERG_ASSIGN_OR_RAISE( + auto entries, + ParallelCollect( + context.plan_executor, delete_manifests, + [&](const ManifestFile& manifest) -> Result> { + ICEBERG_ASSIGN_OR_RAISE( + auto reader, ManifestReader::Make(manifest, io, schema, specs_by_id)); + reader->TryDropStats(); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + auto in_range = [&snapshot_ids](const ManifestEntry& entry) { + return entry.snapshot_id.has_value() && + snapshot_ids.contains(entry.snapshot_id.value()); + }; + for (const auto& entry : entries) { + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Invalid manifest entry with missing delete file"); + // A delete file removed before the range no longer applies to any data + // file, so it cannot affect the changelog. Every other delete file can: a + // live one applies to data files whose rows the changelog reports, and one + // removed by a changelog snapshot was applied to the rows that snapshot + // reports as deleted. Those must be deletion vectors. + if (entry.status == ManifestStatus::kDeleted && !in_range(entry)) { + continue; + } + ICEBERG_RETURN_UNEXPECTED(ValidateChangelogDeleteFile(*entry.data_file)); + } + std::erase_if(entries, [&in_range](const ManifestEntry& entry) { + return entry.status == ManifestStatus::kExisting || !in_range(entry); + }); + return entries; + })); + + std::unordered_map dvs_by_snapshot; + for (auto& entry : entries) { + const int64_t snapshot_id = entry.snapshot_id.value(); + auto& dvs = dvs_by_snapshot[snapshot_id]; + auto& dvs_by_path = entry.status == ManifestStatus::kAdded ? dvs.added : dvs.removed; + const std::string& data_file_path = entry.data_file->referenced_data_file.value(); + auto [it, inserted] = + dvs_by_path.try_emplace(data_file_path, std::move(entry.data_file)); + ICEBERG_PRECHECK( + inserted, "Snapshot {} {} multiple deletion vectors for {}", snapshot_id, + entry.status == ManifestStatus::kAdded ? "added" : "removed", data_file_path); + } + + return std::unique_ptr( + new DeletionVectorChangelogPlanner( + std::move(io), std::move(schema), std::move(specs_by_id), context, + std::move(scan_columns), std::move(filter), std::move(dvs_by_snapshot))); +} + +std::vector> DeletionVectorChangelogPlanner::AddedDeletes( + int64_t snapshot_id, const std::string& data_file_path) const { + auto it = dvs_by_snapshot_.find(snapshot_id); + if (it == dvs_by_snapshot_.end()) { + return {}; + } + return FindDeletionVector(it->second.added, data_file_path); +} + +std::vector> DeletionVectorChangelogPlanner::RemovedDeletes( + int64_t snapshot_id, const std::string& data_file_path) const { + auto it = dvs_by_snapshot_.find(snapshot_id); + if (it == dvs_by_snapshot_.end()) { + return {}; + } + return FindDeletionVector(it->second.removed, data_file_path); +} + +Result>> +DeletionVectorChangelogPlanner::PlanDeletedRows( + int64_t snapshot_id, int32_t change_ordinal, std::span data_manifests, + const std::unordered_set& added_data_file_paths) const { + auto dvs_it = dvs_by_snapshot_.find(snapshot_id); + if (dvs_it == dvs_by_snapshot_.end()) { + return std::vector>{}; + } + const SnapshotDeletionVectors& dvs = dvs_it->second; + + std::unordered_set pending_paths; + for (const auto& [data_file_path, dv] : dvs.added) { + if (!added_data_file_paths.contains(data_file_path)) { + pending_paths.insert(data_file_path); + } + } + if (pending_paths.empty()) { + return std::vector>{}; + } + + ICEBERG_ASSIGN_OR_RAISE( + auto manifest_group, + ManifestGroup::Make( + io_, schema_, specs_by_id_, + std::vector(data_manifests.begin(), data_manifests.end()), + /*delete_manifests=*/{})); + manifest_group->CaseSensitive(context_.case_sensitive) + .Select(scan_columns_) + .FilterData(filter_) + .FilterManifestEntries([&pending_paths](const ManifestEntry& entry) { + return entry.data_file != nullptr && + pending_paths.contains(entry.data_file->file_path); + }) + .IgnoreDeleted() + .ColumnsToKeepStats(context_.columns_to_keep_stats) + .PlanWith(context_.plan_executor); + if (context_.ignore_residuals) { + manifest_group->IgnoreResiduals(); + } + + auto create_tasks_func = + [&](std::vector&& entries, + const TaskContext& ctx) -> Result>> { + std::vector> tasks; + tasks.reserve(entries.size()); + + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Invalid manifest entry with missing data file"); + + 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)); + const std::string& data_file_path = entry.data_file->file_path; + auto added_deletes = FindDeletionVector(dvs.added, data_file_path); + auto existing_deletes = FindDeletionVector(dvs.removed, data_file_path); + tasks.push_back(std::make_shared( + change_ordinal, snapshot_id, std::move(entry.data_file), + std::move(added_deletes), std::move(existing_deletes), std::move(residual))); + } + 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(task); + }) | + std::ranges::to(); +} + +} // namespace iceberg diff --git a/src/iceberg/changelog_dv_planner_internal.h b/src/iceberg/changelog_dv_planner_internal.h new file mode 100644 index 0000000000..1f76cffb37 --- /dev/null +++ b/src/iceberg/changelog_dv_planner_internal.h @@ -0,0 +1,121 @@ +/* + * 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/changelog_dv_planner_internal.h +/// Deletion vector handling for changelog scans of format version 3 tables. + +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/manifest/manifest_list.h" +#include "iceberg/result.h" +#include "iceberg/table_scan.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +/// \brief Plans the delete side of a changelog scan from deletion vectors. +/// +/// Format version 3 stores row-level deletes as deletion vectors, with at most one live +/// vector per data file, and a new vector replaces the previous one. The rows a snapshot +/// deleted are therefore the difference between the vector it added and the vector it +/// removed, which DeletedRowsScanTask carries as added and existing deletes. Position +/// delete files and equality delete files accumulate instead, so attributing their rows +/// to a snapshot would require reading every earlier delete file; they are rejected +/// wherever they could still apply to a reported row, whether or not a changelog +/// snapshot wrote them. +class DeletionVectorChangelogPlanner { + public: + /// \brief Index the deletion vectors added and removed by the changelog snapshots. + /// + /// \param context Scan context that supplies filtering and planning options. + /// \param scan_columns Manifest columns to read for data files. + /// \param filter Row filter of the scan. + /// \param delete_manifests All delete manifests of the changelog snapshots. Only the + /// manifests written by those snapshots can hold the vectors, because removing a + /// vector rewrites its manifest, but a manifest untouched during the range can still + /// carry position delete files or equality delete files that apply to the reported + /// rows. Every manifest is therefore checked, and only entries written by the + /// changelog snapshots are indexed. + /// \param snapshot_ids IDs of the changelog snapshots. + static Result> Make( + std::shared_ptr io, std::shared_ptr schema, + std::unordered_map> specs_by_id, + const internal::TableScanContext& context, std::vector scan_columns, + std::shared_ptr filter, + const std::vector& delete_manifests, + const std::unordered_set& snapshot_ids); + + /// \brief The deletion vector that the snapshot committed for the data file, if any. + std::vector> AddedDeletes( + int64_t snapshot_id, const std::string& data_file_path) const; + + /// \brief The deletion vector that the snapshot removed from the data file, if any. + std::vector> RemovedDeletes( + int64_t snapshot_id, const std::string& data_file_path) const; + + /// \brief Plan DeletedRowsScanTasks for the deletion vectors a snapshot committed + /// against data files it did not add. + /// + /// \param snapshot_id The changelog snapshot. + /// \param change_ordinal Position of the snapshot in the changelog order. + /// \param data_manifests All data manifests of the snapshot. The referenced data files + /// are located there because an earlier snapshot added them and a later snapshot in the + /// range may remove them again. + /// \param added_data_file_paths Data files added by the snapshot, whose deletion + /// vectors belong to their AddedRowsScanTask instead. + Result>> PlanDeletedRows( + int64_t snapshot_id, int32_t change_ordinal, std::span data_manifests, + const std::unordered_set& added_data_file_paths) const; + + private: + using DeletionVectorsByPath = + std::unordered_map>; + + // Deletion vectors committed and removed by one changelog snapshot, keyed by the path + // of the data file they reference. + struct SnapshotDeletionVectors { + DeletionVectorsByPath added; + DeletionVectorsByPath removed; + }; + + DeletionVectorChangelogPlanner( + std::shared_ptr io, std::shared_ptr schema, + std::unordered_map> specs_by_id, + internal::TableScanContext context, std::vector scan_columns, + std::shared_ptr filter, + std::unordered_map dvs_by_snapshot); + + std::shared_ptr io_; + std::shared_ptr schema_; + std::unordered_map> specs_by_id_; + internal::TableScanContext context_; + std::vector scan_columns_; + std::shared_ptr filter_; + std::unordered_map dvs_by_snapshot_; +}; + +} // namespace iceberg diff --git a/src/iceberg/incremental_changelog_scan.cc b/src/iceberg/incremental_changelog_scan.cc index 5a1dbba111..0395572bb7 100644 --- a/src/iceberg/incremental_changelog_scan.cc +++ b/src/iceberg/incremental_changelog_scan.cc @@ -19,12 +19,14 @@ #include "iceberg/incremental_changelog_scan.h" +#include #include #include #include #include #include +#include "iceberg/changelog_dv_planner_internal.h" #include "iceberg/expression/residual_evaluator.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/manifest/manifest_group.h" @@ -59,6 +61,10 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl SnapshotUtil::AncestorsBetween(*metadata_, to_snapshot_id_inclusive, from_snapshot_id_exclusive)); + // Row-level deletes are only understood as deletion vectors, which format version 3 + // requires; earlier versions store them as position and equality delete files. + const bool deletes_supported = metadata_->format_version >= 3; + std::vector, std::unique_ptr>> changelog_snapshots; @@ -66,11 +72,15 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl auto operation = snapshot->Operation(); if (!operation.has_value() || operation.value() != DataOperation::kReplace) { auto snapshot_reader = std::make_unique(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"); + if (!deletes_supported) { + ICEBERG_ASSIGN_OR_RAISE(auto delete_manifests, + snapshot_reader->DeleteManifests(io_)); + if (!delete_manifests.empty()) { + return NotSupported( + "Delete files are only supported in changelog scans of format version 3 " + "tables, but the table uses format version {}", + metadata_->format_version); + } } changelog_snapshots.emplace_back(snapshot, std::move(snapshot_reader)); } @@ -91,93 +101,157 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl } std::vector data_manifests; + std::vector delete_manifests; std::unordered_set seen_manifest_paths; - for (const auto& snapshot : changelog_snapshots) { - ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot.second->DataManifests(io_)); - for (auto& manifest : manifests) { + for (const auto& [snapshot, snapshot_reader] : changelog_snapshots) { + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_data_manifests, + snapshot_reader->DataManifests(io_)); + for (auto& manifest : snapshot_data_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 (deletes_supported) { + // Every delete manifest is collected, not only those written in the range, so + // that position delete files and equality delete files carried in older manifests + // are rejected rather than silently left out of the changelog. + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_delete_manifests, + snapshot_reader->DeleteManifests(io_)); + for (auto& manifest : snapshot_delete_manifests) { + if (seen_manifest_paths.insert(manifest.manifest_path).second) { + delete_manifests.push_back(manifest); + } + } + } } - if (data_manifests.empty()) { + if (data_manifests.empty() && delete_manifests.empty()) { return std::vector>{}; } 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(); + std::unique_ptr dv_planner; + if (!delete_manifests.empty()) { + ICEBERG_ASSIGN_OR_RAISE( + dv_planner, DeletionVectorChangelogPlanner::Make( + io_, schema_, specs_by_id, context_, ScanColumns(), filter(), + delete_manifests, snapshot_ids)); } - auto create_tasks_func = - [&snapshot_ordinals]( - std::vector&& entries, - const TaskContext& ctx) -> Result>> { - std::vector> tasks; - tasks.reserve(entries.size()); + std::vector> tasks; + // Data files added by each snapshot. A deletion vector committed together with its data + // file belongs to the AddedRowsScanTask rather than to a DeletedRowsScanTask. + std::unordered_map> added_paths_by_snapshot; - for (auto& entry : entries) { - ICEBERG_PRECHECK(entry.snapshot_id.has_value() && entry.data_file, - "Invalid manifest entry with missing snapshot id or data file"); + if (!data_manifests.empty()) { + ICEBERG_ASSIGN_OR_RAISE( + auto manifest_group, + ManifestGroup::Make(io_, schema_, specs_by_id, std::move(data_manifests), + /*delete_manifests=*/{})); - 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"); + 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); - int32_t change_ordinal = ordinal_it->second; + if (context_.ignore_residuals) { + manifest_group->IgnoreResiduals(); + } - 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); - } + auto create_tasks_func = + [&](std::vector&& entries, + const TaskContext& ctx) -> Result>> { + std::vector> tasks; + tasks.reserve(entries.size()); - 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( - change_ordinal, commit_snapshot_id, std::move(entry.data_file), - std::vector>{}, std::move(residual))); - break; - case ManifestStatus::kDeleted: - tasks.push_back(std::make_shared( - change_ordinal, commit_snapshot_id, std::move(entry.data_file), - std::vector>{}, std::move(residual))); - break; - case ManifestStatus::kExisting: - return InvalidArgument("Unexpected entry status: EXISTING"); + 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)); + const std::string& data_file_path = entry.data_file->file_path; + switch (entry.status) { + case ManifestStatus::kAdded: { + std::vector> deletes; + if (dv_planner) { + deletes = dv_planner->AddedDeletes(commit_snapshot_id, data_file_path); + added_paths_by_snapshot[commit_snapshot_id].insert(data_file_path); + } + tasks.push_back(std::make_shared( + change_ordinal, commit_snapshot_id, std::move(entry.data_file), + std::move(deletes), std::move(residual))); + break; + } + case ManifestStatus::kDeleted: { + std::vector> existing_deletes; + if (dv_planner) { + existing_deletes = + dv_planner->RemovedDeletes(commit_snapshot_id, data_file_path); + } + tasks.push_back(std::make_shared( + change_ordinal, commit_snapshot_id, std::move(entry.data_file), + std::move(existing_deletes), std::move(residual))); + break; + } + case ManifestStatus::kExisting: + return InvalidArgument("Unexpected entry status: EXISTING"); + } } + return tasks; + }; + + ICEBERG_ASSIGN_OR_RAISE(auto data_file_tasks, + manifest_group->Plan(create_tasks_func)); + tasks = data_file_tasks | std::views::transform([](const auto& task) { + return std::static_pointer_cast(task); + }) | + std::ranges::to(); + } + + if (dv_planner) { + const std::unordered_set no_added_paths; + for (const auto& [snapshot, snapshot_reader] : changelog_snapshots) { + const int64_t snapshot_id = snapshot->snapshot_id; + auto added_paths_it = added_paths_by_snapshot.find(snapshot_id); + const auto& added_paths = added_paths_it == added_paths_by_snapshot.end() + ? no_added_paths + : added_paths_it->second; + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_data_manifests, + snapshot_reader->DataManifests(io_)); + ICEBERG_ASSIGN_OR_RAISE( + auto deleted_rows_tasks, + dv_planner->PlanDeletedRows(snapshot_id, snapshot_ordinals.at(snapshot_id), + snapshot_data_manifests, added_paths)); + tasks.insert(tasks.end(), std::make_move_iterator(deleted_rows_tasks.begin()), + std::make_move_iterator(deleted_rows_tasks.end())); } - 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(task); - }) | - std::ranges::to(); + } + + return tasks; } } // namespace iceberg diff --git a/src/iceberg/incremental_changelog_scan.h b/src/iceberg/incremental_changelog_scan.h index 6625b68f22..e4995c2ea7 100644 --- a/src/iceberg/incremental_changelog_scan.h +++ b/src/iceberg/incremental_changelog_scan.h @@ -35,6 +35,15 @@ namespace iceberg { /// \brief A scan that reads changelog entries between snapshots. +/// +/// Delete files are supported only for format version 3 tables, where row-level deletes +/// are deletion vectors. A deletion vector committed for a data file added by the same +/// snapshot is attached to the file's AddedRowsScanTask; one committed for a data file +/// that already existed produces a DeletedRowsScanTask; and one removed together with +/// its data file becomes an existing delete of the file's DeletedDataFileScanTask. +/// Position delete files and equality delete files are not supported: planning fails +/// with NotSupported when any changelog snapshot carries one that is live or that a +/// changelog snapshot removed. class ICEBERG_EXPORT IncrementalChangelogScan : public IncrementalScan { public: diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index d04e3385b6..3fd9b51a6d 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -290,6 +290,41 @@ int64_t ChangelogScanTask::estimated_row_count() const { return data_file_->record_count; } +// DeletedRowsScanTask implementation + +DeletedRowsScanTask::DeletedRowsScanTask( + int32_t change_ordinal, int64_t commit_snapshot_id, + std::shared_ptr data_file, + std::vector> added_deletes, + std::vector> existing_deletes, + std::shared_ptr residual_filter) + : ChangelogScanTask(change_ordinal, commit_snapshot_id, std::move(data_file), + std::move(added_deletes), std::move(residual_filter)), + existing_deletes_(std::move(existing_deletes)) {} + +int64_t DeletedRowsScanTask::size_bytes() const { + int64_t total_size = ChangelogScanTask::size_bytes(); + for (const auto& delete_file : existing_deletes_) { + total_size += ContentFileUtil::ContentSizeInBytes(*delete_file); + } + return total_size; +} + +int32_t DeletedRowsScanTask::files_count() const { + return ChangelogScanTask::files_count() + + static_cast(existing_deletes_.size()); +} + +// The record count of a deletion vector is its cardinality, which bounds the number of +// rows this task can produce. +int64_t DeletedRowsScanTask::estimated_row_count() const { + int64_t deleted_rows = 0; + for (const auto& delete_file : delete_files_) { + deleted_rows += delete_file->record_count; + } + return std::min(deleted_rows, data_file_->record_count); +} + // Generic template implementation for Make template Result>> TableScanBuilder::Make( diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h index 1307089e39..97f164b9b6 100644 --- a/src/iceberg/table_scan.h +++ b/src/iceberg/table_scan.h @@ -254,6 +254,64 @@ class ICEBERG_EXPORT DeletedDataFileScanTask : public ChangelogScanTask { } }; +/// \brief A scan task for deletes generated by adding deletion vectors to the table. +/// +/// Suppose snapshot S1 contains data files F1 and F2. Then snapshot S2 adds a deletion +/// vector, DV1, that deletes records from F2, and snapshot S3 replaces DV1 with DV2 that +/// deletes further records from F2. A scan for changes from S2 to S3 (inclusive) should +/// include the following tasks: +/// - DeletedRowsScanTask(file=F2, added-deletes=[DV1], existing-deletes=[], snapshot=S2) +/// - DeletedRowsScanTask(file=F2, added-deletes=[DV2], existing-deletes=[DV1], +/// snapshot=S3) +/// +/// Readers consuming these tasks should produce the records that are deleted by the added +/// deletes but not by the existing deletes, with metadata like change ordinal and commit +/// snapshot ID. +class ICEBERG_EXPORT DeletedRowsScanTask : public ChangelogScanTask { + public: + /// \brief Construct a DeletedRowsScanTask. + /// + /// \param change_ordinal Position in the changelog order (0-based). + /// \param commit_snapshot_id The snapshot ID that committed the added deletes. + /// \param data_file The data file containing the deleted rows. + /// \param added_deletes Delete files committed by the snapshot for this data file. + /// \param existing_deletes Delete files that applied to this data file before the + /// snapshot. + /// \param residual_filter Optional residual filter to apply after reading. + DeletedRowsScanTask(int32_t change_ordinal, int64_t commit_snapshot_id, + std::shared_ptr data_file, + std::vector> added_deletes, + std::vector> existing_deletes, + std::shared_ptr residual_filter = nullptr); + + ChangelogOperation operation() const override { return ChangelogOperation::kDelete; } + + int64_t size_bytes() const override; + int32_t files_count() const override; + int64_t estimated_row_count() const override; + + /// \brief The data file containing the deleted rows. + const std::shared_ptr& data_file() const { return data_file_; } + + /// \brief Delete files added by the commit snapshot that apply to the data file. + /// + /// Records removed by these delete files appear as deletes in the changelog. + const std::vector>& added_deletes() const { + return delete_files_; + } + + /// \brief Delete files that existed before the commit snapshot and must be applied + /// prior to determining which records are deleted by added_deletes(). + /// + /// Records removed by these delete files do not appear in the changelog. + const std::vector>& existing_deletes() const { + return existing_deletes_; + } + + private: + std::vector> existing_deletes_; +}; + namespace internal { // Internal table scan context used by different scan implementations. diff --git a/src/iceberg/test/incremental_changelog_scan_test.cc b/src/iceberg/test/incremental_changelog_scan_test.cc index e8b3740447..1afef55763 100644 --- a/src/iceberg/test/incremental_changelog_scan_test.cc +++ b/src/iceberg/test/incremental_changelog_scan_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include @@ -28,6 +29,7 @@ #include #include "iceberg/expression/expressions.h" +#include "iceberg/manifest/manifest_entry.h" #include "iceberg/snapshot.h" #include "iceberg/table_scan.h" #include "iceberg/test/scan_test_base.h" @@ -43,11 +45,19 @@ const std::string& TaskFilePath(const std::shared_ptr& task) if (auto deleted = std::dynamic_pointer_cast(task)) { return deleted->data_file()->file_path; } + if (auto deleted_rows = std::dynamic_pointer_cast(task)) { + return deleted_rows->data_file()->file_path; + } static const std::string empty_path; return empty_path; } +std::vector FilePaths(const std::vector>& files) { + return files | std::views::transform([](const auto& file) { return file->file_path; }) | + std::ranges::to>(); +} + /// \brief Sort changelog scan tasks for deterministic ordering. /// Sorts by change_ordinal, then by operation type name, then by file path. template @@ -67,7 +77,100 @@ void SortTasks(std::vector>& tasks) { } // namespace -class IncrementalChangelogScanTest : public ScanTestBase {}; +class IncrementalChangelogScanTest : public ScanTestBase { + protected: + std::shared_ptr MakeDeleteFile( + DataFile::Content content, FileFormatType format, const std::string& path, + std::optional referenced_data_file = std::nullopt, + PartitionValues partition = PartitionValues(std::vector{}), + std::shared_ptr spec = nullptr) { + auto effective_spec = spec ? spec : unpartitioned_spec_; + DataFile file{ + .content = content, + .file_path = path, + .file_format = format, + .partition = std::move(partition), + .record_count = 1, + .file_size_in_bytes = 10, + .partition_spec_id = effective_spec->spec_id(), + }; + if (content == DataFile::Content::kEqualityDeletes) { + file.equality_ids = {1}; + } + if (format == FileFormatType::kPuffin) { + file.referenced_data_file = std::move(referenced_data_file); + file.content_offset = 4L; + file.content_size_in_bytes = 6L; + } + return std::make_shared(std::move(file)); + } + + std::shared_ptr MakeDV( + const std::string& path, const std::string& referenced_data_file, + PartitionValues partition = PartitionValues(std::vector{}), + std::shared_ptr spec = nullptr) { + return MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kPuffin, + path, referenced_data_file, std::move(partition), + std::move(spec)); + } + + std::vector ManifestsOf(const Snapshot& snapshot) { + SnapshotReader reader(&snapshot); + auto manifests = reader.Manifests(file_io_); + EXPECT_THAT(manifests, IsOk()); + if (!manifests.has_value()) { + return {}; + } + return {manifests->begin(), manifests->end()}; + } + + std::shared_ptr MakeSnapshot(int8_t format_version, int64_t snapshot_id, + int64_t parent_snapshot_id, + int64_t sequence_number, + const std::vector& manifests, + const std::string& operation) { + auto manifest_list = WriteManifestList( + format_version, snapshot_id, parent_snapshot_id, sequence_number, manifests); + return std::make_shared(Snapshot{ + .snapshot_id = snapshot_id, + .parent_snapshot_id = parent_snapshot_id, + .sequence_number = sequence_number, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L + sequence_number * 1000), + .manifest_list = manifest_list, + .summary = {{"operation", operation}}, + .schema_id = schema_->schema_id(), + }); + } + + std::shared_ptr MakeMetadata( + const std::vector>& snapshots, + std::shared_ptr default_spec = nullptr) { + int64_t current_snapshot_id = snapshots.back()->snapshot_id; + return MakeTableMetadata(snapshots, current_snapshot_id, + {{"main", std::make_shared(SnapshotRef{ + .snapshot_id = current_snapshot_id, + .retention = SnapshotRef::Branch{}})}}, + std::move(default_spec)); + } + + Result>> PlanChangelog( + std::shared_ptr metadata, std::optional from_snapshot_id, + int64_t to_snapshot_id, std::shared_ptr filter = nullptr) { + ICEBERG_ASSIGN_OR_RAISE( + auto builder, MakeScanBuilder(std::move(metadata))); + if (from_snapshot_id.has_value()) { + builder->FromSnapshot(from_snapshot_id.value()); + } + builder->ToSnapshot(to_snapshot_id); + if (filter != nullptr) { + builder->Filter(std::move(filter)); + } + ICEBERG_ASSIGN_OR_RAISE(auto scan, builder->Build()); + ICEBERG_ASSIGN_OR_RAISE(auto tasks, scan->PlanFiles()); + SortTasks(tasks); + return tasks; + } +}; TEST_P(IncrementalChangelogScanTest, DataFilters) { auto version = GetParam(); @@ -554,68 +657,452 @@ TEST_P(IncrementalChangelogScanTest, PlanDeletedRowLineage) { EXPECT_EQ(deleted->commit_snapshot_id(), 2000L); } -TEST_P(IncrementalChangelogScanTest, DeleteFilesAreNotSupported) { +TEST_P(IncrementalChangelogScanTest, DeletionVectorOnExistingFile) { auto version = GetParam(); - if (version < 2) { - GTEST_SKIP() << "Delete files only exist in format version 2+"; + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; } auto snapshot_a = MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet", "/path/to/file_b.parquet"}); - // Create a snapshot with delete files (positional deletes) - // This simulates table.newRowDelta().addDeletes(FILE_A_DELETES).commit() - std::vector data_entries; - auto file_a = MakeDataFile("/path/to/file_a.parquet"); - data_entries.push_back(MakeEntry(ManifestStatus::kExisting, 1000L, 1L, file_a)); - auto file_b = MakeDataFile("/path/to/file_b.parquet"); - data_entries.push_back(MakeEntry(ManifestStatus::kExisting, 1000L, 1L, file_b)); - auto data_manifest = WriteDataManifest(version, 2000L, std::move(data_entries)); - - // Create a delete file entry - auto delete_file = std::make_shared(DataFile{ - .content = DataFile::Content::kPositionDeletes, - .file_path = "/path/to/file_a_deletes.parquet", - .file_format = FileFormatType::kParquet, - .partition = PartitionValues(std::vector{}), - .record_count = 1, - .file_size_in_bytes = 10, - .sort_order_id = 0, - .partition_spec_id = unpartitioned_spec_->spec_id(), - }); - std::vector delete_entries; - delete_entries.push_back(MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)); - auto delete_manifest = - WriteDeleteManifest(version, 2000L, std::move(delete_entries), unpartitioned_spec_); + auto dv_a = MakeDV("/path/to/dv_a.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); - auto manifest_list = - WriteManifestList(version, 2000L, 1000L, 2L, {data_manifest, delete_manifest}); - TimePointMs timestamp_ms = TimePointMsFromUnixMs(1609459200000L + 2000); - auto snapshot_b = std::make_shared(Snapshot{ - .snapshot_id = 2000L, - .parent_snapshot_id = 1000L, - .sequence_number = 2L, - .timestamp_ms = timestamp_ms, - .manifest_list = manifest_list, - .summary = {{"operation", "delete"}}, - .schema_id = schema_->schema_id(), - }); + auto metadata = MakeMetadata({snapshot_a, snapshot_b}); - auto metadata = MakeTableMetadata( - {snapshot_a, snapshot_b}, 2000L, - {{"main", std::make_shared(SnapshotRef{ - .snapshot_id = 2000L, .retention = SnapshotRef::Branch{}})}}); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 1000L, 2000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->change_ordinal(), 0); + EXPECT_EQ(task->commit_snapshot_id(), 2000L); + EXPECT_EQ(task->operation(), ChangelogOperation::kDelete); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_a.parquet"); + EXPECT_THAT(FilePaths(task->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a.puffin")); + EXPECT_TRUE(task->existing_deletes().empty()); + EXPECT_EQ(task->files_count(), 2); + EXPECT_EQ(task->size_bytes(), 16); + EXPECT_EQ(task->estimated_row_count(), 1); + + ICEBERG_UNWRAP_OR_FAIL(auto all_tasks, PlanChangelog(metadata, std::nullopt, 2000L)); + ASSERT_EQ(all_tasks.size(), 3); + EXPECT_EQ(all_tasks[0]->change_ordinal(), 0); + EXPECT_EQ(all_tasks[0]->operation(), ChangelogOperation::kInsert); + EXPECT_EQ(all_tasks[1]->change_ordinal(), 0); + EXPECT_EQ(all_tasks[1]->operation(), ChangelogOperation::kInsert); + EXPECT_EQ(all_tasks[2]->change_ordinal(), 1); + EXPECT_NE(std::dynamic_pointer_cast(all_tasks[2]), nullptr); +} - ICEBERG_UNWRAP_OR_FAIL(auto builder, - MakeScanBuilder(metadata)); - builder->ToSnapshot(2000L); - ICEBERG_UNWRAP_OR_FAIL(auto scan, builder->Build()); - EXPECT_THAT(scan->PlanFiles(), - ::testing::AllOf( - IsError(ErrorKind::kNotSupported), - HasErrorMessage( - "Delete files are currently not supported in changelog scans"))); +TEST_P(IncrementalChangelogScanTest, ReplacedDeletionVector) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto dv_a1 = MakeDV("/path/to/dv_a1.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto dv_a2 = MakeDV("/path/to/dv_a2.puffin", "/path/to/file_a.parquet"); + auto manifests_c = ManifestsOf(*snapshot_a); + manifests_c.push_back( + WriteDeleteManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, dv_a2), + MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 2000L, 3000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->change_ordinal(), 0); + EXPECT_EQ(task->commit_snapshot_id(), 3000L); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_a.parquet"); + EXPECT_THAT(FilePaths(task->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a2.puffin")); + EXPECT_THAT(FilePaths(task->existing_deletes()), + ::testing::ElementsAre("/path/to/dv_a1.puffin")); + EXPECT_EQ(task->files_count(), 3); + EXPECT_EQ(task->size_bytes(), 22); + + ICEBERG_UNWRAP_OR_FAIL(auto all_tasks, PlanChangelog(metadata, 1000L, 3000L)); + ASSERT_EQ(all_tasks.size(), 2); + auto first = std::dynamic_pointer_cast(all_tasks[0]); + ASSERT_NE(first, nullptr); + EXPECT_EQ(first->change_ordinal(), 0); + EXPECT_EQ(first->commit_snapshot_id(), 2000L); + EXPECT_THAT(FilePaths(first->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a1.puffin")); + EXPECT_TRUE(first->existing_deletes().empty()); + auto second = std::dynamic_pointer_cast(all_tasks[1]); + ASSERT_NE(second, nullptr); + EXPECT_EQ(second->change_ordinal(), 1); + EXPECT_EQ(second->commit_snapshot_id(), 3000L); + EXPECT_THAT(FilePaths(second->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a2.puffin")); + EXPECT_THAT(FilePaths(second->existing_deletes()), + ::testing::ElementsAre("/path/to/dv_a1.puffin")); +} + +TEST_P(IncrementalChangelogScanTest, DeletionVectorForAddedFile) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto file_c = MakeDataFile("/path/to/file_c.parquet"); + auto dv_c = MakeDV("/path/to/dv_c.puffin", "/path/to/file_c.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDataManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, file_c)})); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_c)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "overwrite"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 1000L, 2000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->change_ordinal(), 0); + EXPECT_EQ(task->commit_snapshot_id(), 2000L); + EXPECT_EQ(task->operation(), ChangelogOperation::kInsert); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_c.parquet"); + EXPECT_THAT(FilePaths(task->delete_files()), + ::testing::ElementsAre("/path/to/dv_c.puffin")); + EXPECT_EQ(task->files_count(), 2); +} + +TEST_P(IncrementalChangelogScanTest, DeletedFileWithDeletionVector) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, + {"/path/to/file_a.parquet", "/path/to/file_b.parquet"}); + + auto dv_a = MakeDV("/path/to/dv_a.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + std::vector manifests_c; + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet")), + MakeEntry(ManifestStatus::kExisting, 1000L, 1L, + MakeDataFile("/path/to/file_b.parquet"))})); + manifests_c.push_back(WriteDeleteManifest( + version, 3000L, {MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, dv_a)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 2000L, 3000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->change_ordinal(), 0); + EXPECT_EQ(task->commit_snapshot_id(), 3000L); + EXPECT_EQ(task->operation(), ChangelogOperation::kDelete); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_a.parquet"); + EXPECT_THAT(FilePaths(task->existing_deletes()), + ::testing::ElementsAre("/path/to/dv_a.puffin")); +} + +TEST_P(IncrementalChangelogScanTest, DeletionVectorsRespectDataFilter) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto partition_a = PartitionValues({Literal::Int(8)}); + auto partition_b = PartitionValues({Literal::Int(1)}); + auto snapshot_a = + MakeAppendSnapshotWithPartitionValues(version, 1000L, std::nullopt, 1L, + {{"/path/to/file_a.parquet", partition_a}, + {"/path/to/file_b.parquet", partition_b}}, + partitioned_spec_); + + auto dv_a = MakeDV("/path/to/dv_a.puffin", "/path/to/file_a.parquet", partition_a, + partitioned_spec_); + auto dv_b = MakeDV("/path/to/dv_b.puffin", "/path/to/file_b.parquet", partition_b, + partitioned_spec_); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back( + WriteDeleteManifest(version, 2000L, + {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a), + MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_b)}, + partitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b}, partitioned_spec_); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, + PlanChangelog(metadata, 1000L, 2000L, + Expressions::Equal("data", Literal::String("k")))); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_b.parquet"); + EXPECT_THAT(FilePaths(task->added_deletes()), + ::testing::ElementsAre("/path/to/dv_b.puffin")); +} + +TEST_P(IncrementalChangelogScanTest, DeletionVectorsOutsideRangeAreIgnored) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto dv_a = MakeDV("/path/to/dv_a.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, + MakeDataFile("/path/to/file_b.parquet"))})); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "append"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 2000L, 3000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->commit_snapshot_id(), 3000L); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_b.parquet"); + EXPECT_TRUE(task->delete_files().empty()); +} + +TEST_P(IncrementalChangelogScanTest, DeleteFilesRequireFormatVersion3) { + auto version = GetParam(); + if (version != 2) { + GTEST_SKIP() << "Delete files exist in format version 2+ and are supported in 3+"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto delete_file = + MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b}); + + EXPECT_THAT(PlanChangelog(metadata, std::nullopt, 2000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Delete files are only supported in " + "changelog scans of format version 3"))); +} + +TEST_P(IncrementalChangelogScanTest, PositionDeleteFilesAreNotSupported) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto delete_file = + MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b}); + + EXPECT_THAT(PlanChangelog(metadata, std::nullopt, 2000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Position delete files are not supported " + "in changelog scans"))); +} + +TEST_P(IncrementalChangelogScanTest, EqualityDeleteFilesAreNotSupported) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto delete_file = + MakeDeleteFile(DataFile::Content::kEqualityDeletes, FileFormatType::kParquet, + "/path/to/eq_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b}); + + EXPECT_THAT(PlanChangelog(metadata, std::nullopt, 2000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Equality delete files are not supported " + "in changelog scans"))); +} + +TEST_P(IncrementalChangelogScanTest, PositionDeleteFilesOutsideRangeAreRejected) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto delete_file = + MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, + MakeDataFile("/path/to/file_b.parquet"))})); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "append"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + EXPECT_THAT(PlanChangelog(metadata, 2000L, 3000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Position delete files are not supported " + "in changelog scans"))); +} + +TEST_P(IncrementalChangelogScanTest, EqualityDeleteFilesOutsideRangeAreRejected) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto delete_file = + MakeDeleteFile(DataFile::Content::kEqualityDeletes, FileFormatType::kParquet, + "/path/to/eq_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, + MakeDataFile("/path/to/file_b.parquet"))})); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "append"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + EXPECT_THAT(PlanChangelog(metadata, 2000L, 3000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Equality delete files are not supported " + "in changelog scans"))); +} + +TEST_P(IncrementalChangelogScanTest, DeleteFilesRemovedBeforeRangeAreIgnored) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto delete_file = + MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto dv_a = MakeDV("/path/to/dv_a.puffin", "/path/to/file_a.parquet"); + auto manifests_c = ManifestsOf(*snapshot_a); + manifests_c.push_back( + WriteDeleteManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, dv_a), + MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto manifests_d = ManifestsOf(*snapshot_c); + manifests_d.push_back( + WriteDataManifest(version, 4000L, + {MakeEntry(ManifestStatus::kAdded, 4000L, 4L, + MakeDataFile("/path/to/file_b.parquet"))})); + auto snapshot_d = MakeSnapshot(version, 4000L, 3000L, 4L, manifests_d, "append"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c, snapshot_d}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 3000L, 4000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->commit_snapshot_id(), 4000L); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_b.parquet"); + EXPECT_TRUE(task->delete_files().empty()); + + EXPECT_THAT(PlanChangelog(metadata, 2000L, 4000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Position delete files are not supported " + "in changelog scans"))); } INSTANTIATE_TEST_SUITE_P(IncrementalChangelogScanVersions, IncrementalChangelogScanTest, From d471670792514cfcb55303b6d55835df31e3e683 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Sat, 10 Oct 2026 17:23:44 +0800 Subject: [PATCH 2/2] fix(scan): track deletion vector history in changelog planning - Attach the deletion vector that was live before a snapshot, not only the one it removed, so a vector left behind with its deleted data file is still an existing delete. - Reject position and equality delete files only when they apply to a reported data file, and prune delete manifests by the scan filter. - Subtract existing deletes from DeletedRowsScanTask::estimated_row_count. - Read each data manifest once when planning deleted rows instead of once per snapshot. - Share the stats and residual handling of changelog task planning, and stop copying the scan context into the planner. Co-Authored-By: Claude Opus 5.5 --- src/iceberg/changelog_dv_planner_internal.cc | 344 ++++++++++-------- src/iceberg/changelog_dv_planner_internal.h | 146 ++++---- src/iceberg/incremental_changelog_scan.cc | 298 ++++++++++----- src/iceberg/incremental_changelog_scan.h | 8 +- src/iceberg/table_scan.cc | 11 +- .../test/incremental_changelog_scan_test.cc | 246 ++++++++++++- 6 files changed, 714 insertions(+), 339 deletions(-) diff --git a/src/iceberg/changelog_dv_planner_internal.cc b/src/iceberg/changelog_dv_planner_internal.cc index 6496595e27..d895335687 100644 --- a/src/iceberg/changelog_dv_planner_internal.cc +++ b/src/iceberg/changelog_dv_planner_internal.cc @@ -19,210 +19,234 @@ #include "iceberg/changelog_dv_planner_internal.h" -#include +#include #include -#include "iceberg/expression/residual_evaluator.h" +#include "iceberg/expression/manifest_evaluator.h" +#include "iceberg/expression/projections.h" #include "iceberg/manifest/manifest_entry.h" -#include "iceberg/manifest/manifest_group.h" #include "iceberg/manifest/manifest_reader.h" -#include "iceberg/util/content_file_util.h" +#include "iceberg/partition_spec.h" #include "iceberg/util/executor_util_internal.h" #include "iceberg/util/macros.h" namespace iceberg { -namespace { - -Status ValidateChangelogDeleteFile(const DataFile& file) { - if (file.content == DataFile::Content::kEqualityDeletes) { - return NotSupported("Equality delete files are not supported in changelog scans: {}", - file.file_path); - } - if (!file.IsDeletionVector()) { - return NotSupported("Position delete files are not supported in changelog scans: {}", - file.file_path); - } - ICEBERG_PRECHECK(file.referenced_data_file.has_value(), - "Deletion vector {} does not reference a data file", file.file_path); - return {}; -} - -std::vector> FindDeletionVector( - const std::unordered_map>& dvs_by_path, - const std::string& data_file_path) { - auto it = dvs_by_path.find(data_file_path); - if (it == dvs_by_path.end()) { - return {}; - } - return {it->second}; -} - -} // namespace - -DeletionVectorChangelogPlanner::DeletionVectorChangelogPlanner( - std::shared_ptr io, std::shared_ptr schema, - std::unordered_map> specs_by_id, - internal::TableScanContext context, std::vector scan_columns, - std::shared_ptr filter, - std::unordered_map dvs_by_snapshot) - : io_(std::move(io)), - schema_(std::move(schema)), - specs_by_id_(std::move(specs_by_id)), - context_(std::move(context)), - scan_columns_(std::move(scan_columns)), - filter_(std::move(filter)), - dvs_by_snapshot_(std::move(dvs_by_snapshot)) {} - Result> DeletionVectorChangelogPlanner::Make( std::shared_ptr io, std::shared_ptr schema, - std::unordered_map> specs_by_id, - const internal::TableScanContext& context, std::vector scan_columns, - std::shared_ptr filter, const std::vector& delete_manifests, - const std::unordered_set& snapshot_ids) { + const std::unordered_map>& specs_by_id, + std::shared_ptr filter, bool case_sensitive, OptionalExecutor executor, + const std::vector& delete_manifests, + const std::unordered_map& sequence_numbers) { ICEBERG_ASSIGN_OR_RAISE( auto entries, ParallelCollect( - context.plan_executor, delete_manifests, + executor, delete_manifests, [&](const ManifestFile& manifest) -> Result> { + auto spec_it = specs_by_id.find(manifest.partition_spec_id); + ICEBERG_CHECK(spec_it != specs_by_id.end(), + "Partition spec ID {} not found when reading delete manifest", + manifest.partition_spec_id); + std::shared_ptr partition_filter; + if (filter != nullptr) { + auto projector = + Projections::Inclusive(*spec_it->second, *schema, case_sensitive); + ICEBERG_ASSIGN_OR_RAISE(partition_filter, projector->Project(filter)); + ICEBERG_ASSIGN_OR_RAISE( + auto evaluator, + ManifestEvaluator::MakePartitionFilter( + partition_filter, spec_it->second, *schema, case_sensitive)); + ICEBERG_ASSIGN_OR_RAISE(bool may_match, evaluator->Evaluate(manifest)); + if (!may_match) { + return std::vector{}; + } + } + ICEBERG_ASSIGN_OR_RAISE( auto reader, ManifestReader::Make(manifest, io, schema, specs_by_id)); + if (partition_filter != nullptr) { + reader->FilterPartitions(partition_filter).CaseSensitive(case_sensitive); + } reader->TryDropStats(); ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); - auto in_range = [&snapshot_ids](const ManifestEntry& entry) { - return entry.snapshot_id.has_value() && - snapshot_ids.contains(entry.snapshot_id.value()); - }; - for (const auto& entry : entries) { - ICEBERG_PRECHECK(entry.data_file != nullptr, - "Invalid manifest entry with missing delete file"); - // A delete file removed before the range no longer applies to any data - // file, so it cannot affect the changelog. Every other delete file can: a - // live one applies to data files whose rows the changelog reports, and one - // removed by a changelog snapshot was applied to the rows that snapshot - // reports as deleted. Those must be deletion vectors. - if (entry.status == ManifestStatus::kDeleted && !in_range(entry)) { - continue; - } - ICEBERG_RETURN_UNEXPECTED(ValidateChangelogDeleteFile(*entry.data_file)); - } - std::erase_if(entries, [&in_range](const ManifestEntry& entry) { - return entry.status == ManifestStatus::kExisting || !in_range(entry); + // A delete file removed before the range no longer applies to any data + // file. One removed by a snapshot in the range applied until then. + std::erase_if(entries, [&](const ManifestEntry& entry) { + return entry.status == ManifestStatus::kDeleted && + !(entry.snapshot_id.has_value() && + sequence_numbers.contains(entry.snapshot_id.value())); }); return entries; })); - std::unordered_map dvs_by_snapshot; + auto planner = std::unique_ptr( + new DeletionVectorChangelogPlanner()); for (auto& entry : entries) { - const int64_t snapshot_id = entry.snapshot_id.value(); - auto& dvs = dvs_by_snapshot[snapshot_id]; - auto& dvs_by_path = entry.status == ManifestStatus::kAdded ? dvs.added : dvs.removed; - const std::string& data_file_path = entry.data_file->referenced_data_file.value(); - auto [it, inserted] = - dvs_by_path.try_emplace(data_file_path, std::move(entry.data_file)); + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Invalid manifest entry with missing delete file"); + const DataFile& file = *entry.data_file; + // The file sequence number belongs to the snapshot that committed the file, unlike + // the data sequence number, which a rewrite can carry over from an older file. ICEBERG_PRECHECK( - inserted, "Snapshot {} {} multiple deletion vectors for {}", snapshot_id, - entry.status == ManifestStatus::kAdded ? "added" : "removed", data_file_path); + entry.file_sequence_number.has_value() && entry.sequence_number.has_value(), + "Missing sequence number from delete file {}", file.file_path); + ICEBERG_PRECHECK(file.partition_spec_id.has_value(), + "Missing partition spec ID from delete file {}", file.file_path); + + if (file.content == DataFile::Content::kEqualityDeletes) { + auto spec_it = specs_by_id.find(file.partition_spec_id.value()); + ICEBERG_CHECK(spec_it != specs_by_id.end(), + "Partition spec ID {} not found for delete file {}", + file.partition_spec_id.value(), file.file_path); + KeepLatest(spec_it->second->IsUnpartitioned() + ? planner->global_equality_deletes_ + : LatestFor(planner->equality_deletes_by_partition_, file), + entry.sequence_number.value(), file.file_path); + continue; + } + + if (!file.IsDeletionVector()) { + KeepLatest( + file.referenced_data_file.has_value() + ? planner->position_deletes_by_path_[file.referenced_data_file.value()] + : LatestFor(planner->position_deletes_by_partition_, file), + entry.sequence_number.value(), file.file_path); + continue; + } + + ICEBERG_PRECHECK(file.referenced_data_file.has_value(), + "Deletion vector {} does not reference a data file", file.file_path); + // The same vector appears once per manifest that lists it, for example as added in + // the manifest of the snapshot that committed it and as deleted in the manifest of + // the snapshot that removed it. + auto& versions = planner->dvs_by_path_[file.referenced_data_file.value()]; + auto version_it = std::ranges::find_if(versions, [&file](const auto& version) { + return version.file->file_path == file.file_path && + version.file->content_offset == file.content_offset; + }); + if (version_it == versions.end()) { + versions.push_back({.file = entry.data_file, + .added_sequence_number = entry.file_sequence_number.value()}); + version_it = std::prev(versions.end()); + } + if (entry.status == ManifestStatus::kDeleted) { + version_it->removed_sequence_number = + sequence_numbers.at(entry.snapshot_id.value()); + } + } + + for (const auto& [data_file_path, versions] : planner->dvs_by_path_) { + for (auto it = versions.begin(); it != versions.end(); ++it) { + ICEBERG_PRECHECK( + std::ranges::find(std::next(it), versions.end(), it->added_sequence_number, + &DeletionVectorVersion::added_sequence_number) == + versions.end(), + "Snapshot with sequence number {} added multiple deletion vectors for {}", + it->added_sequence_number, data_file_path); + } } - return std::unique_ptr( - new DeletionVectorChangelogPlanner( - std::move(io), std::move(schema), std::move(specs_by_id), context, - std::move(scan_columns), std::move(filter), std::move(dvs_by_snapshot))); + return planner; } -std::vector> DeletionVectorChangelogPlanner::AddedDeletes( - int64_t snapshot_id, const std::string& data_file_path) const { - auto it = dvs_by_snapshot_.find(snapshot_id); - if (it == dvs_by_snapshot_.end()) { - return {}; +std::optional& +DeletionVectorChangelogPlanner::LatestFor( + PartitionMap>& deletes_by_partition, + const DataFile& file) { + const int32_t spec_id = file.partition_spec_id.value(); + if (!deletes_by_partition.contains(spec_id, file.partition)) { + deletes_by_partition.put(spec_id, file.partition, std::nullopt); } - return FindDeletionVector(it->second.added, data_file_path); + return deletes_by_partition.get(spec_id, file.partition)->get(); } -std::vector> DeletionVectorChangelogPlanner::RemovedDeletes( - int64_t snapshot_id, const std::string& data_file_path) const { - auto it = dvs_by_snapshot_.find(snapshot_id); - if (it == dvs_by_snapshot_.end()) { - return {}; +void DeletionVectorChangelogPlanner::KeepLatest(std::optional& latest, + int64_t sequence_number, + const std::string& file_path) { + if (!latest.has_value() || latest->sequence_number < sequence_number) { + latest = LatestDeleteFile{.sequence_number = sequence_number, .file_path = file_path}; } - return FindDeletionVector(it->second.removed, data_file_path); } -Result>> -DeletionVectorChangelogPlanner::PlanDeletedRows( - int64_t snapshot_id, int32_t change_ordinal, std::span data_manifests, - const std::unordered_set& added_data_file_paths) const { - auto dvs_it = dvs_by_snapshot_.find(snapshot_id); - if (dvs_it == dvs_by_snapshot_.end()) { - return std::vector>{}; +std::shared_ptr DeletionVectorChangelogPlanner::AddedDeletionVector( + int64_t sequence_number, const std::string& data_file_path) const { + auto it = dvs_by_path_.find(data_file_path); + if (it == dvs_by_path_.end()) { + return nullptr; } - const SnapshotDeletionVectors& dvs = dvs_it->second; + auto version_it = std::ranges::find(it->second, sequence_number, + &DeletionVectorVersion::added_sequence_number); + return version_it == it->second.end() ? nullptr : version_it->file; +} - std::unordered_set pending_paths; - for (const auto& [data_file_path, dv] : dvs.added) { - if (!added_data_file_paths.contains(data_file_path)) { - pending_paths.insert(data_file_path); +std::unordered_map> +DeletionVectorChangelogPlanner::AddedDeletionVectors( + const std::unordered_set& sequence_numbers) const { + std::unordered_map> added; + for (const auto& [data_file_path, versions] : dvs_by_path_) { + for (const auto& version : versions) { + if (sequence_numbers.contains(version.added_sequence_number)) { + added[data_file_path].push_back(version.added_sequence_number); + } } } - if (pending_paths.empty()) { - return std::vector>{}; - } + return added; +} - ICEBERG_ASSIGN_OR_RAISE( - auto manifest_group, - ManifestGroup::Make( - io_, schema_, specs_by_id_, - std::vector(data_manifests.begin(), data_manifests.end()), - /*delete_manifests=*/{})); - manifest_group->CaseSensitive(context_.case_sensitive) - .Select(scan_columns_) - .FilterData(filter_) - .FilterManifestEntries([&pending_paths](const ManifestEntry& entry) { - return entry.data_file != nullptr && - pending_paths.contains(entry.data_file->file_path); - }) - .IgnoreDeleted() - .ColumnsToKeepStats(context_.columns_to_keep_stats) - .PlanWith(context_.plan_executor); - if (context_.ignore_residuals) { - manifest_group->IgnoreResiduals(); +Result> DeletionVectorChangelogPlanner::ExistingDeletionVector( + int64_t sequence_number, const std::string& data_file_path) const { + auto it = dvs_by_path_.find(data_file_path); + if (it == dvs_by_path_.end()) { + return nullptr; } - - auto create_tasks_func = - [&](std::vector&& entries, - const TaskContext& ctx) -> Result>> { - std::vector> tasks; - tasks.reserve(entries.size()); - - for (auto& entry : entries) { - ICEBERG_PRECHECK(entry.data_file != nullptr, - "Invalid manifest entry with missing data file"); - - 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)); - const std::string& data_file_path = entry.data_file->file_path; - auto added_deletes = FindDeletionVector(dvs.added, data_file_path); - auto existing_deletes = FindDeletionVector(dvs.removed, data_file_path); - tasks.push_back(std::make_shared( - change_ordinal, snapshot_id, std::move(entry.data_file), - std::move(added_deletes), std::move(existing_deletes), std::move(residual))); + std::shared_ptr existing; + for (const auto& version : it->second) { + const bool live = version.added_sequence_number < sequence_number && + (!version.removed_sequence_number.has_value() || + version.removed_sequence_number.value() >= sequence_number); + if (!live) { + continue; } - return tasks; - }; + ICEBERG_PRECHECK(existing == nullptr, + "Multiple deletion vectors apply to {} before sequence number {}", + data_file_path, sequence_number); + existing = version.file; + } + return existing; +} - 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(task); - }) | - std::ranges::to(); +Status DeletionVectorChangelogPlanner::ValidateNoOtherDeletes( + int64_t data_sequence_number, const DataFile& data_file) const { + ICEBERG_PRECHECK(data_file.partition_spec_id.has_value(), + "Missing partition spec ID from data file {}", data_file.file_path); + const int32_t spec_id = data_file.partition_spec_id.value(); + auto in_partition = [&](const PartitionMap>& deletes) { + auto latest = deletes.get(spec_id, data_file.partition); + return latest.has_value() ? latest->get() : std::nullopt; + }; + auto for_path = position_deletes_by_path_.find(data_file.file_path); + + // An equality delete file applies to data files written before it, and a position + // delete file also to those written in the same commit. + for (const auto& latest : + {global_equality_deletes_, in_partition(equality_deletes_by_partition_)}) { + if (latest.has_value() && latest->sequence_number > data_sequence_number) { + return NotSupported( + "Equality delete files are not supported in changelog scans: {} applies to {}", + latest->file_path, data_file.file_path); + } + } + for (const auto& latest : + {in_partition(position_deletes_by_partition_), + for_path == position_deletes_by_path_.end() ? std::nullopt : for_path->second}) { + if (latest.has_value() && latest->sequence_number >= data_sequence_number) { + return NotSupported( + "Position delete files are not supported in changelog scans: {} applies to {}", + latest->file_path, data_file.file_path); + } + } + return {}; } } // namespace iceberg diff --git a/src/iceberg/changelog_dv_planner_internal.h b/src/iceberg/changelog_dv_planner_internal.h index 1f76cffb37..cad4f8d03f 100644 --- a/src/iceberg/changelog_dv_planner_internal.h +++ b/src/iceberg/changelog_dv_planner_internal.h @@ -24,7 +24,7 @@ #include #include -#include +#include #include #include #include @@ -32,8 +32,9 @@ #include "iceberg/manifest/manifest_list.h" #include "iceberg/result.h" -#include "iceberg/table_scan.h" #include "iceberg/type_fwd.h" +#include "iceberg/util/executor.h" +#include "iceberg/util/partition_value_util.h" namespace iceberg { @@ -41,81 +42,100 @@ namespace iceberg { /// /// Format version 3 stores row-level deletes as deletion vectors, with at most one live /// vector per data file, and a new vector replaces the previous one. The rows a snapshot -/// deleted are therefore the difference between the vector it added and the vector it -/// removed, which DeletedRowsScanTask carries as added and existing deletes. Position -/// delete files and equality delete files accumulate instead, so attributing their rows -/// to a snapshot would require reading every earlier delete file; they are rejected -/// wherever they could still apply to a reported row, whether or not a changelog -/// snapshot wrote them. +/// deleted are therefore the difference between the vector it added and the vector that +/// was live just before it, which DeletedRowsScanTask carries as added and existing +/// deletes. Snapshots are identified by sequence number, so the history of each vector is +/// the sequence number of the snapshot that added it and, if it was removed, of the +/// snapshot that removed it. +/// +/// Position delete files and equality delete files accumulate instead, so attributing +/// their rows to a snapshot would require reading every earlier delete file. A data file +/// that the changelog reports is rejected when one of them applies to it. class DeletionVectorChangelogPlanner { public: - /// \brief Index the deletion vectors added and removed by the changelog snapshots. + /// \brief Index the delete files that can apply to the rows of a changelog. /// - /// \param context Scan context that supplies filtering and planning options. - /// \param scan_columns Manifest columns to read for data files. - /// \param filter Row filter of the scan. - /// \param delete_manifests All delete manifests of the changelog snapshots. Only the - /// manifests written by those snapshots can hold the vectors, because removing a - /// vector rewrites its manifest, but a manifest untouched during the range can still - /// carry position delete files or equality delete files that apply to the reported - /// rows. Every manifest is therefore checked, and only entries written by the - /// changelog snapshots are indexed. - /// \param snapshot_ids IDs of the changelog snapshots. + /// \param io FileIO used to read the delete manifests. + /// \param schema Schema of the scan. + /// \param specs_by_id Partition specs of the table. + /// \param filter Row filter of the scan, used to prune delete manifests and entries by + /// partition. A data file outside the filter is never reported, so neither are the + /// delete files of its partition. + /// \param case_sensitive Whether the filter binds case-sensitively. + /// \param executor Executor used to read the delete manifests. + /// \param delete_manifests Delete manifests of every snapshot in the range, including + /// replace snapshots. A snapshot's own manifests are the only ones guaranteed to hold + /// the entries it removed, because later snapshots drop removed entries when they + /// rewrite a manifest. + /// \param sequence_numbers Sequence numbers of every snapshot in the range, keyed by + /// snapshot ID. A delete file removed by any other snapshot was removed before the + /// range and can no longer apply to a reported row. static Result> Make( std::shared_ptr io, std::shared_ptr schema, - std::unordered_map> specs_by_id, - const internal::TableScanContext& context, std::vector scan_columns, - std::shared_ptr filter, + const std::unordered_map>& specs_by_id, + std::shared_ptr filter, bool case_sensitive, OptionalExecutor executor, const std::vector& delete_manifests, - const std::unordered_set& snapshot_ids); + const std::unordered_map& sequence_numbers); + + /// \brief The deletion vector that the snapshot with the given sequence number added + /// for the data file, or nullptr. + std::shared_ptr AddedDeletionVector(int64_t sequence_number, + const std::string& data_file_path) const; - /// \brief The deletion vector that the snapshot committed for the data file, if any. - std::vector> AddedDeletes( - int64_t snapshot_id, const std::string& data_file_path) const; + /// \brief The deletion vector that applied to the data file just before the snapshot + /// with the given sequence number, or nullptr. + /// + /// This is the vector the snapshot replaced or removed, or one it left behind when it + /// removed the data file, so every row it deletes is excluded from the changelog. + Result> ExistingDeletionVector( + int64_t sequence_number, const std::string& data_file_path) const; - /// \brief The deletion vector that the snapshot removed from the data file, if any. - std::vector> RemovedDeletes( - int64_t snapshot_id, const std::string& data_file_path) const; + /// \brief Which of the snapshots with the given sequence numbers added a deletion + /// vector, keyed by the path of the data file the vector references. + std::unordered_map> AddedDeletionVectors( + const std::unordered_set& sequence_numbers) const; - /// \brief Plan DeletedRowsScanTasks for the deletion vectors a snapshot committed - /// against data files it did not add. + /// \brief Fail with NotSupported if a position delete file or an equality delete file + /// applies to the data file. /// - /// \param snapshot_id The changelog snapshot. - /// \param change_ordinal Position of the snapshot in the changelog order. - /// \param data_manifests All data manifests of the snapshot. The referenced data files - /// are located there because an earlier snapshot added them and a later snapshot in the - /// range may remove them again. - /// \param added_data_file_paths Data files added by the snapshot, whose deletion - /// vectors belong to their AddedRowsScanTask instead. - Result>> PlanDeletedRows( - int64_t snapshot_id, int32_t change_ordinal, std::span data_manifests, - const std::unordered_set& added_data_file_paths) const; + /// \param data_sequence_number Data sequence number of the data file. + /// \param data_file The data file that the changelog reports. + Status ValidateNoOtherDeletes(int64_t data_sequence_number, + const DataFile& data_file) const; private: - using DeletionVectorsByPath = - std::unordered_map>; - - // Deletion vectors committed and removed by one changelog snapshot, keyed by the path - // of the data file they reference. - struct SnapshotDeletionVectors { - DeletionVectorsByPath added; - DeletionVectorsByPath removed; + // One deletion vector and the sequence numbers of the snapshots that added and + // removed it. + struct DeletionVectorVersion { + std::shared_ptr file; + int64_t added_sequence_number; + std::optional removed_sequence_number; }; - DeletionVectorChangelogPlanner( - std::shared_ptr io, std::shared_ptr schema, - std::unordered_map> specs_by_id, - internal::TableScanContext context, std::vector scan_columns, - std::shared_ptr filter, - std::unordered_map dvs_by_snapshot); - - std::shared_ptr io_; - std::shared_ptr schema_; - std::unordered_map> specs_by_id_; - internal::TableScanContext context_; - std::vector scan_columns_; - std::shared_ptr filter_; - std::unordered_map dvs_by_snapshot_; + // The delete file with the highest sequence number among those sharing a scope. A + // delete file applies to a data file based on sequence numbers alone once the scope + // matches, so the latest one decides whether any of them applies. + struct LatestDeleteFile { + int64_t sequence_number; + std::string file_path; + }; + + DeletionVectorChangelogPlanner() = default; + + static std::optional& LatestFor( + PartitionMap>& deletes_by_partition, + const DataFile& file); + static void KeepLatest(std::optional& latest, int64_t sequence_number, + const std::string& file_path); + + // Deletion vector history keyed by the path of the data file it references. + std::unordered_map> dvs_by_path_; + + std::optional global_equality_deletes_; + PartitionMap> equality_deletes_by_partition_; + PartitionMap> position_deletes_by_partition_; + std::unordered_map> + position_deletes_by_path_; }; } // namespace iceberg diff --git a/src/iceberg/incremental_changelog_scan.cc b/src/iceberg/incremental_changelog_scan.cc index 0395572bb7..1bbcc23952 100644 --- a/src/iceberg/incremental_changelog_scan.cc +++ b/src/iceberg/incremental_changelog_scan.cc @@ -19,6 +19,7 @@ #include "iceberg/incremental_changelog_scan.h" +#include #include #include #include @@ -38,6 +39,49 @@ namespace iceberg { +namespace { + +// Appends the changelog tasks for one data file of a changelog scan. +using ChangelogTaskFactory = + std::function residual, + std::vector>& tasks)>; + +// Plans the data files of the manifest group, keeping the stats and computing the +// residual that the scan requests before handing each file to make_tasks. +Result>> PlanChangelogTasks( + ManifestGroup& manifest_group, const ChangelogTaskFactory& make_tasks) { + auto create_tasks_func = + [&](std::vector&& entries, + const TaskContext& ctx) -> Result>> { + std::vector> tasks; + tasks.reserve(entries.size()); + + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Invalid manifest entry with missing data file"); + + 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)); + ICEBERG_RETURN_UNEXPECTED(make_tasks(std::move(entry), std::move(residual), tasks)); + } + 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(task); + }) | + std::ranges::to(); +} + +} // namespace + Result> IncrementalChangelogScan::Make( std::shared_ptr metadata, std::shared_ptr schema, std::shared_ptr io, internal::TableScanContext context) { @@ -91,6 +135,8 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl std::unordered_set snapshot_ids; std::unordered_map snapshot_ordinals; + std::unordered_set changelog_sequence_numbers; + std::unordered_map snapshot_ids_by_sequence_number; for (const auto& snapshot : changelog_snapshots) { ICEBERG_PRECHECK( std::cmp_less_equal(snapshot_ids.size(), std::numeric_limits::max()), @@ -98,27 +144,43 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl snapshot_ids.insert(snapshot.first->snapshot_id); snapshot_ordinals.try_emplace(snapshot.first->snapshot_id, static_cast(snapshot_ordinals.size())); + changelog_sequence_numbers.insert(snapshot.first->sequence_number); + snapshot_ids_by_sequence_number.try_emplace(snapshot.first->sequence_number, + snapshot.first->snapshot_id); } + // Data manifests written by the changelog snapshots hold their added and deleted data + // files; all data manifests of the changelog snapshots hold the data files that a + // deletion vector can reference. + std::vector added_data_manifests; std::vector data_manifests; - std::vector delete_manifests; std::unordered_set seen_manifest_paths; for (const auto& [snapshot, snapshot_reader] : changelog_snapshots) { ICEBERG_ASSIGN_OR_RAISE(auto snapshot_data_manifests, snapshot_reader->DataManifests(io_)); for (auto& manifest : snapshot_data_manifests) { + if (!seen_manifest_paths.insert(manifest.manifest_path).second) { + continue; + } 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); + snapshot_ids.contains(manifest.added_snapshot_id.value())) { + added_data_manifests.push_back(manifest); } + data_manifests.push_back(manifest); } - if (deletes_supported) { - // Every delete manifest is collected, not only those written in the range, so - // that position delete files and equality delete files carried in older manifests - // are rejected rather than silently left out of the changelog. + } + + // Delete manifests of every snapshot in the range, replace snapshots included, so that + // each removal of a deletion vector is seen in the manifest of the snapshot that made + // it. + std::vector delete_manifests; + std::unordered_map sequence_numbers; + if (deletes_supported) { + for (const auto& snapshot : ancestors_snapshots) { + sequence_numbers.emplace(snapshot->snapshot_id, snapshot->sequence_number); + SnapshotReader snapshot_reader(snapshot.get()); ICEBERG_ASSIGN_OR_RAISE(auto snapshot_delete_manifests, - snapshot_reader->DeleteManifests(io_)); + snapshot_reader.DeleteManifests(io_)); for (auto& manifest : snapshot_delete_manifests) { if (seen_manifest_paths.insert(manifest.manifest_path).second) { delete_manifests.push_back(manifest); @@ -126,7 +188,7 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl } } } - if (data_manifests.empty() && delete_manifests.empty()) { + if (added_data_manifests.empty() && delete_manifests.empty()) { return std::vector>{}; } @@ -137,120 +199,154 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl if (!delete_manifests.empty()) { ICEBERG_ASSIGN_OR_RAISE( dv_planner, DeletionVectorChangelogPlanner::Make( - io_, schema_, specs_by_id, context_, ScanColumns(), filter(), - delete_manifests, snapshot_ids)); + io_, schema_, specs_by_id, filter(), context_.case_sensitive, + context_.plan_executor, delete_manifests, sequence_numbers)); } - std::vector> tasks; - // Data files added by each snapshot. A deletion vector committed together with its data - // file belongs to the AddedRowsScanTask rather than to a DeletedRowsScanTask. - std::unordered_map> added_paths_by_snapshot; - - if (!data_manifests.empty()) { + auto make_manifest_group = + [&](std::vector manifests) -> Result> { ICEBERG_ASSIGN_OR_RAISE( auto manifest_group, - ManifestGroup::Make(io_, schema_, specs_by_id, std::move(data_manifests), + ManifestGroup::Make(io_, schema_, specs_by_id, std::move(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(); } + return manifest_group; + }; - auto create_tasks_func = - [&](std::vector&& entries, - const TaskContext& ctx) -> Result>> { - std::vector> tasks; - tasks.reserve(entries.size()); + std::vector> tasks; - for (auto& entry : entries) { - ICEBERG_PRECHECK(entry.snapshot_id.has_value() && entry.data_file, - "Invalid manifest entry with missing snapshot id or data file"); + if (!added_data_manifests.empty()) { + ICEBERG_ASSIGN_OR_RAISE(auto manifest_group, + make_manifest_group(std::move(added_data_manifests))); + manifest_group + ->FilterManifestEntries([&snapshot_ids](const ManifestEntry& entry) { + return entry.snapshot_id.has_value() && + snapshot_ids.contains(entry.snapshot_id.value()); + }) + .IgnoreExisting(); - 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"); + ICEBERG_ASSIGN_OR_RAISE( + tasks, + PlanChangelogTasks( + *manifest_group, + [&](ManifestEntry&& entry, std::shared_ptr residual, + std::vector>& planned) -> Status { + ICEBERG_PRECHECK(entry.snapshot_id.has_value(), + "Invalid manifest entry with missing snapshot id"); - int32_t change_ordinal = ordinal_it->second; + 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"); - 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); - } + int32_t change_ordinal = ordinal_it->second; - ICEBERG_ASSIGN_OR_RAISE(auto residual, - ctx.residuals->ResidualFor(entry.data_file->partition)); - const std::string& data_file_path = entry.data_file->file_path; - switch (entry.status) { - case ManifestStatus::kAdded: { - std::vector> deletes; - if (dv_planner) { - deletes = dv_planner->AddedDeletes(commit_snapshot_id, data_file_path); - added_paths_by_snapshot[commit_snapshot_id].insert(data_file_path); - } - tasks.push_back(std::make_shared( - change_ordinal, commit_snapshot_id, std::move(entry.data_file), - std::move(deletes), std::move(residual))); - break; - } - case ManifestStatus::kDeleted: { - std::vector> existing_deletes; - if (dv_planner) { - existing_deletes = - dv_planner->RemovedDeletes(commit_snapshot_id, data_file_path); - } - tasks.push_back(std::make_shared( - change_ordinal, commit_snapshot_id, std::move(entry.data_file), - std::move(existing_deletes), std::move(residual))); - break; - } - case ManifestStatus::kExisting: - return InvalidArgument("Unexpected entry status: EXISTING"); - } - } - return tasks; - }; - - ICEBERG_ASSIGN_OR_RAISE(auto data_file_tasks, - manifest_group->Plan(create_tasks_func)); - tasks = data_file_tasks | std::views::transform([](const auto& task) { - return std::static_pointer_cast(task); - }) | - std::ranges::to(); + std::vector> deletes; + if (dv_planner) { + ICEBERG_PRECHECK(entry.sequence_number.has_value(), + "Missing sequence number from data file {}", + entry.data_file->file_path); + ICEBERG_RETURN_UNEXPECTED(dv_planner->ValidateNoOtherDeletes( + entry.sequence_number.value(), *entry.data_file)); + const int64_t sequence_number = sequence_numbers.at(commit_snapshot_id); + const std::string& data_file_path = entry.data_file->file_path; + std::shared_ptr dv; + if (entry.status == ManifestStatus::kAdded) { + dv = dv_planner->AddedDeletionVector(sequence_number, data_file_path); + } else { + ICEBERG_ASSIGN_OR_RAISE(dv, dv_planner->ExistingDeletionVector( + sequence_number, data_file_path)); + } + if (dv != nullptr) { + deletes.push_back(std::move(dv)); + } + } + + switch (entry.status) { + case ManifestStatus::kAdded: + planned.push_back(std::make_shared( + change_ordinal, commit_snapshot_id, std::move(entry.data_file), + std::move(deletes), std::move(residual))); + return {}; + case ManifestStatus::kDeleted: + planned.push_back(std::make_shared( + change_ordinal, commit_snapshot_id, std::move(entry.data_file), + std::move(deletes), std::move(residual))); + return {}; + case ManifestStatus::kExisting: + return InvalidArgument("Unexpected entry status: EXISTING"); + } + return {}; + })); } - if (dv_planner) { - const std::unordered_set no_added_paths; - for (const auto& [snapshot, snapshot_reader] : changelog_snapshots) { - const int64_t snapshot_id = snapshot->snapshot_id; - auto added_paths_it = added_paths_by_snapshot.find(snapshot_id); - const auto& added_paths = added_paths_it == added_paths_by_snapshot.end() - ? no_added_paths - : added_paths_it->second; - ICEBERG_ASSIGN_OR_RAISE(auto snapshot_data_manifests, - snapshot_reader->DataManifests(io_)); - ICEBERG_ASSIGN_OR_RAISE( - auto deleted_rows_tasks, - dv_planner->PlanDeletedRows(snapshot_id, snapshot_ordinals.at(snapshot_id), - snapshot_data_manifests, added_paths)); - tasks.insert(tasks.end(), std::make_move_iterator(deleted_rows_tasks.begin()), - std::make_move_iterator(deleted_rows_tasks.end())); - } + if (dv_planner == nullptr) { + return tasks; + } + auto added_dvs = dv_planner->AddedDeletionVectors(changelog_sequence_numbers); + if (added_dvs.empty()) { + return tasks; } + ICEBERG_ASSIGN_OR_RAISE(auto manifest_group, + make_manifest_group(std::move(data_manifests))); + manifest_group + ->FilterManifestEntries([&added_dvs](const ManifestEntry& entry) { + return entry.data_file != nullptr && + added_dvs.contains(entry.data_file->file_path); + }) + .IgnoreDeleted(); + + // A data file is listed in the manifests of every snapshot it is live in. + std::unordered_set planned_paths; + ICEBERG_ASSIGN_OR_RAISE( + auto deleted_rows_tasks, + PlanChangelogTasks( + *manifest_group, + [&](ManifestEntry&& entry, std::shared_ptr residual, + std::vector>& planned) -> Status { + const std::string& data_file_path = entry.data_file->file_path; + if (!planned_paths.insert(data_file_path).second) { + return {}; + } + ICEBERG_PRECHECK(entry.sequence_number.has_value() && + entry.file_sequence_number.has_value(), + "Missing sequence number from data file {}", data_file_path); + ICEBERG_RETURN_UNEXPECTED(dv_planner->ValidateNoOtherDeletes( + entry.sequence_number.value(), *entry.data_file)); + + for (int64_t sequence_number : added_dvs.at(data_file_path)) { + // A deletion vector committed together with its data file belongs to the + // file's AddedRowsScanTask. + if (sequence_number == entry.file_sequence_number.value()) { + continue; + } + const int64_t snapshot_id = + snapshot_ids_by_sequence_number.at(sequence_number); + std::vector> existing_deletes; + ICEBERG_ASSIGN_OR_RAISE( + auto existing_dv, + dv_planner->ExistingDeletionVector(sequence_number, data_file_path)); + if (existing_dv != nullptr) { + existing_deletes.push_back(std::move(existing_dv)); + } + planned.push_back(std::make_shared( + snapshot_ordinals.at(snapshot_id), snapshot_id, entry.data_file, + std::vector>{ + dv_planner->AddedDeletionVector(sequence_number, data_file_path)}, + std::move(existing_deletes), residual)); + } + return {}; + })); + tasks.insert(tasks.end(), std::make_move_iterator(deleted_rows_tasks.begin()), + std::make_move_iterator(deleted_rows_tasks.end())); return tasks; } diff --git a/src/iceberg/incremental_changelog_scan.h b/src/iceberg/incremental_changelog_scan.h index e4995c2ea7..2c41d1d5e6 100644 --- a/src/iceberg/incremental_changelog_scan.h +++ b/src/iceberg/incremental_changelog_scan.h @@ -40,10 +40,10 @@ namespace iceberg { /// are deletion vectors. A deletion vector committed for a data file added by the same /// snapshot is attached to the file's AddedRowsScanTask; one committed for a data file /// that already existed produces a DeletedRowsScanTask; and one removed together with -/// its data file becomes an existing delete of the file's DeletedDataFileScanTask. -/// Position delete files and equality delete files are not supported: planning fails -/// with NotSupported when any changelog snapshot carries one that is live or that a -/// changelog snapshot removed. +/// its data file, or left behind when the data file was removed, becomes an existing +/// delete of the file's DeletedDataFileScanTask. Position delete files and equality +/// delete files are not supported: planning fails with NotSupported when one applies to a +/// data file that the changelog reports. class ICEBERG_EXPORT IncrementalChangelogScan : public IncrementalScan { public: diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index 3fd9b51a6d..0eb404c0a6 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -315,14 +316,18 @@ int32_t DeletedRowsScanTask::files_count() const { static_cast(existing_deletes_.size()); } -// The record count of a deletion vector is its cardinality, which bounds the number of -// rows this task can produce. +// The record count of a deletion vector is its cardinality. A deletion vector that +// replaces another one keeps the positions it deleted, so the rows this task produces are +// those of the added deletes minus those of the existing deletes. int64_t DeletedRowsScanTask::estimated_row_count() const { int64_t deleted_rows = 0; for (const auto& delete_file : delete_files_) { deleted_rows += delete_file->record_count; } - return std::min(deleted_rows, data_file_->record_count); + for (const auto& delete_file : existing_deletes_) { + deleted_rows -= delete_file->record_count; + } + return std::clamp(deleted_rows, 0, data_file_->record_count); } // Generic template implementation for Make diff --git a/src/iceberg/test/incremental_changelog_scan_test.cc b/src/iceberg/test/incremental_changelog_scan_test.cc index 1afef55763..62aec83cce 100644 --- a/src/iceberg/test/incremental_changelog_scan_test.cc +++ b/src/iceberg/test/incremental_changelog_scan_test.cc @@ -1007,12 +1007,15 @@ TEST_P(IncrementalChangelogScanTest, PositionDeleteFilesOutsideRangeAreRejected) unpartitioned_spec_)); auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); - auto manifests_c = ManifestsOf(*snapshot_b); + std::vector manifests_c; manifests_c.push_back( WriteDataManifest(version, 3000L, - {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, - MakeDataFile("/path/to/file_b.parquet"))})); - auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "append"); + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet"))})); + manifests_c.push_back(WriteDeleteManifest( + version, 3000L, {MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); @@ -1040,12 +1043,15 @@ TEST_P(IncrementalChangelogScanTest, EqualityDeleteFilesOutsideRangeAreRejected) unpartitioned_spec_)); auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); - auto manifests_c = ManifestsOf(*snapshot_b); + std::vector manifests_c; manifests_c.push_back( WriteDataManifest(version, 3000L, - {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, - MakeDataFile("/path/to/file_b.parquet"))})); - auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "append"); + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet"))})); + manifests_c.push_back(WriteDeleteManifest( + version, 3000L, {MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); @@ -1105,6 +1111,230 @@ TEST_P(IncrementalChangelogScanTest, DeleteFilesRemovedBeforeRangeAreIgnored) { "in changelog scans"))); } +TEST_P(IncrementalChangelogScanTest, DeleteFilesThatCannotApplyAreIgnored) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto position_deletes = + MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet"); + auto equality_deletes = + MakeDeleteFile(DataFile::Content::kEqualityDeletes, FileFormatType::kParquet, + "/path/to/eq_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, + {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, position_deletes), + MakeEntry(ManifestStatus::kAdded, 2000L, 2L, equality_deletes)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, + MakeDataFile("/path/to/file_b.parquet"))})); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "append"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 2000L, 3000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_b.parquet"); + EXPECT_TRUE(task->delete_files().empty()); +} + +TEST_P(IncrementalChangelogScanTest, DeleteFilesOutsideDataFilterAreIgnored) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto partition_a = PartitionValues({Literal::Int(8)}); + auto partition_b = PartitionValues({Literal::Int(1)}); + auto snapshot_a = + MakeAppendSnapshotWithPartitionValues(version, 1000L, std::nullopt, 1L, + {{"/path/to/file_a.parquet", partition_a}, + {"/path/to/file_b.parquet", partition_b}}, + partitioned_spec_); + + auto delete_file = MakeDeleteFile( + DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet", std::nullopt, partition_a, partitioned_spec_); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + partitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + manifests_c.push_back(WriteDataManifest( + version, 3000L, + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet", partition_a, partitioned_spec_)), + MakeEntry( + ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_b.parquet", partition_b, partitioned_spec_))}, + partitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}, partitioned_spec_); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, + PlanChangelog(metadata, 2000L, 3000L, + Expressions::Equal("data", Literal::String("k")))); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_b.parquet"); + + EXPECT_THAT(PlanChangelog(metadata, 2000L, 3000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Position delete files are not supported " + "in changelog scans"))); +} + +TEST_P(IncrementalChangelogScanTest, DeletedFileWithRetainedDeletionVector) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto dv_a = MakeDV("/path/to/dv_a.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + std::erase_if(manifests_c, [](const ManifestFile& manifest) { + return manifest.content == ManifestContent::kData; + }); + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet"))})); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 2000L, 3000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->commit_snapshot_id(), 3000L); + EXPECT_THAT(FilePaths(task->existing_deletes()), + ::testing::ElementsAre("/path/to/dv_a.puffin")); +} + +TEST_P(IncrementalChangelogScanTest, DeletionVectorRewrittenByReplaceSnapshot) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto dv_a1 = MakeDV("/path/to/dv_a1.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto dv_a1_rewritten = + MakeDV("/path/to/dv_a1_rewritten.puffin", "/path/to/file_a.parquet"); + auto manifests_c = ManifestsOf(*snapshot_a); + manifests_c.push_back( + WriteDeleteManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, dv_a1_rewritten), + MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "replace"); + + auto dv_a2 = MakeDV("/path/to/dv_a2.puffin", "/path/to/file_a.parquet"); + auto manifests_d = ManifestsOf(*snapshot_a); + manifests_d.push_back(WriteDeleteManifest( + version, 4000L, + {MakeEntry(ManifestStatus::kAdded, 4000L, 4L, dv_a2), + MakeEntry(ManifestStatus::kDeleted, 4000L, 3L, dv_a1_rewritten)}, + unpartitioned_spec_)); + auto snapshot_d = MakeSnapshot(version, 4000L, 3000L, 4L, manifests_d, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c, snapshot_d}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 1000L, 4000L)); + ASSERT_EQ(tasks.size(), 2); + auto first = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(first, nullptr); + EXPECT_EQ(first->change_ordinal(), 0); + EXPECT_EQ(first->commit_snapshot_id(), 2000L); + EXPECT_THAT(FilePaths(first->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a1.puffin")); + EXPECT_TRUE(first->existing_deletes().empty()); + auto second = std::dynamic_pointer_cast(tasks[1]); + ASSERT_NE(second, nullptr); + EXPECT_EQ(second->change_ordinal(), 1); + EXPECT_EQ(second->commit_snapshot_id(), 4000L); + EXPECT_THAT(FilePaths(second->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a2.puffin")); + EXPECT_THAT(FilePaths(second->existing_deletes()), + ::testing::ElementsAre("/path/to/dv_a1_rewritten.puffin")); +} + +TEST_P(IncrementalChangelogScanTest, DeletedRowsEstimateExcludesExistingDeletes) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = MakeSnapshot( + version, 1000L, 0L, 1L, + {WriteDataManifest(version, 1000L, + {MakeEntry(ManifestStatus::kAdded, 1000L, 1L, + MakeDataFile("/path/to/file_a.parquet", + PartitionValues(std::vector{}), + nullptr, /*record_count=*/100))})}, + "append"); + + auto dv_a1 = MakeDV("/path/to/dv_a1.puffin", "/path/to/file_a.parquet"); + dv_a1->record_count = 30; + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto dv_a2 = MakeDV("/path/to/dv_a2.puffin", "/path/to/file_a.parquet"); + dv_a2->record_count = 35; + auto manifests_c = ManifestsOf(*snapshot_a); + manifests_c.push_back( + WriteDeleteManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, dv_a2), + MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 1000L, 3000L)); + ASSERT_EQ(tasks.size(), 2); + EXPECT_EQ(tasks[0]->estimated_row_count(), 30); + EXPECT_EQ(tasks[1]->estimated_row_count(), 5); +} + INSTANTIATE_TEST_SUITE_P(IncrementalChangelogScanVersions, IncrementalChangelogScanTest, testing::Values(1, 2, 3));