From ab1e688a215bf5bfb8c74ffefad5e683213abb34 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Wed, 23 Sep 2026 11:16:53 +0800 Subject: [PATCH] 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/table_scan.cc | 248 +++++--- src/iceberg/table_scan.h | 67 ++ .../test/incremental_changelog_scan_test.cc | 593 ++++++++++++++++-- 6 files changed, 1135 insertions(+), 123 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 8a98274ff..99c1ab1f3 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 delete_file_index.cc deletes/dv_util.cc deletes/dv_writer.cc diff --git a/src/iceberg/changelog_dv_planner_internal.cc b/src/iceberg/changelog_dv_planner_internal.cc new file mode 100644 index 000000000..6496595e2 --- /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 000000000..1f76cffb3 --- /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/table_scan.cc b/src/iceberg/table_scan.cc index e4a85e9a6..e9e11dee9 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -25,6 +25,7 @@ #include #include +#include "iceberg/changelog_dv_planner_internal.h" #include "iceberg/expression/binder.h" #include "iceberg/expression/expression.h" #include "iceberg/expression/residual_evaluator.h" @@ -309,6 +310,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( @@ -868,6 +904,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; @@ -875,11 +915,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)); } @@ -900,93 +944,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()); - - 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"); + 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; - int32_t change_ordinal = ordinal_it->second; + if (!data_manifests.empty()) { + 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(); + } - 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()); + + 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(); + } - 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"); - } + 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/table_scan.h b/src/iceberg/table_scan.h index 7310d435b..99f3a8167 100644 --- a/src/iceberg/table_scan.h +++ b/src/iceberg/table_scan.h @@ -216,6 +216,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. @@ -533,6 +591,15 @@ class ICEBERG_EXPORT IncrementalAppendScan : public IncrementalScan { public: diff --git a/src/iceberg/test/incremental_changelog_scan_test.cc b/src/iceberg/test/incremental_changelog_scan_test.cc index dbf715375..79d485f65 100644 --- a/src/iceberg/test/incremental_changelog_scan_test.cc +++ b/src/iceberg/test/incremental_changelog_scan_test.cc @@ -19,6 +19,7 @@ #include #include +#include #include #include @@ -26,6 +27,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" @@ -41,11 +43,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 @@ -65,7 +75,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(); @@ -552,68 +655,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,