From 412f6a4036e2df3a1753dc69ad0ee0694a321e70 Mon Sep 17 00:00:00 2001 From: Zehua Zou Date: Tue, 9 Jun 2026 10:52:03 +0800 Subject: [PATCH 1/2] feat: implement merge multiple DVs --- src/iceberg/CMakeLists.txt | 2 + src/iceberg/data/meson.build | 1 + src/iceberg/data/puffin_dv_register.cc | 79 ++++++++++ src/iceberg/data/puffin_dv_register.h | 36 +++++ src/iceberg/meson.build | 2 + src/iceberg/puffin_dv_io.cc | 56 ++++++++ src/iceberg/puffin_dv_io.h | 67 +++++++++ .../test/merging_snapshot_update_test.cc | 136 ++++++++++++++++++ src/iceberg/transaction.cc | 5 + src/iceberg/transaction.h | 1 + src/iceberg/update/merging_snapshot_update.cc | 112 +++++++++++++-- src/iceberg/update/merging_snapshot_update.h | 4 +- src/iceberg/util/struct_like_set.cc | 10 +- src/iceberg/util/struct_like_set.h | 3 + 14 files changed, 501 insertions(+), 13 deletions(-) create mode 100644 src/iceberg/data/puffin_dv_register.cc create mode 100644 src/iceberg/data/puffin_dv_register.h create mode 100644 src/iceberg/puffin_dv_io.cc create mode 100644 src/iceberg/puffin_dv_io.h diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 35b7b91b5..1cd1d4467 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -24,6 +24,7 @@ set(ICEBERG_SOURCES catalog/memory/in_memory_catalog.cc catalog/session_catalog.cc catalog/session_context.cc + puffin_dv_io.cc delete_file_index.cc expression/aggregate.cc expression/binder.cc @@ -196,6 +197,7 @@ set(ICEBERG_DATA_SOURCES data/data_writer.cc data/delete_filter.cc data/delete_loader.cc + data/puffin_dv_register.cc data/deletion_vector_writer.cc data/dv_util.cc data/equality_delete_writer.cc diff --git a/src/iceberg/data/meson.build b/src/iceberg/data/meson.build index eaae8a2b4..2ddac93df 100644 --- a/src/iceberg/data/meson.build +++ b/src/iceberg/data/meson.build @@ -20,6 +20,7 @@ install_headers( 'data_writer.h', 'delete_filter.h', 'delete_loader.h', + 'puffin_dv_register.h', 'deletion_vector_writer.h', 'equality_delete_writer.h', 'file_scan_task_reader.h', diff --git a/src/iceberg/data/puffin_dv_register.cc b/src/iceberg/data/puffin_dv_register.cc new file mode 100644 index 000000000..5d68bd0d7 --- /dev/null +++ b/src/iceberg/data/puffin_dv_register.cc @@ -0,0 +1,79 @@ +/* + * 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/data/puffin_dv_register.h" + +#include +#include +#include + +#include "iceberg/data/deletion_vector_writer.h" +#include "iceberg/data/dv_util_internal.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +class DataPuffinDVIO final : public PuffinDVIO { + public: + Result>> MergeAndWriteDVs( + std::span groups, std::string_view output_path, + const std::shared_ptr& io) override { + if (groups.empty()) { + return std::vector>{}; + } + + ICEBERG_ASSIGN_OR_RAISE( + auto writer, + DeletionVectorWriter::Make(DeletionVectorWriterOptions{ + .path = std::string(output_path), + .io = io, + .load_previous_deletes = [](std::string_view) + -> Result> { return std::nullopt; }})); + + for (const auto& group : groups) { + PositionDeleteIndex merged; + for (const auto& delete_file : group.delete_files) { + ICEBERG_ASSIGN_OR_RAISE(auto index, DVUtil::ReadDV(delete_file, io)); + merged.Merge(index); + } + ICEBERG_RETURN_UNEXPECTED(writer->Delete(group.referenced_data_file, merged, + group.spec, group.partition)); + } + + ICEBERG_RETURN_UNEXPECTED(writer->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto metadata, writer->Metadata()); + return std::move(metadata.data_files); + } +}; + +} // namespace + +std::shared_ptr MakePuffinDVIO() { + return std::make_shared(); +} + +void RegisterPuffinDVIO() { + static PuffinDVIORegistry registry( + []() -> Result> { return MakePuffinDVIO(); }); +} + +} // namespace iceberg diff --git a/src/iceberg/data/puffin_dv_register.h b/src/iceberg/data/puffin_dv_register.h new file mode 100644 index 000000000..fa942819d --- /dev/null +++ b/src/iceberg/data/puffin_dv_register.h @@ -0,0 +1,36 @@ +/* + * 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/data/puffin_dv_register.h +/// Puffin deletion vector I/O registration. + +#include + +#include "iceberg/iceberg_data_export.h" +#include "iceberg/puffin_dv_io.h" + +namespace iceberg { + +ICEBERG_DATA_EXPORT std::shared_ptr MakePuffinDVIO(); + +ICEBERG_DATA_EXPORT void RegisterPuffinDVIO(); + +} // namespace iceberg diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index e3bef01af..1de293cb5 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -127,6 +127,7 @@ iceberg_sources = files( 'partition_field.cc', 'partition_spec.cc', 'partition_summary.cc', + 'puffin_dv_io.cc', 'row/arrow_array_wrapper.cc', 'row/manifest_wrapper.cc', 'row/partition_values.cc', @@ -203,6 +204,7 @@ iceberg_data_sources = files( 'data/equality_delete_writer.cc', 'data/file_scan_task_reader.cc', 'data/position_delete_writer.cc', + 'data/puffin_dv_register.cc', 'data/writer.cc', 'deletes/position_delete_index.cc', 'deletes/position_delete_range_consumer.cc', diff --git a/src/iceberg/puffin_dv_io.cc b/src/iceberg/puffin_dv_io.cc new file mode 100644 index 000000000..477de630e --- /dev/null +++ b/src/iceberg/puffin_dv_io.cc @@ -0,0 +1,56 @@ +/* + * 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/puffin_dv_io.h" + +#include + +#include "iceberg/util/macros.h" + +namespace iceberg { +namespace { + +PuffinDVIOFactory GetNotImplementedFactory() { + return []() -> Result> { + return NotImplemented("Missing Puffin DV I/O factory"); + }; +} + +} // namespace + +PuffinDVIO::~PuffinDVIO() = default; + +PuffinDVIORegistry::PuffinDVIORegistry(PuffinDVIOFactory factory) { + GetFactory() = std::move(factory); +} + +PuffinDVIOFactory& PuffinDVIORegistry::GetFactory() { + static auto* factory = new PuffinDVIOFactory(GetNotImplementedFactory()); + return *factory; +} + +Result>> PuffinDVIORegistry::MergeAndWriteDVs( + std::span groups, std::string_view output_path, + const std::shared_ptr& io) { + ICEBERG_ASSIGN_OR_RAISE(auto dv_io, GetFactory()()); + ICEBERG_PRECHECK(dv_io != nullptr, "Puffin DV I/O factory returned null"); + return dv_io->MergeAndWriteDVs(groups, output_path, io); +} + +} // namespace iceberg diff --git a/src/iceberg/puffin_dv_io.h b/src/iceberg/puffin_dv_io.h new file mode 100644 index 000000000..9c42cae8e --- /dev/null +++ b/src/iceberg/puffin_dv_io.h @@ -0,0 +1,67 @@ +/* + * 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/puffin_dv_io.h +/// Deletion vector merge hooks. + +#include +#include +#include +#include +#include +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/type_fwd.h" + +namespace iceberg { + +struct ICEBERG_EXPORT DeletionVectorMergeGroup { + std::string referenced_data_file; + std::vector> delete_files; + std::shared_ptr spec; + PartitionValues partition; +}; + +class ICEBERG_EXPORT PuffinDVIO { + public: + virtual ~PuffinDVIO(); + + virtual Result>> MergeAndWriteDVs( + std::span groups, std::string_view output_path, + const std::shared_ptr& io) = 0; +}; + +using PuffinDVIOFactory = std::function>()>; + +struct ICEBERG_EXPORT PuffinDVIORegistry { + explicit PuffinDVIORegistry(PuffinDVIOFactory factory); + + static PuffinDVIOFactory& GetFactory(); + + static Result>> MergeAndWriteDVs( + std::span groups, std::string_view output_path, + const std::shared_ptr& io); +}; + +} // namespace iceberg diff --git a/src/iceberg/test/merging_snapshot_update_test.cc b/src/iceberg/test/merging_snapshot_update_test.cc index 5306ba1b3..6684cf221 100644 --- a/src/iceberg/test/merging_snapshot_update_test.cc +++ b/src/iceberg/test/merging_snapshot_update_test.cc @@ -38,6 +38,7 @@ #include "iceberg/manifest/manifest_reader.h" #include "iceberg/manifest/manifest_writer.h" #include "iceberg/partition_spec.h" +#include "iceberg/puffin_dv_io.h" #include "iceberg/row/partition_values.h" #include "iceberg/schema.h" #include "iceberg/snapshot.h" @@ -57,6 +58,55 @@ namespace iceberg { +class RecordingPuffinDVIO final : public PuffinDVIO { + public: + struct Call { + std::string output_path; + std::vector groups; + }; + + Result>> MergeAndWriteDVs( + std::span groups, std::string_view output_path, + const std::shared_ptr& /*io*/) override { + calls.push_back(Call{.output_path = std::string(output_path), + .groups = {groups.begin(), groups.end()}}); + + std::vector> result; + result.reserve(groups.size()); + for (const auto& group : groups) { + auto file = std::make_shared(*group.delete_files.front()); + file->file_path = std::string(output_path); + file->file_format = FileFormatType::kPuffin; + file->referenced_data_file = group.referenced_data_file; + file->record_count = 0; + for (const auto& delete_file : group.delete_files) { + file->record_count += delete_file->record_count; + } + file->content_offset = static_cast(result.size()) * 100; + file->content_size_in_bytes = 100; + result.push_back(std::move(file)); + } + return result; + } + + std::vector calls; +}; + +class ScopedPuffinDVIORegistry { + public: + explicit ScopedPuffinDVIORegistry(PuffinDVIOFactory factory) + : previous_factory_(std::move(PuffinDVIORegistry::GetFactory())) { + PuffinDVIORegistry::GetFactory() = std::move(factory); + } + + ~ScopedPuffinDVIORegistry() { + PuffinDVIORegistry::GetFactory() = std::move(previous_factory_); + } + + private: + PuffinDVIOFactory previous_factory_; +}; + /// \brief Concrete subclass of MergingSnapshotUpdate for testing. class TestMergeAppend : public MergingSnapshotUpdate { public: @@ -264,6 +314,20 @@ class MergingSnapshotUpdateTest : public MinimalUpdateTestBase { return f; } + std::shared_ptr MakeDeletionVector(const std::string& path, + const std::shared_ptr& data_file, + int64_t record_count) { + auto f = MakeDeleteFile(path, 0); + f->file_format = FileFormatType::kPuffin; + f->partition = data_file->partition; + f->partition_spec_id = data_file->partition_spec_id; + f->referenced_data_file = data_file->file_path; + f->record_count = record_count; + f->content_offset = 0; + f->content_size_in_bytes = 100; + return f; + } + std::shared_ptr MakeEqualityDeleteFile(const std::string& path, int64_t partition_x) { auto f = MakeDeleteFile(path, partition_x); @@ -978,6 +1042,78 @@ TEST_F(MergingSnapshotUpdateTest, ValidateNewDeleteFileV3AllowsDeletionVector) { EXPECT_THAT(op->AddDelete(del_file), IsOk()); } +TEST_F(MergingSnapshotUpdateTest, ApplyMergesDuplicateDeletionVectors) { + SetTableFormatVersion(3); + + auto dv_io = std::make_shared(); + ScopedPuffinDVIORegistry registry( + [dv_io]() -> Result> { return dv_io; }); + ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend()); + + auto dv_a1 = MakeDeletionVector("/delete/dv_a1.puffin", file_a_, 1); + auto dv_a2 = MakeDeletionVector("/delete/dv_a2.puffin", file_a_, 2); + auto dv_b = MakeDeletionVector("/delete/dv_b.puffin", file_b_, 3); + + EXPECT_THAT(op->AddDelete(dv_a1, 7), IsOk()); + EXPECT_THAT(op->AddDelete(dv_a2, 7), IsOk()); + EXPECT_THAT(op->AddDelete(dv_b, 8), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto manifests, op->Apply(*table_->metadata(), nullptr)); + auto delete_manifest_it = + std::ranges::find_if(manifests, [](const ManifestFile& manifest) { + return manifest.content == ManifestContent::kDeletes; + }); + ASSERT_NE(delete_manifest_it, manifests.end()); + ICEBERG_UNWRAP_OR_FAIL(auto entries, + ReadAllEntries(std::vector{*delete_manifest_it}, + *table_->metadata())); + + ASSERT_EQ(dv_io->calls.size(), 1U); + EXPECT_THAT(dv_io->calls[0].output_path, ::testing::HasSubstr("/data/merged-dvs-")); + ASSERT_EQ(dv_io->calls[0].groups.size(), 1U); + EXPECT_EQ(dv_io->calls[0].groups[0].referenced_data_file, file_a_->file_path); + EXPECT_THAT(dv_io->calls[0].groups[0].delete_files, ::testing::SizeIs(2)); + + ASSERT_EQ(entries.size(), 2U); + auto merged_it = std::ranges::find_if(entries, [&](const ManifestEntry& entry) { + return entry.data_file->referenced_data_file == file_a_->file_path; + }); + ASSERT_NE(merged_it, entries.end()); + EXPECT_THAT(merged_it->data_file->file_path, ::testing::HasSubstr("/data/merged-dvs-")); + EXPECT_EQ(merged_it->data_file->record_count, 3); + ASSERT_TRUE(merged_it->sequence_number.has_value()); + EXPECT_EQ(*merged_it->sequence_number, 7); + + auto single_it = std::ranges::find_if(entries, [&](const ManifestEntry& entry) { + return entry.data_file->referenced_data_file == file_b_->file_path; + }); + ASSERT_NE(single_it, entries.end()); + EXPECT_EQ(single_it->data_file->file_path, dv_b->file_path); + ASSERT_TRUE(single_it->sequence_number.has_value()); + EXPECT_EQ(*single_it->sequence_number, 8); +} + +TEST_F(MergingSnapshotUpdateTest, ApplyMergesDuplicateDeletionVectorsWithNullPartition) { + SetTableFormatVersion(3); + + auto dv_io = std::make_shared(); + ScopedPuffinDVIORegistry registry( + [dv_io]() -> Result> { return dv_io; }); + ICEBERG_UNWRAP_OR_FAIL(auto op, NewMergeAppend()); + + file_a_->partition = PartitionValues({Literal::Null(int64())}); + auto dv_a1 = MakeDeletionVector("/delete/dv_a1.puffin", file_a_, 1); + auto dv_a2 = MakeDeletionVector("/delete/dv_a2.puffin", file_a_, 2); + + EXPECT_THAT(op->AddDelete(dv_a1, 7), IsOk()); + EXPECT_THAT(op->AddDelete(dv_a2, 7), IsOk()); + + EXPECT_THAT(op->Apply(*table_->metadata(), nullptr), IsOk()); + ASSERT_EQ(dv_io->calls.size(), 1U); + ASSERT_EQ(dv_io->calls[0].groups.size(), 1U); + EXPECT_EQ(dv_io->calls[0].groups[0].referenced_data_file, file_a_->file_path); +} + TEST_F(MergingSnapshotUpdateTest, ValidateNewDeleteFileRejectsUnsupportedVersion) { SetTableFormatVersion(TableMetadata::kSupportedTableFormatVersion + 1); diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc index 7abce27cb..8889d6eaa 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -23,6 +23,7 @@ #include #include "iceberg/catalog.h" +#include "iceberg/location_provider.h" #include "iceberg/schema.h" #include "iceberg/snapshot.h" #include "iceberg/statistics_file.h" @@ -86,6 +87,10 @@ const TableMetadata& TransactionContext::current() const { return metadata_builder->current(); } +Result> TransactionContext::LocationProvider() const { + return iceberg::LocationProvider::Make(current().location, current().properties); +} + std::string TransactionContext::MetadataFileLocation(std::string_view filename) const { const auto metadata_location = current().properties.Get(TableProperties::kWriteMetadataLocation); diff --git a/src/iceberg/transaction.h b/src/iceberg/transaction.h index 284222351..65e990b76 100644 --- a/src/iceberg/transaction.h +++ b/src/iceberg/transaction.h @@ -188,6 +188,7 @@ class ICEBERG_EXPORT TransactionContext { const TableMetadata* base() const; const TableMetadata& current() const; std::string MetadataFileLocation(std::string_view filename) const; + Result> LocationProvider() const; std::shared_ptr table; std::unique_ptr metadata_builder; diff --git a/src/iceberg/update/merging_snapshot_update.cc b/src/iceberg/update/merging_snapshot_update.cc index f1f0f511d..af869e2f4 100644 --- a/src/iceberg/update/merging_snapshot_update.cc +++ b/src/iceberg/update/merging_snapshot_update.cc @@ -21,10 +21,13 @@ #include #include +#include +#include #include #include #include #include +#include #include #include "iceberg/constants.h" @@ -32,6 +35,8 @@ #include "iceberg/expression/expressions.h" #include "iceberg/expression/manifest_evaluator.h" #include "iceberg/expression/projections.h" +#include "iceberg/file_io.h" +#include "iceberg/location_provider.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/manifest/manifest_group.h" #include "iceberg/manifest/manifest_list.h" @@ -39,6 +44,7 @@ #include "iceberg/manifest/manifest_util_internal.h" #include "iceberg/manifest/manifest_writer.h" #include "iceberg/partition_spec.h" +#include "iceberg/puffin_dv_io.h" #include "iceberg/schema.h" #include "iceberg/snapshot.h" #include "iceberg/table.h" @@ -48,6 +54,7 @@ #include "iceberg/util/content_file_util.h" #include "iceberg/util/macros.h" #include "iceberg/util/snapshot_util_internal.h" +#include "iceberg/util/struct_like_set.h" namespace iceberg { @@ -751,9 +758,11 @@ Result> MergingSnapshotUpdate::WriteNewDataManifests() } Result> -MergingSnapshotUpdate::MergeDVs() const { +MergingSnapshotUpdate::MergeDVs() { std::vector result; result.reserve(dvs_by_referenced_file_.size()); + std::vector groups; + std::vector> data_sequence_numbers; for (const auto& entry : dvs_by_referenced_file_.entries()) { const auto& referenced_file = entry.referenced_file; @@ -761,16 +770,85 @@ MergingSnapshotUpdate::MergeDVs() const { if (dvs.empty()) { continue; } - if (dvs.size() > 1) { - // TODO(Guotao): Merge duplicate DVs for one referenced data file once C++ - // has DVUtil/Puffin DV rewriting; Java merges them before writing manifests. - return NotImplemented( - "Merging multiple deletion vectors is not supported yet for referenced " - "data file: {}", - referenced_file); + if (dvs.size() == 1) { + result.push_back(dvs.front()); + continue; + } + + const auto& first_file = dvs.front().file; + const auto first_data_sequence_number = dvs.front().data_sequence_number; + const auto first_spec_id = first_file->partition_spec_id; + ICEBERG_PRECHECK(first_spec_id.has_value(), "DV must have a partition spec ID: {}", + first_file->file_path); + + for (const auto& dv : dvs) { + ICEBERG_PRECHECK(dv.data_sequence_number == first_data_sequence_number, + "Cannot merge DVs, mismatched sequence numbers for {}", + referenced_file); + ICEBERG_PRECHECK(dv.file->partition_spec_id == first_spec_id, + "Cannot merge DVs, mismatched partition specs for {}", + referenced_file); + // TODO(gangwu): Java DVUtil.validateCanMerge compares partitions with + // Comparators.forType(spec.partitionType()). Use a typed C++ comparator if + // StructLikeEqual's scalar equality is not enough for future partition types. + ICEBERG_ASSIGN_OR_RAISE(auto same_partition, + StructLikeEqual(dv.file->partition, first_file->partition)); + ICEBERG_PRECHECK(same_partition, + "Cannot merge DVs, mismatched partition tuples for {}", + referenced_file); + } + + ICEBERG_ASSIGN_OR_RAISE(auto spec, base().PartitionSpecById(*first_spec_id)); + DeletionVectorMergeGroup group{ + .referenced_data_file = referenced_file, + .spec = std::move(spec), + .partition = first_file->partition, + }; + group.delete_files.reserve(dvs.size()); + for (const auto& dv : dvs) { + group.delete_files.push_back(dv.file); } + groups.push_back(std::move(group)); + data_sequence_numbers.push_back(first_data_sequence_number); + } + + if (groups.empty()) { + return result; + } + + ICEBERG_ASSIGN_OR_RAISE(auto location_provider, ctx_->LocationProvider()); + auto output_path = location_provider->NewDataLocation( + std::format("merged-dvs-{}-{}.puffin", SnapshotId(), ++dv_merge_attempt_)); - result.push_back(dvs.front()); + auto merged_files = + PuffinDVIORegistry::MergeAndWriteDVs(groups, output_path, ctx_->table->io()); + if (!merged_files) { + std::ignore = DeleteFile(output_path); + return std::unexpected(std::move(merged_files.error())); + } + + std::unordered_map> merged_by_ref; + for (auto& file : merged_files.value()) { + ICEBERG_PRECHECK(file != nullptr, "Merged DV file must not be null"); + ICEBERG_PRECHECK(file->referenced_data_file.has_value(), + "Merged DV must have a referenced data file: {}", file->file_path); + const auto referenced_data_file = *file->referenced_data_file; + auto insert_result = merged_by_ref.emplace(referenced_data_file, std::move(file)); + ICEBERG_PRECHECK(insert_result.second, "Duplicate merged DV for {}", + referenced_data_file); + } + ICEBERG_PRECHECK(merged_by_ref.size() == groups.size(), + "Expected {} merged DVs but got {}", groups.size(), + merged_by_ref.size()); + + for (size_t pos = 0; pos < groups.size(); ++pos) { + auto iter = merged_by_ref.find(groups[pos].referenced_data_file); + ICEBERG_CHECK(iter != merged_by_ref.end(), "Missing merged DV for {}", + groups[pos].referenced_data_file); + PendingDeleteFile merged{.file = std::move(iter->second), + .data_sequence_number = data_sequence_numbers[pos]}; + merged_dvs_.push_back(merged); + result.push_back(std::move(merged)); } return result; @@ -783,6 +861,13 @@ Result> MergingSnapshotUpdate::WriteNewDeleteManifests std::ignore = DeleteFile(m.manifest_path); } cached_new_delete_manifests_.clear(); + std::unordered_set deleted_merged_dv_paths; + for (const auto& dv : merged_dvs_) { + if (deleted_merged_dv_paths.insert(dv.file->file_path).second) { + std::ignore = DeleteFile(dv.file->file_path); + } + } + merged_dvs_.clear(); added_delete_files_summary_.Clear(); } @@ -988,6 +1073,15 @@ Status MergingSnapshotUpdate::CleanUncommittedAppends( // rewritten_append_manifests_ are always owned by the table. ICEBERG_RETURN_UNEXPECTED( DeleteUncommitted(rewritten_append_manifests_, committed, /*clear=*/true)); + if (committed.empty()) { + std::unordered_set deleted_merged_dv_paths; + for (const auto& dv : merged_dvs_) { + if (deleted_merged_dv_paths.insert(dv.file->file_path).second) { + std::ignore = DeleteFile(dv.file->file_path); + } + } + } + merged_dvs_.clear(); // append_manifests_ are only owned by the table if the commit succeeded. if (!committed.empty()) { diff --git a/src/iceberg/update/merging_snapshot_update.h b/src/iceberg/update/merging_snapshot_update.h index dbfb79937..1c08a0667 100644 --- a/src/iceberg/update/merging_snapshot_update.h +++ b/src/iceberg/update/merging_snapshot_update.h @@ -347,7 +347,7 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { void SetSummaryProperty(const std::string& property, const std::string& value) override; - Result> MergeDVs() const; + Result> MergeDVs(); /// \brief Write new data manifests for staged data files; caches the result. Result> WriteNewDataManifests(); @@ -396,6 +396,8 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { std::vector cached_new_data_manifests_; std::vector cached_new_delete_manifests_; + std::vector merged_dvs_; + int32_t dv_merge_attempt_ = 0; }; } // namespace iceberg diff --git a/src/iceberg/util/struct_like_set.cc b/src/iceberg/util/struct_like_set.cc index 35a0f9e28..fe622c8e5 100644 --- a/src/iceberg/util/struct_like_set.cc +++ b/src/iceberg/util/struct_like_set.cc @@ -365,7 +365,7 @@ Status ValidateRowAgainstTypes(const StructLike& row, return {}; } -Result StructLikeEqual(const StructLike& lhs, const StructLike& rhs) { +Result StructLikeEqualInternal(const StructLike& lhs, const StructLike& rhs) { if (lhs.num_fields() != rhs.num_fields()) { return false; } @@ -417,7 +417,7 @@ Result MapLikeEqual(const MapLike& lhs, const MapLike& rhs) { } bool StructLikeEqualUnchecked(const StructLike& lhs, const StructLike& rhs) noexcept { - auto result = StructLikeEqual(lhs, rhs); + auto result = StructLikeEqualInternal(lhs, rhs); ICEBERG_DCHECK(result.has_value(), "Validated StructLike equality must not fail"); return result.value_or(false); } @@ -455,7 +455,7 @@ Result ScalarEqual(const Scalar& lhs, const Scalar& rhs) { const auto& r = std::get>(other); if (!l && !r) return true; if (!l || !r) return false; - return StructLikeEqual(*l, *r); + return StructLikeEqualInternal(*l, *r); }, [](const std::shared_ptr& l, const Scalar& other) -> Result { const auto& r = std::get>(other); @@ -475,6 +475,10 @@ Result ScalarEqual(const Scalar& lhs, const Scalar& rhs) { } // namespace +Result StructLikeEqual(const StructLike& lhs, const StructLike& rhs) { + return StructLikeEqualInternal(lhs, rhs); +} + template StructLikeSet::StructLikeSet(const StructType& type, size_t arena_initial_size) : arena_(arena_initial_size) { diff --git a/src/iceberg/util/struct_like_set.h b/src/iceberg/util/struct_like_set.h index 1dda0621c..db263c06d 100644 --- a/src/iceberg/util/struct_like_set.h +++ b/src/iceberg/util/struct_like_set.h @@ -34,6 +34,9 @@ namespace iceberg { +/// \brief Compare two StructLike rows by scalar values. +ICEBERG_EXPORT Result StructLikeEqual(const StructLike& lhs, const StructLike& rhs); + /// \brief A set of StructLike rows with type-aware hashing and equality. /// /// As StructLike uses view semantics, this set makes deep copies of inserted rows From afa317505dbf6872aaebbe320e6f0ef13a09f854 Mon Sep 17 00:00:00 2001 From: Gang Wu Date: Wed, 15 Jul 2026 23:36:51 +0800 Subject: [PATCH 2/2] fix ci --- .clang-tidy | 3 - .github/workflows/aws_test.yml | 75 ------------------- src/iceberg/transaction.cc | 3 +- src/iceberg/transaction.h | 2 +- src/iceberg/update/merging_snapshot_update.cc | 2 +- 5 files changed, 4 insertions(+), 81 deletions(-) diff --git a/.clang-tidy b/.clang-tidy index 229bb0ec1..e7e8abd33 100644 --- a/.clang-tidy +++ b/.clang-tidy @@ -21,7 +21,6 @@ Checks: | clang-analyzer-*, google-*, modernize-*, - misc-include-cleaner, readability-identifier-naming, readability-isolate-declaration, -modernize-use-nodiscard, @@ -42,7 +41,5 @@ CheckOptions: value: '_' - key: modernize-use-scoped-lock.WarnOnSingleLocks value: 'false' - - key: misc-include-cleaner.IgnoreHeaders - value: 'arrow/.*;avro/.*;aws/.*;cpr/.*;gmock/.*;gtest/.*;nanoarrow/.*;nlohmann/.*;parquet/.*;roaring/.*;spdlog/.*;sqlpp23/.*;thrift/.*' HeaderFilterRegex: 'src/iceberg|example' diff --git a/.github/workflows/aws_test.yml b/.github/workflows/aws_test.yml index 3e955d84f..0ed9685b7 100644 --- a/.github/workflows/aws_test.yml +++ b/.github/workflows/aws_test.yml @@ -137,78 +137,3 @@ jobs: with: path: ${{ github.workspace }}/.sccache key: sccache-aws-${{ matrix.runs-on }}-bundle${{ matrix.bundle_awssdk }}-s3${{ matrix.s3 }}-sigv4${{ matrix.sigv4 }}-${{ github.run_id }} - - # Exercise the Meson build with SigV4 enabled (resolves aws-cpp-sdk-core via - # its CMake config, not pkg-config whose Cflags force -std=c++11). - meson-sigv4: - if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }} - name: Meson SigV4 (AMD64 Ubuntu 26.04) - runs-on: ubuntu-26.04 - timeout-minutes: 45 - env: - SCCACHE_DIR: ${{ github.workspace }}/.sccache - SCCACHE_CACHE_SIZE: "2G" - steps: - - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0 - with: - python-version: '3.x' - - name: Checkout iceberg-cpp - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0 - with: - persist-credentials: false - - name: Install build dependencies - shell: bash - run: | - sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev libjitterentropy3-dev - python3 -m pip install --upgrade pip - python3 -m pip install -r requirements.txt - - name: Cache vcpkg packages - uses: actions/cache@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 - id: vcpkg-cache - with: - path: /usr/local/share/vcpkg/installed - key: vcpkg-x64-linux-aws-sdk-cpp-core-${{ hashFiles('.github/workflows/aws_test.yml') }} - - name: Install AWS SDK via vcpkg - if: ${{ steps.vcpkg-cache.outputs.cache-hit != 'true' }} - shell: bash - # Retry to ride out transient GitHub/mirror download failures (504s). - run: | - for attempt in 1 2 3; do - if vcpkg install aws-sdk-cpp[core]:x64-linux; then - exit 0 - fi - echo "::warning::vcpkg install failed (attempt ${attempt}/3), retrying in 30s" - sleep 30 - done - echo "::error::vcpkg install failed after 3 attempts" - exit 1 - - name: Set Ubuntu Compilers - # Wrap the compiler with sccache: Meson uses an explicit CC/CXX verbatim, - # so the launcher must be prepended here to route compiles through sccache. - run: | - echo "CC=sccache gcc-14" >> $GITHUB_ENV - echo "CXX=sccache g++-14" >> $GITHUB_ENV - - name: Restore sccache cache - uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 - with: - path: ${{ github.workspace }}/.sccache - key: sccache-meson-sigv4-${{ github.run_id }} - restore-keys: | - sccache-meson-sigv4- - - name: Setup sccache - uses: mozilla-actions/sccache-action@9e7fa8a12102821edf02ca5dbea1acd0f89a2696 # v0.0.10 - - name: Build and test Iceberg - shell: bash - env: - CMAKE_PREFIX_PATH: /usr/local/share/vcpkg/installed/x64-linux - run: | - meson setup builddir -Dsigv4=enabled - meson compile -C builddir - sccache --show-stats - meson test -C builddir --timeout-multiplier 0 --print-errorlogs - - name: Save sccache cache - if: github.ref == 'refs/heads/main' - uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6.1.0 - with: - path: ${{ github.workspace }}/.sccache - key: sccache-meson-sigv4-${{ github.run_id }} diff --git a/src/iceberg/transaction.cc b/src/iceberg/transaction.cc index 8889d6eaa..503121658 100644 --- a/src/iceberg/transaction.cc +++ b/src/iceberg/transaction.cc @@ -87,7 +87,8 @@ const TableMetadata& TransactionContext::current() const { return metadata_builder->current(); } -Result> TransactionContext::LocationProvider() const { +Result> TransactionContext::NewLocationProvider() + const { return iceberg::LocationProvider::Make(current().location, current().properties); } diff --git a/src/iceberg/transaction.h b/src/iceberg/transaction.h index 65e990b76..42a504565 100644 --- a/src/iceberg/transaction.h +++ b/src/iceberg/transaction.h @@ -188,7 +188,7 @@ class ICEBERG_EXPORT TransactionContext { const TableMetadata* base() const; const TableMetadata& current() const; std::string MetadataFileLocation(std::string_view filename) const; - Result> LocationProvider() const; + Result> NewLocationProvider() const; std::shared_ptr
table; std::unique_ptr metadata_builder; diff --git a/src/iceberg/update/merging_snapshot_update.cc b/src/iceberg/update/merging_snapshot_update.cc index af869e2f4..09e495cf1 100644 --- a/src/iceberg/update/merging_snapshot_update.cc +++ b/src/iceberg/update/merging_snapshot_update.cc @@ -816,7 +816,7 @@ MergingSnapshotUpdate::MergeDVs() { return result; } - ICEBERG_ASSIGN_OR_RAISE(auto location_provider, ctx_->LocationProvider()); + ICEBERG_ASSIGN_OR_RAISE(auto location_provider, ctx_->NewLocationProvider()); auto output_path = location_provider->NewDataLocation( std::format("merged-dvs-{}-{}.puffin", SnapshotId(), ++dv_merge_attempt_));