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,