diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index cf465e5fd..e2035033d 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -26,6 +26,7 @@ set(ICEBERG_SOURCES catalog/memory/in_memory_catalog.cc catalog/session_catalog.cc catalog/session_context.cc + changelog_dv_planner_internal.cc data_table_scan.cc delete_file_index.cc deletes/dv_util.cc diff --git a/src/iceberg/changelog_dv_planner_internal.cc b/src/iceberg/changelog_dv_planner_internal.cc new file mode 100644 index 000000000..d89533568 --- /dev/null +++ b/src/iceberg/changelog_dv_planner_internal.cc @@ -0,0 +1,252 @@ +/* + * 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/manifest_evaluator.h" +#include "iceberg/expression/projections.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/partition_spec.h" +#include "iceberg/util/executor_util_internal.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result> +DeletionVectorChangelogPlanner::Make( + std::shared_ptr io, std::shared_ptr schema, + const std::unordered_map>& specs_by_id, + std::shared_ptr filter, bool case_sensitive, OptionalExecutor executor, + const std::vector& delete_manifests, + const std::unordered_map& sequence_numbers) { + ICEBERG_ASSIGN_OR_RAISE( + auto entries, + ParallelCollect( + executor, delete_manifests, + [&](const ManifestFile& manifest) -> Result> { + auto spec_it = specs_by_id.find(manifest.partition_spec_id); + ICEBERG_CHECK(spec_it != specs_by_id.end(), + "Partition spec ID {} not found when reading delete manifest", + manifest.partition_spec_id); + std::shared_ptr partition_filter; + if (filter != nullptr) { + auto projector = + Projections::Inclusive(*spec_it->second, *schema, case_sensitive); + ICEBERG_ASSIGN_OR_RAISE(partition_filter, projector->Project(filter)); + ICEBERG_ASSIGN_OR_RAISE( + auto evaluator, + ManifestEvaluator::MakePartitionFilter( + partition_filter, spec_it->second, *schema, case_sensitive)); + ICEBERG_ASSIGN_OR_RAISE(bool may_match, evaluator->Evaluate(manifest)); + if (!may_match) { + return std::vector{}; + } + } + + ICEBERG_ASSIGN_OR_RAISE( + auto reader, ManifestReader::Make(manifest, io, schema, specs_by_id)); + if (partition_filter != nullptr) { + reader->FilterPartitions(partition_filter).CaseSensitive(case_sensitive); + } + reader->TryDropStats(); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->Entries()); + // A delete file removed before the range no longer applies to any data + // file. One removed by a snapshot in the range applied until then. + std::erase_if(entries, [&](const ManifestEntry& entry) { + return entry.status == ManifestStatus::kDeleted && + !(entry.snapshot_id.has_value() && + sequence_numbers.contains(entry.snapshot_id.value())); + }); + return entries; + })); + + auto planner = std::unique_ptr( + new DeletionVectorChangelogPlanner()); + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Invalid manifest entry with missing delete file"); + const DataFile& file = *entry.data_file; + // The file sequence number belongs to the snapshot that committed the file, unlike + // the data sequence number, which a rewrite can carry over from an older file. + ICEBERG_PRECHECK( + entry.file_sequence_number.has_value() && entry.sequence_number.has_value(), + "Missing sequence number from delete file {}", file.file_path); + ICEBERG_PRECHECK(file.partition_spec_id.has_value(), + "Missing partition spec ID from delete file {}", file.file_path); + + if (file.content == DataFile::Content::kEqualityDeletes) { + auto spec_it = specs_by_id.find(file.partition_spec_id.value()); + ICEBERG_CHECK(spec_it != specs_by_id.end(), + "Partition spec ID {} not found for delete file {}", + file.partition_spec_id.value(), file.file_path); + KeepLatest(spec_it->second->IsUnpartitioned() + ? planner->global_equality_deletes_ + : LatestFor(planner->equality_deletes_by_partition_, file), + entry.sequence_number.value(), file.file_path); + continue; + } + + if (!file.IsDeletionVector()) { + KeepLatest( + file.referenced_data_file.has_value() + ? planner->position_deletes_by_path_[file.referenced_data_file.value()] + : LatestFor(planner->position_deletes_by_partition_, file), + entry.sequence_number.value(), file.file_path); + continue; + } + + ICEBERG_PRECHECK(file.referenced_data_file.has_value(), + "Deletion vector {} does not reference a data file", file.file_path); + // The same vector appears once per manifest that lists it, for example as added in + // the manifest of the snapshot that committed it and as deleted in the manifest of + // the snapshot that removed it. + auto& versions = planner->dvs_by_path_[file.referenced_data_file.value()]; + auto version_it = std::ranges::find_if(versions, [&file](const auto& version) { + return version.file->file_path == file.file_path && + version.file->content_offset == file.content_offset; + }); + if (version_it == versions.end()) { + versions.push_back({.file = entry.data_file, + .added_sequence_number = entry.file_sequence_number.value()}); + version_it = std::prev(versions.end()); + } + if (entry.status == ManifestStatus::kDeleted) { + version_it->removed_sequence_number = + sequence_numbers.at(entry.snapshot_id.value()); + } + } + + for (const auto& [data_file_path, versions] : planner->dvs_by_path_) { + for (auto it = versions.begin(); it != versions.end(); ++it) { + ICEBERG_PRECHECK( + std::ranges::find(std::next(it), versions.end(), it->added_sequence_number, + &DeletionVectorVersion::added_sequence_number) == + versions.end(), + "Snapshot with sequence number {} added multiple deletion vectors for {}", + it->added_sequence_number, data_file_path); + } + } + + return planner; +} + +std::optional& +DeletionVectorChangelogPlanner::LatestFor( + PartitionMap>& deletes_by_partition, + const DataFile& file) { + const int32_t spec_id = file.partition_spec_id.value(); + if (!deletes_by_partition.contains(spec_id, file.partition)) { + deletes_by_partition.put(spec_id, file.partition, std::nullopt); + } + return deletes_by_partition.get(spec_id, file.partition)->get(); +} + +void DeletionVectorChangelogPlanner::KeepLatest(std::optional& latest, + int64_t sequence_number, + const std::string& file_path) { + if (!latest.has_value() || latest->sequence_number < sequence_number) { + latest = LatestDeleteFile{.sequence_number = sequence_number, .file_path = file_path}; + } +} + +std::shared_ptr DeletionVectorChangelogPlanner::AddedDeletionVector( + int64_t sequence_number, const std::string& data_file_path) const { + auto it = dvs_by_path_.find(data_file_path); + if (it == dvs_by_path_.end()) { + return nullptr; + } + auto version_it = std::ranges::find(it->second, sequence_number, + &DeletionVectorVersion::added_sequence_number); + return version_it == it->second.end() ? nullptr : version_it->file; +} + +std::unordered_map> +DeletionVectorChangelogPlanner::AddedDeletionVectors( + const std::unordered_set& sequence_numbers) const { + std::unordered_map> added; + for (const auto& [data_file_path, versions] : dvs_by_path_) { + for (const auto& version : versions) { + if (sequence_numbers.contains(version.added_sequence_number)) { + added[data_file_path].push_back(version.added_sequence_number); + } + } + } + return added; +} + +Result> DeletionVectorChangelogPlanner::ExistingDeletionVector( + int64_t sequence_number, const std::string& data_file_path) const { + auto it = dvs_by_path_.find(data_file_path); + if (it == dvs_by_path_.end()) { + return nullptr; + } + std::shared_ptr existing; + for (const auto& version : it->second) { + const bool live = version.added_sequence_number < sequence_number && + (!version.removed_sequence_number.has_value() || + version.removed_sequence_number.value() >= sequence_number); + if (!live) { + continue; + } + ICEBERG_PRECHECK(existing == nullptr, + "Multiple deletion vectors apply to {} before sequence number {}", + data_file_path, sequence_number); + existing = version.file; + } + return existing; +} + +Status DeletionVectorChangelogPlanner::ValidateNoOtherDeletes( + int64_t data_sequence_number, const DataFile& data_file) const { + ICEBERG_PRECHECK(data_file.partition_spec_id.has_value(), + "Missing partition spec ID from data file {}", data_file.file_path); + const int32_t spec_id = data_file.partition_spec_id.value(); + auto in_partition = [&](const PartitionMap>& deletes) { + auto latest = deletes.get(spec_id, data_file.partition); + return latest.has_value() ? latest->get() : std::nullopt; + }; + auto for_path = position_deletes_by_path_.find(data_file.file_path); + + // An equality delete file applies to data files written before it, and a position + // delete file also to those written in the same commit. + for (const auto& latest : + {global_equality_deletes_, in_partition(equality_deletes_by_partition_)}) { + if (latest.has_value() && latest->sequence_number > data_sequence_number) { + return NotSupported( + "Equality delete files are not supported in changelog scans: {} applies to {}", + latest->file_path, data_file.file_path); + } + } + for (const auto& latest : + {in_partition(position_deletes_by_partition_), + for_path == position_deletes_by_path_.end() ? std::nullopt : for_path->second}) { + if (latest.has_value() && latest->sequence_number >= data_sequence_number) { + return NotSupported( + "Position delete files are not supported in changelog scans: {} applies to {}", + latest->file_path, data_file.file_path); + } + } + return {}; +} + +} // namespace iceberg diff --git a/src/iceberg/changelog_dv_planner_internal.h b/src/iceberg/changelog_dv_planner_internal.h new file mode 100644 index 000000000..cad4f8d03 --- /dev/null +++ b/src/iceberg/changelog_dv_planner_internal.h @@ -0,0 +1,141 @@ +/* + * 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/type_fwd.h" +#include "iceberg/util/executor.h" +#include "iceberg/util/partition_value_util.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 that +/// was live just before it, which DeletedRowsScanTask carries as added and existing +/// deletes. Snapshots are identified by sequence number, so the history of each vector is +/// the sequence number of the snapshot that added it and, if it was removed, of the +/// snapshot that removed it. +/// +/// Position delete files and equality delete files accumulate instead, so attributing +/// their rows to a snapshot would require reading every earlier delete file. A data file +/// that the changelog reports is rejected when one of them applies to it. +class DeletionVectorChangelogPlanner { + public: + /// \brief Index the delete files that can apply to the rows of a changelog. + /// + /// \param io FileIO used to read the delete manifests. + /// \param schema Schema of the scan. + /// \param specs_by_id Partition specs of the table. + /// \param filter Row filter of the scan, used to prune delete manifests and entries by + /// partition. A data file outside the filter is never reported, so neither are the + /// delete files of its partition. + /// \param case_sensitive Whether the filter binds case-sensitively. + /// \param executor Executor used to read the delete manifests. + /// \param delete_manifests Delete manifests of every snapshot in the range, including + /// replace snapshots. A snapshot's own manifests are the only ones guaranteed to hold + /// the entries it removed, because later snapshots drop removed entries when they + /// rewrite a manifest. + /// \param sequence_numbers Sequence numbers of every snapshot in the range, keyed by + /// snapshot ID. A delete file removed by any other snapshot was removed before the + /// range and can no longer apply to a reported row. + static Result> Make( + std::shared_ptr io, std::shared_ptr schema, + const std::unordered_map>& specs_by_id, + std::shared_ptr filter, bool case_sensitive, OptionalExecutor executor, + const std::vector& delete_manifests, + const std::unordered_map& sequence_numbers); + + /// \brief The deletion vector that the snapshot with the given sequence number added + /// for the data file, or nullptr. + std::shared_ptr AddedDeletionVector(int64_t sequence_number, + const std::string& data_file_path) const; + + /// \brief The deletion vector that applied to the data file just before the snapshot + /// with the given sequence number, or nullptr. + /// + /// This is the vector the snapshot replaced or removed, or one it left behind when it + /// removed the data file, so every row it deletes is excluded from the changelog. + Result> ExistingDeletionVector( + int64_t sequence_number, const std::string& data_file_path) const; + + /// \brief Which of the snapshots with the given sequence numbers added a deletion + /// vector, keyed by the path of the data file the vector references. + std::unordered_map> AddedDeletionVectors( + const std::unordered_set& sequence_numbers) const; + + /// \brief Fail with NotSupported if a position delete file or an equality delete file + /// applies to the data file. + /// + /// \param data_sequence_number Data sequence number of the data file. + /// \param data_file The data file that the changelog reports. + Status ValidateNoOtherDeletes(int64_t data_sequence_number, + const DataFile& data_file) const; + + private: + // One deletion vector and the sequence numbers of the snapshots that added and + // removed it. + struct DeletionVectorVersion { + std::shared_ptr file; + int64_t added_sequence_number; + std::optional removed_sequence_number; + }; + + // The delete file with the highest sequence number among those sharing a scope. A + // delete file applies to a data file based on sequence numbers alone once the scope + // matches, so the latest one decides whether any of them applies. + struct LatestDeleteFile { + int64_t sequence_number; + std::string file_path; + }; + + DeletionVectorChangelogPlanner() = default; + + static std::optional& LatestFor( + PartitionMap>& deletes_by_partition, + const DataFile& file); + static void KeepLatest(std::optional& latest, int64_t sequence_number, + const std::string& file_path); + + // Deletion vector history keyed by the path of the data file it references. + std::unordered_map> dvs_by_path_; + + std::optional global_equality_deletes_; + PartitionMap> equality_deletes_by_partition_; + PartitionMap> position_deletes_by_partition_; + std::unordered_map> + position_deletes_by_path_; +}; + +} // namespace iceberg diff --git a/src/iceberg/incremental_changelog_scan.cc b/src/iceberg/incremental_changelog_scan.cc index 5a1dbba11..1bbcc2395 100644 --- a/src/iceberg/incremental_changelog_scan.cc +++ b/src/iceberg/incremental_changelog_scan.cc @@ -19,12 +19,15 @@ #include "iceberg/incremental_changelog_scan.h" +#include +#include #include #include #include #include #include +#include "iceberg/changelog_dv_planner_internal.h" #include "iceberg/expression/residual_evaluator.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/manifest/manifest_group.h" @@ -36,6 +39,49 @@ namespace iceberg { +namespace { + +// Appends the changelog tasks for one data file of a changelog scan. +using ChangelogTaskFactory = + std::function residual, + std::vector>& tasks)>; + +// Plans the data files of the manifest group, keeping the stats and computing the +// residual that the scan requests before handing each file to make_tasks. +Result>> PlanChangelogTasks( + ManifestGroup& manifest_group, const ChangelogTaskFactory& make_tasks) { + auto create_tasks_func = + [&](std::vector&& entries, + const TaskContext& ctx) -> Result>> { + std::vector> tasks; + tasks.reserve(entries.size()); + + for (auto& entry : entries) { + ICEBERG_PRECHECK(entry.data_file != nullptr, + "Invalid manifest entry with missing data file"); + + if (ctx.drop_stats) { + ContentFileUtil::DropAllStats(*entry.data_file); + } else if (!ctx.columns_to_keep_stats.empty()) { + ContentFileUtil::DropUnselectedStats(*entry.data_file, ctx.columns_to_keep_stats); + } + + ICEBERG_ASSIGN_OR_RAISE(auto residual, + ctx.residuals->ResidualFor(entry.data_file->partition)); + ICEBERG_RETURN_UNEXPECTED(make_tasks(std::move(entry), std::move(residual), tasks)); + } + return tasks; + }; + + ICEBERG_ASSIGN_OR_RAISE(auto tasks, manifest_group.Plan(create_tasks_func)); + return tasks | std::views::transform([](const auto& task) { + return std::static_pointer_cast(task); + }) | + std::ranges::to(); +} + +} // namespace + Result> IncrementalChangelogScan::Make( std::shared_ptr metadata, std::shared_ptr schema, std::shared_ptr io, internal::TableScanContext context) { @@ -59,6 +105,10 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl SnapshotUtil::AncestorsBetween(*metadata_, to_snapshot_id_inclusive, from_snapshot_id_exclusive)); + // Row-level deletes are only understood as deletion vectors, which format version 3 + // requires; earlier versions store them as position and equality delete files. + const bool deletes_supported = metadata_->format_version >= 3; + std::vector, std::unique_ptr>> changelog_snapshots; @@ -66,11 +116,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)); } @@ -81,6 +135,8 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl std::unordered_set snapshot_ids; std::unordered_map snapshot_ordinals; + std::unordered_set changelog_sequence_numbers; + std::unordered_map snapshot_ids_by_sequence_number; for (const auto& snapshot : changelog_snapshots) { ICEBERG_PRECHECK( std::cmp_less_equal(snapshot_ids.size(), std::numeric_limits::max()), @@ -88,96 +144,210 @@ IncrementalChangelogScan::PlanFiles(std::optional from_snapshot_id_excl snapshot_ids.insert(snapshot.first->snapshot_id); snapshot_ordinals.try_emplace(snapshot.first->snapshot_id, static_cast(snapshot_ordinals.size())); + changelog_sequence_numbers.insert(snapshot.first->sequence_number); + snapshot_ids_by_sequence_number.try_emplace(snapshot.first->sequence_number, + snapshot.first->snapshot_id); } + // Data manifests written by the changelog snapshots hold their added and deleted data + // files; all data manifests of the changelog snapshots hold the data files that a + // deletion vector can reference. + std::vector added_data_manifests; std::vector data_manifests; std::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 (!seen_manifest_paths.insert(manifest.manifest_path).second) { + continue; + } if (manifest.added_snapshot_id.has_value() && - snapshot_ids.contains(manifest.added_snapshot_id.value()) && - seen_manifest_paths.insert(manifest.manifest_path).second) { - data_manifests.push_back(manifest); + snapshot_ids.contains(manifest.added_snapshot_id.value())) { + added_data_manifests.push_back(manifest); + } + data_manifests.push_back(manifest); + } + } + + // Delete manifests of every snapshot in the range, replace snapshots included, so that + // each removal of a deletion vector is seen in the manifest of the snapshot that made + // it. + std::vector delete_manifests; + std::unordered_map sequence_numbers; + if (deletes_supported) { + for (const auto& snapshot : ancestors_snapshots) { + sequence_numbers.emplace(snapshot->snapshot_id, snapshot->sequence_number); + SnapshotReader snapshot_reader(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto snapshot_delete_manifests, + snapshot_reader.DeleteManifests(io_)); + 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 (added_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, filter(), context_.case_sensitive, + context_.plan_executor, delete_manifests, sequence_numbers)); } - auto create_tasks_func = - [&snapshot_ordinals]( - std::vector&& entries, - const TaskContext& ctx) -> Result>> { - std::vector> tasks; - tasks.reserve(entries.size()); + auto make_manifest_group = + [&](std::vector manifests) -> Result> { + ICEBERG_ASSIGN_OR_RAISE( + auto manifest_group, + ManifestGroup::Make(io_, schema_, specs_by_id, std::move(manifests), + /*delete_manifests=*/{})); + manifest_group->CaseSensitive(context_.case_sensitive) + .Select(ScanColumns()) + .FilterData(filter()) + .ColumnsToKeepStats(context_.columns_to_keep_stats) + .PlanWith(context_.plan_executor); + if (context_.ignore_residuals) { + manifest_group->IgnoreResiduals(); + } + return manifest_group; + }; - for (auto& entry : entries) { - ICEBERG_PRECHECK(entry.snapshot_id.has_value() && entry.data_file, - "Invalid manifest entry with missing snapshot id or data file"); + std::vector> tasks; - int64_t commit_snapshot_id = entry.snapshot_id.value(); - auto ordinal_it = snapshot_ordinals.find(commit_snapshot_id); - ICEBERG_PRECHECK(ordinal_it != snapshot_ordinals.end(), - "Invalid manifest entry with missing snapshot ordinal"); + if (!added_data_manifests.empty()) { + ICEBERG_ASSIGN_OR_RAISE(auto manifest_group, + make_manifest_group(std::move(added_data_manifests))); + manifest_group + ->FilterManifestEntries([&snapshot_ids](const ManifestEntry& entry) { + return entry.snapshot_id.has_value() && + snapshot_ids.contains(entry.snapshot_id.value()); + }) + .IgnoreExisting(); - int32_t change_ordinal = ordinal_it->second; + ICEBERG_ASSIGN_OR_RAISE( + tasks, + PlanChangelogTasks( + *manifest_group, + [&](ManifestEntry&& entry, std::shared_ptr residual, + std::vector>& planned) -> Status { + ICEBERG_PRECHECK(entry.snapshot_id.has_value(), + "Invalid manifest entry with missing snapshot id"); - 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); - } + int64_t commit_snapshot_id = entry.snapshot_id.value(); + auto ordinal_it = snapshot_ordinals.find(commit_snapshot_id); + ICEBERG_PRECHECK(ordinal_it != snapshot_ordinals.end(), + "Invalid manifest entry with missing snapshot ordinal"); - ICEBERG_ASSIGN_OR_RAISE(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"); - } - } + int32_t change_ordinal = ordinal_it->second; + + std::vector> deletes; + if (dv_planner) { + ICEBERG_PRECHECK(entry.sequence_number.has_value(), + "Missing sequence number from data file {}", + entry.data_file->file_path); + ICEBERG_RETURN_UNEXPECTED(dv_planner->ValidateNoOtherDeletes( + entry.sequence_number.value(), *entry.data_file)); + const int64_t sequence_number = sequence_numbers.at(commit_snapshot_id); + const std::string& data_file_path = entry.data_file->file_path; + std::shared_ptr dv; + if (entry.status == ManifestStatus::kAdded) { + dv = dv_planner->AddedDeletionVector(sequence_number, data_file_path); + } else { + ICEBERG_ASSIGN_OR_RAISE(dv, dv_planner->ExistingDeletionVector( + sequence_number, data_file_path)); + } + if (dv != nullptr) { + deletes.push_back(std::move(dv)); + } + } + + switch (entry.status) { + case ManifestStatus::kAdded: + planned.push_back(std::make_shared( + change_ordinal, commit_snapshot_id, std::move(entry.data_file), + std::move(deletes), std::move(residual))); + return {}; + case ManifestStatus::kDeleted: + planned.push_back(std::make_shared( + change_ordinal, commit_snapshot_id, std::move(entry.data_file), + std::move(deletes), std::move(residual))); + return {}; + case ManifestStatus::kExisting: + return InvalidArgument("Unexpected entry status: EXISTING"); + } + return {}; + })); + } + + if (dv_planner == nullptr) { return tasks; - }; + } + auto added_dvs = dv_planner->AddedDeletionVectors(changelog_sequence_numbers); + if (added_dvs.empty()) { + 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(); + ICEBERG_ASSIGN_OR_RAISE(auto manifest_group, + make_manifest_group(std::move(data_manifests))); + manifest_group + ->FilterManifestEntries([&added_dvs](const ManifestEntry& entry) { + return entry.data_file != nullptr && + added_dvs.contains(entry.data_file->file_path); + }) + .IgnoreDeleted(); + + // A data file is listed in the manifests of every snapshot it is live in. + std::unordered_set planned_paths; + ICEBERG_ASSIGN_OR_RAISE( + auto deleted_rows_tasks, + PlanChangelogTasks( + *manifest_group, + [&](ManifestEntry&& entry, std::shared_ptr residual, + std::vector>& planned) -> Status { + const std::string& data_file_path = entry.data_file->file_path; + if (!planned_paths.insert(data_file_path).second) { + return {}; + } + ICEBERG_PRECHECK(entry.sequence_number.has_value() && + entry.file_sequence_number.has_value(), + "Missing sequence number from data file {}", data_file_path); + ICEBERG_RETURN_UNEXPECTED(dv_planner->ValidateNoOtherDeletes( + entry.sequence_number.value(), *entry.data_file)); + + for (int64_t sequence_number : added_dvs.at(data_file_path)) { + // A deletion vector committed together with its data file belongs to the + // file's AddedRowsScanTask. + if (sequence_number == entry.file_sequence_number.value()) { + continue; + } + const int64_t snapshot_id = + snapshot_ids_by_sequence_number.at(sequence_number); + std::vector> existing_deletes; + ICEBERG_ASSIGN_OR_RAISE( + auto existing_dv, + dv_planner->ExistingDeletionVector(sequence_number, data_file_path)); + if (existing_dv != nullptr) { + existing_deletes.push_back(std::move(existing_dv)); + } + planned.push_back(std::make_shared( + snapshot_ordinals.at(snapshot_id), snapshot_id, entry.data_file, + std::vector>{ + dv_planner->AddedDeletionVector(sequence_number, data_file_path)}, + std::move(existing_deletes), residual)); + } + return {}; + })); + tasks.insert(tasks.end(), std::make_move_iterator(deleted_rows_tasks.begin()), + std::make_move_iterator(deleted_rows_tasks.end())); + return tasks; } } // namespace iceberg diff --git a/src/iceberg/incremental_changelog_scan.h b/src/iceberg/incremental_changelog_scan.h index 6625b68f2..2c41d1d5e 100644 --- a/src/iceberg/incremental_changelog_scan.h +++ b/src/iceberg/incremental_changelog_scan.h @@ -35,6 +35,15 @@ namespace iceberg { /// \brief A scan that reads changelog entries between snapshots. +/// +/// Delete files are supported only for format version 3 tables, where row-level deletes +/// are deletion vectors. A deletion vector committed for a data file added by the same +/// snapshot is attached to the file's AddedRowsScanTask; one committed for a data file +/// that already existed produces a DeletedRowsScanTask; and one removed together with +/// its data file, or left behind when the data file was removed, becomes an existing +/// delete of the file's DeletedDataFileScanTask. Position delete files and equality +/// delete files are not supported: planning fails with NotSupported when one applies to a +/// data file that the changelog reports. class ICEBERG_EXPORT IncrementalChangelogScan : public IncrementalScan { public: diff --git a/src/iceberg/table_scan.cc b/src/iceberg/table_scan.cc index d04e3385b..0eb404c0a 100644 --- a/src/iceberg/table_scan.cc +++ b/src/iceberg/table_scan.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -290,6 +291,45 @@ 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. A deletion vector that +// replaces another one keeps the positions it deleted, so the rows this task produces are +// those of the added deletes minus those of the existing deletes. +int64_t DeletedRowsScanTask::estimated_row_count() const { + int64_t deleted_rows = 0; + for (const auto& delete_file : delete_files_) { + deleted_rows += delete_file->record_count; + } + for (const auto& delete_file : existing_deletes_) { + deleted_rows -= delete_file->record_count; + } + return std::clamp(deleted_rows, 0, data_file_->record_count); +} + // Generic template implementation for Make template Result>> TableScanBuilder::Make( diff --git a/src/iceberg/table_scan.h b/src/iceberg/table_scan.h index 1307089e3..97f164b9b 100644 --- a/src/iceberg/table_scan.h +++ b/src/iceberg/table_scan.h @@ -254,6 +254,64 @@ class ICEBERG_EXPORT DeletedDataFileScanTask : public ChangelogScanTask { } }; +/// \brief A scan task for deletes generated by adding deletion vectors to the table. +/// +/// Suppose snapshot S1 contains data files F1 and F2. Then snapshot S2 adds a deletion +/// vector, DV1, that deletes records from F2, and snapshot S3 replaces DV1 with DV2 that +/// deletes further records from F2. A scan for changes from S2 to S3 (inclusive) should +/// include the following tasks: +/// - DeletedRowsScanTask(file=F2, added-deletes=[DV1], existing-deletes=[], snapshot=S2) +/// - DeletedRowsScanTask(file=F2, added-deletes=[DV2], existing-deletes=[DV1], +/// snapshot=S3) +/// +/// Readers consuming these tasks should produce the records that are deleted by the added +/// deletes but not by the existing deletes, with metadata like change ordinal and commit +/// snapshot ID. +class ICEBERG_EXPORT DeletedRowsScanTask : public ChangelogScanTask { + public: + /// \brief Construct a DeletedRowsScanTask. + /// + /// \param change_ordinal Position in the changelog order (0-based). + /// \param commit_snapshot_id The snapshot ID that committed the added deletes. + /// \param data_file The data file containing the deleted rows. + /// \param added_deletes Delete files committed by the snapshot for this data file. + /// \param existing_deletes Delete files that applied to this data file before the + /// snapshot. + /// \param residual_filter Optional residual filter to apply after reading. + DeletedRowsScanTask(int32_t change_ordinal, int64_t commit_snapshot_id, + std::shared_ptr data_file, + std::vector> added_deletes, + std::vector> existing_deletes, + std::shared_ptr residual_filter = nullptr); + + ChangelogOperation operation() const override { return ChangelogOperation::kDelete; } + + int64_t size_bytes() const override; + int32_t files_count() const override; + int64_t estimated_row_count() const override; + + /// \brief The data file containing the deleted rows. + const std::shared_ptr& data_file() const { return data_file_; } + + /// \brief Delete files added by the commit snapshot that apply to the data file. + /// + /// Records removed by these delete files appear as deletes in the changelog. + const std::vector>& added_deletes() const { + return delete_files_; + } + + /// \brief Delete files that existed before the commit snapshot and must be applied + /// prior to determining which records are deleted by added_deletes(). + /// + /// Records removed by these delete files do not appear in the changelog. + const std::vector>& existing_deletes() const { + return existing_deletes_; + } + + private: + std::vector> existing_deletes_; +}; + namespace internal { // Internal table scan context used by different scan implementations. diff --git a/src/iceberg/test/incremental_changelog_scan_test.cc b/src/iceberg/test/incremental_changelog_scan_test.cc index e8b374044..62aec83cc 100644 --- a/src/iceberg/test/incremental_changelog_scan_test.cc +++ b/src/iceberg/test/incremental_changelog_scan_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include @@ -28,6 +29,7 @@ #include #include "iceberg/expression/expressions.h" +#include "iceberg/manifest/manifest_entry.h" #include "iceberg/snapshot.h" #include "iceberg/table_scan.h" #include "iceberg/test/scan_test_base.h" @@ -43,11 +45,19 @@ const std::string& TaskFilePath(const std::shared_ptr& task) if (auto deleted = std::dynamic_pointer_cast(task)) { return deleted->data_file()->file_path; } + if (auto deleted_rows = std::dynamic_pointer_cast(task)) { + return deleted_rows->data_file()->file_path; + } static const std::string empty_path; return empty_path; } +std::vector FilePaths(const std::vector>& files) { + return files | std::views::transform([](const auto& file) { return file->file_path; }) | + std::ranges::to>(); +} + /// \brief Sort changelog scan tasks for deterministic ordering. /// Sorts by change_ordinal, then by operation type name, then by file path. template @@ -67,7 +77,100 @@ void SortTasks(std::vector>& tasks) { } // namespace -class IncrementalChangelogScanTest : public ScanTestBase {}; +class IncrementalChangelogScanTest : public ScanTestBase { + protected: + std::shared_ptr MakeDeleteFile( + DataFile::Content content, FileFormatType format, const std::string& path, + std::optional referenced_data_file = std::nullopt, + PartitionValues partition = PartitionValues(std::vector{}), + std::shared_ptr spec = nullptr) { + auto effective_spec = spec ? spec : unpartitioned_spec_; + DataFile file{ + .content = content, + .file_path = path, + .file_format = format, + .partition = std::move(partition), + .record_count = 1, + .file_size_in_bytes = 10, + .partition_spec_id = effective_spec->spec_id(), + }; + if (content == DataFile::Content::kEqualityDeletes) { + file.equality_ids = {1}; + } + if (format == FileFormatType::kPuffin) { + file.referenced_data_file = std::move(referenced_data_file); + file.content_offset = 4L; + file.content_size_in_bytes = 6L; + } + return std::make_shared(std::move(file)); + } + + std::shared_ptr MakeDV( + const std::string& path, const std::string& referenced_data_file, + PartitionValues partition = PartitionValues(std::vector{}), + std::shared_ptr spec = nullptr) { + return MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kPuffin, + path, referenced_data_file, std::move(partition), + std::move(spec)); + } + + std::vector ManifestsOf(const Snapshot& snapshot) { + SnapshotReader reader(&snapshot); + auto manifests = reader.Manifests(file_io_); + EXPECT_THAT(manifests, IsOk()); + if (!manifests.has_value()) { + return {}; + } + return {manifests->begin(), manifests->end()}; + } + + std::shared_ptr MakeSnapshot(int8_t format_version, int64_t snapshot_id, + int64_t parent_snapshot_id, + int64_t sequence_number, + const std::vector& manifests, + const std::string& operation) { + auto manifest_list = WriteManifestList( + format_version, snapshot_id, parent_snapshot_id, sequence_number, manifests); + return std::make_shared(Snapshot{ + .snapshot_id = snapshot_id, + .parent_snapshot_id = parent_snapshot_id, + .sequence_number = sequence_number, + .timestamp_ms = TimePointMsFromUnixMs(1609459200000L + sequence_number * 1000), + .manifest_list = manifest_list, + .summary = {{"operation", operation}}, + .schema_id = schema_->schema_id(), + }); + } + + std::shared_ptr MakeMetadata( + const std::vector>& snapshots, + std::shared_ptr default_spec = nullptr) { + int64_t current_snapshot_id = snapshots.back()->snapshot_id; + return MakeTableMetadata(snapshots, current_snapshot_id, + {{"main", std::make_shared(SnapshotRef{ + .snapshot_id = current_snapshot_id, + .retention = SnapshotRef::Branch{}})}}, + std::move(default_spec)); + } + + Result>> PlanChangelog( + std::shared_ptr metadata, std::optional from_snapshot_id, + int64_t to_snapshot_id, std::shared_ptr filter = nullptr) { + ICEBERG_ASSIGN_OR_RAISE( + auto builder, MakeScanBuilder(std::move(metadata))); + if (from_snapshot_id.has_value()) { + builder->FromSnapshot(from_snapshot_id.value()); + } + builder->ToSnapshot(to_snapshot_id); + if (filter != nullptr) { + builder->Filter(std::move(filter)); + } + ICEBERG_ASSIGN_OR_RAISE(auto scan, builder->Build()); + ICEBERG_ASSIGN_OR_RAISE(auto tasks, scan->PlanFiles()); + SortTasks(tasks); + return tasks; + } +}; TEST_P(IncrementalChangelogScanTest, DataFilters) { auto version = GetParam(); @@ -554,68 +657,682 @@ 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}); + + 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); +} + +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"); + + std::vector manifests_c; + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet"))})); + manifests_c.push_back(WriteDeleteManifest( + version, 3000L, {MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + 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"); + + std::vector manifests_c; + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet"))})); + manifests_c.push_back(WriteDeleteManifest( + version, 3000L, {MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, delete_file)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + 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"))); +} + +TEST_P(IncrementalChangelogScanTest, DeleteFilesThatCannotApplyAreIgnored) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto position_deletes = + MakeDeleteFile(DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet"); + auto equality_deletes = + MakeDeleteFile(DataFile::Content::kEqualityDeletes, FileFormatType::kParquet, + "/path/to/eq_deletes.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, + {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, position_deletes), + MakeEntry(ManifestStatus::kAdded, 2000L, 2L, equality_deletes)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, + MakeDataFile("/path/to/file_b.parquet"))})); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "append"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 2000L, 3000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_b.parquet"); + EXPECT_TRUE(task->delete_files().empty()); +} + +TEST_P(IncrementalChangelogScanTest, DeleteFilesOutsideDataFilterAreIgnored) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Delete files are rejected before format version 3"; + } + + auto partition_a = PartitionValues({Literal::Int(8)}); + auto partition_b = PartitionValues({Literal::Int(1)}); + auto snapshot_a = + MakeAppendSnapshotWithPartitionValues(version, 1000L, std::nullopt, 1L, + {{"/path/to/file_a.parquet", partition_a}, + {"/path/to/file_b.parquet", partition_b}}, + partitioned_spec_); + + auto delete_file = MakeDeleteFile( + DataFile::Content::kPositionDeletes, FileFormatType::kParquet, + "/path/to/file_a_deletes.parquet", std::nullopt, partition_a, partitioned_spec_); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, delete_file)}, + partitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + manifests_c.push_back(WriteDataManifest( + version, 3000L, + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet", partition_a, partitioned_spec_)), + MakeEntry( + ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_b.parquet", partition_b, partitioned_spec_))}, + partitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}, partitioned_spec_); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, + PlanChangelog(metadata, 2000L, 3000L, + Expressions::Equal("data", Literal::String("k")))); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->data_file()->file_path, "/path/to/file_b.parquet"); + + EXPECT_THAT(PlanChangelog(metadata, 2000L, 3000L), + ::testing::AllOf(IsError(ErrorKind::kNotSupported), + HasErrorMessage("Position delete files are not supported " + "in changelog scans"))); +} + +TEST_P(IncrementalChangelogScanTest, DeletedFileWithRetainedDeletionVector) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto dv_a = MakeDV("/path/to/dv_a.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto manifests_c = ManifestsOf(*snapshot_b); + std::erase_if(manifests_c, [](const ManifestFile& manifest) { + return manifest.content == ManifestContent::kData; }); + manifests_c.push_back( + WriteDataManifest(version, 3000L, + {MakeEntry(ManifestStatus::kDeleted, 3000L, 1L, + MakeDataFile("/path/to/file_a.parquet"))})); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); - auto metadata = MakeTableMetadata( - {snapshot_a, snapshot_b}, 2000L, - {{"main", std::make_shared(SnapshotRef{ - .snapshot_id = 2000L, .retention = SnapshotRef::Branch{}})}}); + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); - 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"))); + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 2000L, 3000L)); + ASSERT_EQ(tasks.size(), 1); + auto task = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(task, nullptr); + EXPECT_EQ(task->commit_snapshot_id(), 3000L); + EXPECT_THAT(FilePaths(task->existing_deletes()), + ::testing::ElementsAre("/path/to/dv_a.puffin")); +} + +TEST_P(IncrementalChangelogScanTest, DeletionVectorRewrittenByReplaceSnapshot) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = + MakeAppendSnapshot(version, 1000L, std::nullopt, 1L, {"/path/to/file_a.parquet"}); + + auto dv_a1 = MakeDV("/path/to/dv_a1.puffin", "/path/to/file_a.parquet"); + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto dv_a1_rewritten = + MakeDV("/path/to/dv_a1_rewritten.puffin", "/path/to/file_a.parquet"); + auto manifests_c = ManifestsOf(*snapshot_a); + manifests_c.push_back( + WriteDeleteManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, dv_a1_rewritten), + MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "replace"); + + auto dv_a2 = MakeDV("/path/to/dv_a2.puffin", "/path/to/file_a.parquet"); + auto manifests_d = ManifestsOf(*snapshot_a); + manifests_d.push_back(WriteDeleteManifest( + version, 4000L, + {MakeEntry(ManifestStatus::kAdded, 4000L, 4L, dv_a2), + MakeEntry(ManifestStatus::kDeleted, 4000L, 3L, dv_a1_rewritten)}, + unpartitioned_spec_)); + auto snapshot_d = MakeSnapshot(version, 4000L, 3000L, 4L, manifests_d, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c, snapshot_d}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 1000L, 4000L)); + ASSERT_EQ(tasks.size(), 2); + auto first = std::dynamic_pointer_cast(tasks[0]); + ASSERT_NE(first, nullptr); + EXPECT_EQ(first->change_ordinal(), 0); + EXPECT_EQ(first->commit_snapshot_id(), 2000L); + EXPECT_THAT(FilePaths(first->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a1.puffin")); + EXPECT_TRUE(first->existing_deletes().empty()); + auto second = std::dynamic_pointer_cast(tasks[1]); + ASSERT_NE(second, nullptr); + EXPECT_EQ(second->change_ordinal(), 1); + EXPECT_EQ(second->commit_snapshot_id(), 4000L); + EXPECT_THAT(FilePaths(second->added_deletes()), + ::testing::ElementsAre("/path/to/dv_a2.puffin")); + EXPECT_THAT(FilePaths(second->existing_deletes()), + ::testing::ElementsAre("/path/to/dv_a1_rewritten.puffin")); +} + +TEST_P(IncrementalChangelogScanTest, DeletedRowsEstimateExcludesExistingDeletes) { + auto version = GetParam(); + if (version < 3) { + GTEST_SKIP() << "Deletion vectors require format version 3"; + } + + auto snapshot_a = MakeSnapshot( + version, 1000L, 0L, 1L, + {WriteDataManifest(version, 1000L, + {MakeEntry(ManifestStatus::kAdded, 1000L, 1L, + MakeDataFile("/path/to/file_a.parquet", + PartitionValues(std::vector{}), + nullptr, /*record_count=*/100))})}, + "append"); + + auto dv_a1 = MakeDV("/path/to/dv_a1.puffin", "/path/to/file_a.parquet"); + dv_a1->record_count = 30; + auto manifests_b = ManifestsOf(*snapshot_a); + manifests_b.push_back(WriteDeleteManifest( + version, 2000L, {MakeEntry(ManifestStatus::kAdded, 2000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_b = MakeSnapshot(version, 2000L, 1000L, 2L, manifests_b, "delete"); + + auto dv_a2 = MakeDV("/path/to/dv_a2.puffin", "/path/to/file_a.parquet"); + dv_a2->record_count = 35; + auto manifests_c = ManifestsOf(*snapshot_a); + manifests_c.push_back( + WriteDeleteManifest(version, 3000L, + {MakeEntry(ManifestStatus::kAdded, 3000L, 3L, dv_a2), + MakeEntry(ManifestStatus::kDeleted, 3000L, 2L, dv_a1)}, + unpartitioned_spec_)); + auto snapshot_c = MakeSnapshot(version, 3000L, 2000L, 3L, manifests_c, "delete"); + + auto metadata = MakeMetadata({snapshot_a, snapshot_b, snapshot_c}); + + ICEBERG_UNWRAP_OR_FAIL(auto tasks, PlanChangelog(metadata, 1000L, 3000L)); + ASSERT_EQ(tasks.size(), 2); + EXPECT_EQ(tasks[0]->estimated_row_count(), 30); + EXPECT_EQ(tasks[1]->estimated_row_count(), 5); } INSTANTIATE_TEST_SUITE_P(IncrementalChangelogScanVersions, IncrementalChangelogScanTest,