Copilot commented on code in PR #958:
URL: https://github.com/apache/iceberg-cpp/pull/958#discussion_r4425727408
##########
src/iceberg/incremental_changelog_scan.cc:
##########
@@ -36,6 +39,49 @@
namespace iceberg {
+namespace {
+
+// Appends the changelog tasks for one data file of a changelog scan.
+using ChangelogTaskFactory =
+ std::function<Status(ManifestEntry&& entry, std::shared_ptr<Expression>
residual,
+ std::vector<std::shared_ptr<ScanTask>>& 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<std::vector<std::shared_ptr<ChangelogScanTask>>> PlanChangelogTasks(
+ ManifestGroup& manifest_group, const ChangelogTaskFactory& make_tasks) {
+ auto create_tasks_func =
+ [&](std::vector<ManifestEntry>&& entries,
+ const TaskContext& ctx) ->
Result<std::vector<std::shared_ptr<ScanTask>>> {
+ std::vector<std::shared_ptr<ScanTask>> tasks;
+ tasks.reserve(entries.size());
+
+ for (auto& entry : entries) {
+ 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<ChangelogScanTask>(task);
+ }) |
+ std::ranges::to<std::vector>();
Review Comment:
`std::ranges::to` is a C++23 facility; if this project is built as C++20
(common for ranges usage), this will be a compile error. Consider replacing the
pipe-to-container conversion with an explicit
`std::vector<std::shared_ptr<ChangelogScanTask>> out;
out.reserve(tasks.size());` + `std::transform`/loop, or a small helper that
doesn’t rely on C++23.
##########
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 <algorithm>
+#include <utility>
+
+#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<std::unique_ptr<DeletionVectorChangelogPlanner>>
+DeletionVectorChangelogPlanner::Make(
+ std::shared_ptr<FileIO> io, std::shared_ptr<Schema> schema,
+ const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
specs_by_id,
+ std::shared_ptr<Expression> filter, bool case_sensitive, OptionalExecutor
executor,
+ const std::vector<ManifestFile>& delete_manifests,
+ const std::unordered_map<int64_t, int64_t>& sequence_numbers) {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto entries,
+ ParallelCollect(
+ executor, delete_manifests,
+ [&](const ManifestFile& manifest) ->
Result<std::vector<ManifestEntry>> {
+ 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<Expression> 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<ManifestEntry>{};
+ }
+ }
+
+ 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<DeletionVectorChangelogPlanner>(
+ 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::LatestDeleteFile>&
+DeletionVectorChangelogPlanner::LatestFor(
+ PartitionMap<std::optional<LatestDeleteFile>>& 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<LatestDeleteFile>&
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<DataFile> 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<std::string, std::vector<int64_t>>
+DeletionVectorChangelogPlanner::AddedDeletionVectors(
+ const std::unordered_set<int64_t>& sequence_numbers) const {
+ std::unordered_map<std::string, std::vector<int64_t>> 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<std::shared_ptr<DataFile>>
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<DataFile> 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<std::optional<LatestDeleteFile>>&
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);
+ }
Review Comment:
Same issue as with equality deletes: tests appear to expect a stable message
without the appended \": {} applies to {}\" detail. To avoid brittle
exact-message coupling, either standardize the thrown message to the expected
stable string or update test matchers to accept a prefix/substring.
##########
src/iceberg/incremental_changelog_scan.cc:
##########
@@ -59,18 +105,26 @@ IncrementalChangelogScan::PlanFiles(std::optional<int64_t>
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::pair<std::shared_ptr<Snapshot>,
std::unique_ptr<SnapshotReader>>>
changelog_snapshots;
for (const auto& snapshot : std::ranges::reverse_view(ancestors_snapshots)) {
auto operation = snapshot->Operation();
if (!operation.has_value() || operation.value() !=
DataOperation::kReplace) {
auto snapshot_reader = std::make_unique<SnapshotReader>(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);
Review Comment:
The PR description (and the updated test expectation) calls out a new v1/v2
error message: \"Delete files are only supported in changelog scans of format
version 3 tables\". This implementation appends extra context (\", but the
table uses format version {}\"), which will likely break callers/tests that
match the message exactly. If you want to keep the detailed context, consider
adjusting the matcher to a substring/regex; otherwise, make the thrown message
exactly match the documented string and rely on logs for the extra details.
##########
src/iceberg/test/incremental_changelog_scan_test.cc:
##########
@@ -43,11 +45,19 @@ const std::string& TaskFilePath(const
std::shared_ptr<ChangelogScanTask>& task)
if (auto deleted = std::dynamic_pointer_cast<DeletedDataFileScanTask>(task))
{
return deleted->data_file()->file_path;
}
+ if (auto deleted_rows =
std::dynamic_pointer_cast<DeletedRowsScanTask>(task)) {
+ return deleted_rows->data_file()->file_path;
+ }
static const std::string empty_path;
return empty_path;
}
+std::vector<std::string> FilePaths(const
std::vector<std::shared_ptr<DataFile>>& files) {
+ return files | std::views::transform([](const auto& file) { return
file->file_path; }) |
+ std::ranges::to<std::vector<std::string>>();
Review Comment:
This test helper uses `std::ranges::to`, which requires C++23. If the test
target is compiled under C++20, it won’t build. Consider rewriting this helper
using a simple loop/`std::transform` into a pre-sized
`std::vector<std::string>` to keep tests compatible with the project’s language
standard.
##########
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 <algorithm>
+#include <utility>
+
+#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<std::unique_ptr<DeletionVectorChangelogPlanner>>
+DeletionVectorChangelogPlanner::Make(
+ std::shared_ptr<FileIO> io, std::shared_ptr<Schema> schema,
+ const std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>>&
specs_by_id,
+ std::shared_ptr<Expression> filter, bool case_sensitive, OptionalExecutor
executor,
+ const std::vector<ManifestFile>& delete_manifests,
+ const std::unordered_map<int64_t, int64_t>& sequence_numbers) {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto entries,
+ ParallelCollect(
+ executor, delete_manifests,
+ [&](const ManifestFile& manifest) ->
Result<std::vector<ManifestEntry>> {
+ 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<Expression> 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<ManifestEntry>{};
+ }
+ }
+
+ 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<DeletionVectorChangelogPlanner>(
+ 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::LatestDeleteFile>&
+DeletionVectorChangelogPlanner::LatestFor(
+ PartitionMap<std::optional<LatestDeleteFile>>& 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<LatestDeleteFile>&
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<DataFile> 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<std::string, std::vector<int64_t>>
+DeletionVectorChangelogPlanner::AddedDeletionVectors(
+ const std::unordered_set<int64_t>& sequence_numbers) const {
+ std::unordered_map<std::string, std::vector<int64_t>> 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<std::shared_ptr<DataFile>>
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<DataFile> 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<std::optional<LatestDeleteFile>>&
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);
Review Comment:
The new tests in `incremental_changelog_scan_test.cc` assert a plain message
(e.g., \"Equality delete files are not supported in changelog scans\"), but
this implementation includes additional details (\": <file> applies to
<file>\"). If `HasErrorMessage(...)` is an exact matcher, these tests will
fail. Either (a) change the error message to the stable prefix only, or (b)
update tests to match only the prefix/substring while still allowing the extra
diagnostic context.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
