diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 280202ad3..f605e57ad 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -107,6 +107,7 @@ set(ICEBERG_SOURCES update/expire_snapshots.cc update/fast_append.cc update/merge_append.cc + update/replace_partitions.cc update/merging_snapshot_update.cc update/overwrite_files.cc update/pending_update.cc diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index 1de293cb5..de39e53d7 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -159,6 +159,7 @@ iceberg_sources = files( 'update/merging_snapshot_update.cc', 'update/overwrite_files.cc', 'update/pending_update.cc', + 'update/replace_partitions.cc', 'update/rewrite_files.cc', 'update/row_delta.cc', 'update/set_snapshot.cc', diff --git a/src/iceberg/partition_spec.cc b/src/iceberg/partition_spec.cc index 96287fb87..90be924b2 100644 --- a/src/iceberg/partition_spec.cc +++ b/src/iceberg/partition_spec.cc @@ -66,6 +66,12 @@ int32_t PartitionSpec::spec_id() const { return spec_id_; } std::span PartitionSpec::fields() const { return fields_; } +bool PartitionSpec::IsUnpartitioned() const { + return std::ranges::all_of(fields_, [](const PartitionField& field) { + return field.transform()->transform_type() == TransformType::kVoid; + }); +} + Result> PartitionSpec::PartitionType( const Schema& schema) const { if (fields_.empty()) { diff --git a/src/iceberg/partition_spec.h b/src/iceberg/partition_spec.h index 2ed0ddc8a..e9c728dc0 100644 --- a/src/iceberg/partition_spec.h +++ b/src/iceberg/partition_spec.h @@ -61,6 +61,9 @@ class ICEBERG_EXPORT PartitionSpec : public util::Formattable { /// \brief Get a list view of the partition fields. std::span fields() const; + /// \brief Return whether this spec has no non-void partition fields. + bool IsUnpartitioned() const; + /// \brief Get the partition type binding to the input schema. Result> PartitionType(const Schema& schema) const; diff --git a/src/iceberg/snapshot.h b/src/iceberg/snapshot.h index e32f3a715..9dc162be0 100644 --- a/src/iceberg/snapshot.h +++ b/src/iceberg/snapshot.h @@ -254,6 +254,8 @@ struct ICEBERG_EXPORT SnapshotSummaryFields { inline static const std::string kEngineName = "engine-name"; /// \brief Version of the engine that created the snapshot inline static const std::string kEngineVersion = "engine-version"; + /// \brief Whether this is a replace-partitions operation + inline static const std::string kReplacePartitions = "replace-partitions"; }; /// \brief Helper class for building snapshot summaries. diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 55a7c2b85..14313b08f 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -237,7 +237,8 @@ if(ICEBERG_BUILD_BUNDLE) update_schema_test.cc update_sort_order_test.cc update_statistics_test.cc - overwrite_files_test.cc) + overwrite_files_test.cc + replace_partitions_test.cc) add_iceberg_test(data_test USE_BUNDLE diff --git a/src/iceberg/test/partition_spec_test.cc b/src/iceberg/test/partition_spec_test.cc index da1724e49..9a31b065a 100644 --- a/src/iceberg/test/partition_spec_test.cc +++ b/src/iceberg/test/partition_spec_test.cc @@ -60,6 +60,20 @@ TEST(PartitionSpecTest, Basics) { } } +TEST(PartitionSpecTest, IsUnpartitioned) { + ICEBERG_UNWRAP_OR_FAIL(auto empty_spec, PartitionSpec::Make(100, {})); + EXPECT_TRUE(empty_spec->IsUnpartitioned()); + + PartitionField void_field(5, 1000, "void", Transform::Void()); + ICEBERG_UNWRAP_OR_FAIL(auto all_void_spec, PartitionSpec::Make(101, {void_field})); + EXPECT_TRUE(all_void_spec->IsUnpartitioned()); + + PartitionField identity_field(7, 1001, "identity", Transform::Identity()); + ICEBERG_UNWRAP_OR_FAIL(auto partitioned_spec, + PartitionSpec::Make(102, {void_field, identity_field})); + EXPECT_FALSE(partitioned_spec->IsUnpartitioned()); +} + TEST(PartitionSpecTest, Equality) { SchemaField field1(5, "ts", timestamp(), true); SchemaField field2(7, "bar", string(), true); diff --git a/src/iceberg/test/replace_partitions_test.cc b/src/iceberg/test/replace_partitions_test.cc new file mode 100644 index 000000000..86f4f20a2 --- /dev/null +++ b/src/iceberg/test/replace_partitions_test.cc @@ -0,0 +1,554 @@ +/* + * 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/update/replace_partitions.h" + +#include +#include +#include + +#include +#include + +#include "iceberg/avro/avro_register.h" +#include "iceberg/expression/expressions.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/manifest/manifest_reader.h" +#include "iceberg/partition_field.h" +#include "iceberg/partition_spec.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/schema.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" +#include "iceberg/table_metadata.h" +#include "iceberg/table_properties.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/update_test_base.h" +#include "iceberg/transaction.h" +#include "iceberg/transform.h" +#include "iceberg/update/delete_files.h" +#include "iceberg/update/fast_append.h" +#include "iceberg/update/row_delta.h" +#include "iceberg/update/update_properties.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +// The base table (TableMetadataV2ValidMinimal.json) has schema {x: long (id 1), +// y: long (id 2), z: long (id 3)} and partitions by identity(x) as spec 0. +class ReplacePartitionsTest : public UpdateTestBase { + protected: + static void SetUpTestSuite() { avro::RegisterAll(); } + + std::string MetadataResource() const override { + return "TableMetadataV2ValidMinimal.json"; + } + + void SetUp() override { + UpdateTestBase::SetUp(); + + ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec()); + ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema()); + + file_a_ = MakeDataFile("/data/file_a.parquet", /*partition_x=*/1L); + file_b_ = MakeDataFile("/data/file_b.parquet", /*partition_x=*/2L); + } + + std::shared_ptr MakeDataFile(const std::string& path, int64_t partition_x) { + auto f = std::make_shared(); + f->content = DataFile::Content::kData; + f->file_path = table_location_ + path; + f->file_format = FileFormatType::kParquet; + f->partition = PartitionValues(std::vector{Literal::Long(partition_x)}); + f->file_size_in_bytes = 1024; + f->record_count = 100; + f->partition_spec_id = spec_->spec_id(); + return f; + } + + // A data file that belongs to an unpartitioned spec (empty partition tuple). + std::shared_ptr MakeUnpartitionedFile(const std::string& path, + int32_t spec_id) { + auto f = std::make_shared(); + f->content = DataFile::Content::kData; + f->file_path = table_location_ + path; + f->file_format = FileFormatType::kParquet; + f->partition = PartitionValues(std::vector{}); + f->file_size_in_bytes = 1024; + f->record_count = 100; + f->partition_spec_id = spec_id; + return f; + } + + std::shared_ptr MakeEqualityDeleteFile(const std::string& path, + int64_t partition_x) { + auto f = MakeDataFile(path, partition_x); + f->content = DataFile::Content::kEqualityDeletes; + f->equality_ids = {1}; + return f; + } + + std::shared_ptr MakeDeletionVector(const std::string& path, + const std::string& referenced_data_file, + int64_t partition_x) { + auto f = MakeDataFile(path, partition_x); + f->content = DataFile::Content::kPositionDeletes; + f->file_format = FileFormatType::kPuffin; + f->referenced_data_file = referenced_data_file; + f->content_offset = 0; + f->content_size_in_bytes = 10; + f->record_count = 1; + return f; + } + + // Add an extra spec to the in-memory metadata so staged files can resolve it. + // The empty field list makes it unpartitioned (PartitionSpec::IsUnpartitioned()). + void AddUnpartitionedSpec(int32_t spec_id) { + ICEBERG_UNWRAP_OR_FAIL(auto spec, + PartitionSpec::Make(*schema_, spec_id, {}, + /*allow_missing_fields=*/false)); + table_->metadata()->partition_specs.push_back( + std::shared_ptr(std::move(spec))); + ASSERT_THAT(table_->metadata()->PartitionSpecById(spec_id), IsOk()); + } + + void AddIdentitySpec(int32_t spec_id, int32_t field_id, const std::string& name) { + ICEBERG_UNWRAP_OR_FAIL( + auto spec, PartitionSpec::Make(*schema_, spec_id, + {PartitionField(/*source_id=*/1, field_id, name, + Transform::Identity())}, + /*allow_missing_fields=*/false)); + table_->metadata()->partition_specs.push_back( + std::shared_ptr(std::move(spec))); + ASSERT_THAT(table_->metadata()->PartitionSpecById(spec_id), IsOk()); + } + + // Add a spec whose only field uses the void transform. Such a spec has a field + // but is still unpartitioned, so a file staged against it drives a table-wide + // replace rather than a partition drop. + void AddAllVoidSpec(int32_t spec_id, int32_t field_id, const std::string& name) { + ICEBERG_UNWRAP_OR_FAIL( + auto spec, PartitionSpec::Make(*schema_, spec_id, + {PartitionField(/*source_id=*/1, field_id, name, + Transform::Void())}, + /*allow_missing_fields=*/false)); + table_->metadata()->partition_specs.push_back( + std::shared_ptr(std::move(spec))); + ASSERT_THAT(table_->metadata()->PartitionSpecById(spec_id), IsOk()); + } + + // A data file for an all-void spec: one partition field holding a null value. + std::shared_ptr MakeAllVoidFile(const std::string& path, int32_t spec_id) { + auto f = std::make_shared(); + f->content = DataFile::Content::kData; + f->file_path = table_location_ + path; + f->file_format = FileFormatType::kParquet; + f->partition = PartitionValues(std::vector{Literal::Null(int64())}); + f->file_size_in_bytes = 1024; + f->record_count = 100; + f->partition_spec_id = spec_id; + return f; + } + + // Commit a DeleteFiles op that removes an existing data file by path. + void CommitDeleteDataFile(const std::string& path) { + ICEBERG_UNWRAP_OR_FAIL(auto delete_files, table_->NewDeleteFiles()); + delete_files->DeleteFile(path); + EXPECT_THAT(delete_files->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + } + + // Live (non-deleted) data file paths across the current snapshot's manifests. + Result> LiveDataFilePaths() { + std::vector paths; + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot()); + SnapshotCache cache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, cache.DataManifests(file_io_)); + for (const auto& manifest : manifests) { + ICEBERG_ASSIGN_OR_RAISE( + auto spec, table_->metadata()->PartitionSpecById(manifest.partition_spec_id)); + ICEBERG_ASSIGN_OR_RAISE(auto reader, + ManifestReader::Make(manifest, file_io_, schema_, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntries()); + for (const auto& entry : entries) { + if (entry.data_file) { + paths.push_back(entry.data_file->file_path); + } + } + } + return paths; + } + + // Live (non-deleted) delete-file paths in the current snapshot's manifests. + Result> LiveDeleteFilePaths() { + std::vector paths; + ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot()); + SnapshotCache cache(snapshot.get()); + ICEBERG_ASSIGN_OR_RAISE(auto manifests, cache.DeleteManifests(file_io_)); + for (const auto& manifest : manifests) { + ICEBERG_ASSIGN_OR_RAISE( + auto spec, table_->metadata()->PartitionSpecById(manifest.partition_spec_id)); + ICEBERG_ASSIGN_OR_RAISE(auto reader, + ManifestReader::Make(manifest, file_io_, schema_, spec)); + ICEBERG_ASSIGN_OR_RAISE(auto entries, reader->LiveEntries()); + for (const auto& entry : entries) { + if (entry.data_file) { + paths.push_back(entry.data_file->file_path); + } + } + } + return paths; + } + + Result> NewReplace() { + ICEBERG_ASSIGN_OR_RAISE(auto ctx, + TransactionContext::Make(table_, TransactionKind::kUpdate)); + return ReplacePartitions::Make(TableName(), std::move(ctx)); + } + + int64_t CommitFastAppend(const std::shared_ptr& file) { + auto fa = table_->NewFastAppend(); + EXPECT_TRUE(fa.has_value()); + fa.value()->AppendFile(file); + EXPECT_THAT(fa.value()->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + auto snap = table_->current_snapshot(); + EXPECT_TRUE(snap.has_value()); + return snap.value()->snapshot_id; + } + + void CommitEqualityDelete(const std::string& delete_path, int64_t partition_x) { + auto del_file = MakeEqualityDeleteFile(delete_path, partition_x); + ICEBERG_UNWRAP_OR_FAIL(auto row_delta, table_->NewRowDelta()); + row_delta->AddDeletes(del_file); + EXPECT_THAT(row_delta->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + } + + std::shared_ptr spec_; + std::shared_ptr schema_; + std::shared_ptr file_a_; + std::shared_ptr file_b_; +}; + +TEST_F(ReplacePartitionsTest, OperationIsOverwrite) { + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + EXPECT_EQ(op->operation(), DataOperation::kOverwrite); +} + +// Replacing a partition drops its existing file and records the summary flag. +TEST_F(ReplacePartitionsTest, PartitionedReplaceCommit) { + CommitFastAppend(file_a_); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/file_a_new.parquet", /*partition_x=*/1L)); + EXPECT_THAT(op->Commit(), IsOk()); + + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kOperation), + DataOperation::kOverwrite); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kReplacePartitions), "true"); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1"); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "1"); +} + +// Only the referenced partition is replaced; other partitions are untouched. +TEST_F(ReplacePartitionsTest, ReplaceLeavesOtherPartitions) { + CommitFastAppend(file_a_); + CommitFastAppend(file_b_); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/file_a_new.parquet", /*partition_x=*/1L)); + EXPECT_THAT(op->Commit(), IsOk()); + + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "1"); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "2"); + + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, + ::testing::UnorderedElementsAre( + file_b_->file_path, table_location_ + "/data/file_a_new.parquet")); +} + +// An unpartitioned spec triggers a table-wide replace of every existing file. +TEST_F(ReplacePartitionsTest, UnpartitionedReplacesWholeTable) { + CommitFastAppend(file_a_); + CommitFastAppend(file_b_); + + AddUnpartitionedSpec(/*spec_id=*/1); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeUnpartitionedFile("/data/all.parquet", /*spec_id=*/1)); + EXPECT_THAT(op->Commit(), IsOk()); + + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + // Both partitioned files replaced by the single unpartitioned file. + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "2"); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "1"); +} + +// No staged files means there is no partition spec to act on. +TEST_F(ReplacePartitionsTest, EmptyReplaceRejected) { + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + auto result = op->Commit(); + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(result, HasErrorMessage("Cannot determine partition specs")); +} + +// Files from two different partitioned specs cannot be replaced together. +TEST_F(ReplacePartitionsTest, MixedSpecsRejected) { + AddIdentitySpec(/*spec_id=*/1, /*field_id=*/1001, "x_v1"); + + auto file_spec0 = MakeDataFile("/data/spec0.parquet", /*partition_x=*/1L); + auto file_spec1 = MakeDataFile("/data/spec1.parquet", /*partition_x=*/1L); + file_spec1->partition_spec_id = 1; + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(file_spec0); + op->AddFile(file_spec1); + auto result = op->Commit(); + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(result, HasErrorMessage("Cannot return a single partition spec")); +} + +// Regression: an unpartitioned file staged first must not let a later file from +// another spec skip the single-spec check and commit a table-wide replace. +TEST_F(ReplacePartitionsTest, UnpartitionedFirstThenMixedRejected) { + AddUnpartitionedSpec(/*spec_id=*/1); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeUnpartitionedFile("/data/all.parquet", /*spec_id=*/1)); + op->AddFile(MakeDataFile("/data/part.parquet", /*partition_x=*/1L)); // spec 0 + auto result = op->Commit(); + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(result, HasErrorMessage("Cannot return a single partition spec")); +} + +// ValidateAppendOnly reports the partition that would be replaced. +TEST_F(ReplacePartitionsTest, AppendOnlyConflictTranslated) { + CommitFastAppend(file_a_); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/file_a_new.parquet", /*partition_x=*/1L)); + op->ValidateAppendOnly(); + auto result = op->Commit(); + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT( + result, + HasErrorMessage("Cannot commit file that conflicts with existing partition: x=1")); +} + +// ValidateAppendOnly passes when the target partition holds no existing files. +TEST_F(ReplacePartitionsTest, AppendOnlyPassesEmptyPartition) { + CommitFastAppend(file_a_); // partition 1 + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/file_b_new.parquet", /*partition_x=*/2L)); + op->ValidateAppendOnly(); + EXPECT_THAT(op->Commit(), IsOk()); +} + +// A concurrent append in the replaced partition is a conflict. +TEST_F(ReplacePartitionsTest, NoConflictingDataSamePartitionFails) { + const int64_t first_id = CommitFastAppend(file_a_); + CommitFastAppend(MakeDataFile("/data/concurrent_x1.parquet", /*partition_x=*/1L)); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/replacement_x1.parquet", /*partition_x=*/1L)); + op->ValidateFromSnapshot(first_id); + op->ValidateNoConflictingData(); + EXPECT_THAT(op->Commit(), IsError(ErrorKind::kValidationFailed)); +} + +// A concurrent append in a different partition does not conflict. +TEST_F(ReplacePartitionsTest, NoConflictingDataDifferentPartitionPasses) { + const int64_t first_id = CommitFastAppend(file_a_); + CommitFastAppend(MakeDataFile("/data/concurrent_x2.parquet", /*partition_x=*/2L)); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/replacement_x1.parquet", /*partition_x=*/1L)); + op->ValidateFromSnapshot(first_id); + op->ValidateNoConflictingData(); + EXPECT_THAT(op->Commit(), IsOk()); +} + +// A concurrent delete file in the replaced partition is a conflict. +TEST_F(ReplacePartitionsTest, NoConflictingDeletesFails) { + const int64_t first_id = CommitFastAppend(file_a_); + CommitEqualityDelete("/delete/concurrent_x1.parquet", /*partition_x=*/1L); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/replacement_x1.parquet", /*partition_x=*/1L)); + op->ValidateFromSnapshot(first_id); + op->ValidateNoConflictingDeletes(); + EXPECT_THAT(op->Commit(), IsError(ErrorKind::kValidationFailed)); +} + +// A concurrent delete file in another partition does not conflict. +TEST_F(ReplacePartitionsTest, NoConflictingDeletesDifferentPartitionPasses) { + const int64_t first_id = CommitFastAppend(file_a_); + CommitEqualityDelete("/delete/concurrent_x2.parquet", /*partition_x=*/2L); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/replacement_x1.parquet", /*partition_x=*/1L)); + op->ValidateFromSnapshot(first_id); + op->ValidateNoConflictingDeletes(); + EXPECT_THAT(op->Commit(), IsOk()); +} + +// ValidateNoConflictingDeletes also rejects a concurrent removal of a data file +// in the replaced partition, not only a newly added delete file. +TEST_F(ReplacePartitionsTest, NoConflictingDeletesConcurrentDataRemovalFails) { + CommitFastAppend(MakeDataFile("/data/existing_x1.parquet", /*partition_x=*/1L)); + const int64_t base_id = CommitFastAppend(file_a_); // partition 1 + CommitDeleteDataFile(table_location_ + "/data/existing_x1.parquet"); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/replacement_x1.parquet", /*partition_x=*/1L)); + op->ValidateFromSnapshot(base_id); + op->ValidateNoConflictingDeletes(); + EXPECT_THAT(op->Commit(), IsError(ErrorKind::kValidationFailed)); +} + +TEST_F(ReplacePartitionsTest, NoValidationSamePartitionConcurrentReplaceCommits) { + const int64_t first_id = CommitFastAppend(file_a_); + + ICEBERG_UNWRAP_OR_FAIL(auto concurrent, NewReplace()); + concurrent->AddFile(MakeDataFile("/data/concurrent_x1.parquet", /*partition_x=*/1L)); + concurrent->ValidateFromSnapshot(first_id); + EXPECT_THAT(concurrent->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/replacement_x1.parquet", /*partition_x=*/1L)); + op->ValidateFromSnapshot(first_id); + EXPECT_THAT(op->Commit(), IsOk()); + + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto live, LiveDataFilePaths()); + EXPECT_THAT(live, ::testing::UnorderedElementsAre(table_location_ + + "/data/replacement_x1.parquet")); +} + +// An all-void spec (a partition field using the void transform) is unpartitioned, +// so a file staged against it replaces the whole table like an empty spec does. +TEST_F(ReplacePartitionsTest, AllVoidSpecReplacesWholeTable) { + CommitFastAppend(file_a_); + CommitFastAppend(file_b_); + + AddAllVoidSpec(/*spec_id=*/1, /*field_id=*/1001, "x_void"); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeAllVoidFile("/data/all_void.parquet", /*spec_id=*/1)); + EXPECT_THAT(op->Commit(), IsOk()); + + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto snapshot, table_->current_snapshot()); + // The single all-void file is added and both partitioned files are replaced, + // leaving it as the only live file. + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1"); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "2"); + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "1"); +} + +// On the unpartitioned (row-filter) path, conflict validation covers the whole +// table: a concurrent append in any partition is a conflict. +TEST_F(ReplacePartitionsTest, UnpartitionedNoConflictingDataFails) { + const int64_t first_id = CommitFastAppend(file_a_); + CommitFastAppend(MakeDataFile("/data/concurrent_x2.parquet", /*partition_x=*/2L)); + + AddUnpartitionedSpec(/*spec_id=*/1); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeUnpartitionedFile("/data/all.parquet", /*spec_id=*/1)); + op->ValidateFromSnapshot(first_id); + op->ValidateNoConflictingData(); + EXPECT_THAT(op->Commit(), IsError(ErrorKind::kValidationFailed)); +} + +TEST_F(ReplacePartitionsTest, DeleteContentAddFileRejected) { + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + auto delete_file = + MakeEqualityDeleteFile("/delete/invalid.parquet", /*partition_x=*/1L); + op->AddFile(delete_file); + + auto result = op->Commit(); + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT(result, + HasErrorMessage("Invalid data file to add: " + delete_file->file_path + + " has delete-file content")); +} + +TEST_F(ReplacePartitionsTest, NullAddFileRejected) { + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(nullptr); + auto result = op->Commit(); + EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed)); + EXPECT_THAT(result, HasErrorMessage("Invalid data file: null")); +} + +// Format v3: replacing a partition also drops deletion vectors that reference +// data files in that partition. +class ReplacePartitionsV3Test : public ReplacePartitionsTest { + protected: + std::string MetadataResource() const override { + return "TableMetadataV3ValidMinimal.json"; + } +}; + +TEST_F(ReplacePartitionsV3Test, ReplaceCleansUpDeletionVectors) { + ICEBERG_UNWRAP_OR_FAIL(auto props, table_->NewUpdateProperties()); + props->Set(std::string(TableProperties::kManifestMinMergeCount.key()), "1"); + EXPECT_THAT(props->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + + CommitFastAppend(file_a_); + CommitFastAppend(file_b_); + + auto dv_a = MakeDeletionVector("/delete/dv_a.puffin", file_a_->file_path, + /*partition_x=*/1L); + auto dv_b = MakeDeletionVector("/delete/dv_b.puffin", file_b_->file_path, + /*partition_x=*/2L); + ICEBERG_UNWRAP_OR_FAIL(auto row_delta, table_->NewRowDelta()); + row_delta->AddDeletes(dv_a); + row_delta->AddDeletes(dv_b); + EXPECT_THAT(row_delta->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto before, LiveDeleteFilePaths()); + EXPECT_THAT(before, ::testing::UnorderedElementsAre(dv_a->file_path, dv_b->file_path)); + + ICEBERG_UNWRAP_OR_FAIL(auto op, NewReplace()); + op->AddFile(MakeDataFile("/data/file_a_new.parquet", /*partition_x=*/1L)); + EXPECT_THAT(op->Commit(), IsOk()); + EXPECT_THAT(table_->Refresh(), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(auto after, LiveDeleteFilePaths()); + EXPECT_THAT(after, ::testing::ElementsAre(dv_b->file_path)); + + ICEBERG_UNWRAP_OR_FAIL(auto live_data, LiveDataFilePaths()); + EXPECT_THAT(live_data, + ::testing::UnorderedElementsAre( + file_b_->file_path, table_location_ + "/data/file_a_new.parquet")); +} + +} // namespace iceberg diff --git a/src/iceberg/type_fwd.h b/src/iceberg/type_fwd.h index 91e18b24c..f437e3db7 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -246,6 +246,7 @@ class FastAppend; class MergeAppend; class OverwriteFiles; class PendingUpdate; +class ReplacePartitions; class RewriteFiles; class RowDelta; class SetSnapshot; diff --git a/src/iceberg/update/merging_snapshot_update.cc b/src/iceberg/update/merging_snapshot_update.cc index 09e495cf1..558480759 100644 --- a/src/iceberg/update/merging_snapshot_update.cc +++ b/src/iceberg/update/merging_snapshot_update.cc @@ -463,6 +463,10 @@ Status MergingSnapshotUpdate::AddDataFile(std::shared_ptr file) { if (!file) { return InvalidArgument("Cannot add a null data file"); } + if (file->content != DataFile::Content::kData) { + return InvalidArgument("Invalid data file to add: {} has delete-file content", + file->file_path); + } if (!file->partition_spec_id.has_value()) { return InvalidArgument("Data file must have a partition spec ID"); } diff --git a/src/iceberg/update/meson.build b/src/iceberg/update/meson.build index 83c64f363..6b2cf6f33 100644 --- a/src/iceberg/update/meson.build +++ b/src/iceberg/update/meson.build @@ -24,6 +24,7 @@ install_headers( 'merging_snapshot_update.h', 'overwrite_files.h', 'pending_update.h', + 'replace_partitions.h', 'rewrite_files.h', 'row_delta.h', 'set_snapshot.h', diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc new file mode 100644 index 000000000..3d224375c --- /dev/null +++ b/src/iceberg/update/replace_partitions.cc @@ -0,0 +1,146 @@ +/* + * 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/update/replace_partitions.h" + +#include + +#include "iceberg/expression/expressions.h" +#include "iceberg/partition_spec.h" +#include "iceberg/snapshot.h" +#include "iceberg/table.h" // IWYU pragma: keep +#include "iceberg/table_metadata.h" +#include "iceberg/transaction.h" +#include "iceberg/util/error_collector.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +Result> ReplacePartitions::Make( + std::string table_name, std::shared_ptr ctx) { + ICEBERG_PRECHECK(!table_name.empty(), "Table name cannot be empty"); + ICEBERG_PRECHECK(ctx != nullptr, "Cannot create ReplacePartitions without a context"); + return std::unique_ptr( + new ReplacePartitions(std::move(table_name), std::move(ctx))); +} + +ReplacePartitions::ReplacePartitions(std::string table_name, + std::shared_ptr ctx) + : MergingSnapshotUpdate(std::move(table_name), std::move(ctx)) { + Set(SnapshotSummaryFields::kReplacePartitions, "true"); +} + +ReplacePartitions& ReplacePartitions::AddFile(const std::shared_ptr& file) { + ICEBERG_BUILDER_CHECK(file != nullptr, "Invalid data file: null"); + ICEBERG_BUILDER_CHECK(file->partition_spec_id.has_value(), + "Data file must have partition spec ID"); + + int32_t spec_id = file->partition_spec_id.value(); + ICEBERG_BUILDER_RETURN_IF_ERROR(DropPartition(spec_id, file->partition)); + replaced_partitions_.add(spec_id, file->partition); + ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(file)); + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateAppendOnly() { + FailAnyDelete(); + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateFromSnapshot(int64_t snapshot_id) { + starting_snapshot_id_ = snapshot_id; + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateNoConflictingData() { + validate_conflicting_data_ = true; + return *this; +} + +ReplacePartitions& ReplacePartitions::ValidateNoConflictingDeletes() { + validate_conflicting_deletes_ = true; + return *this; +} + +std::string ReplacePartitions::operation() { return DataOperation::kOverwrite; } + +Result> ReplacePartitions::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr& snapshot) { + ICEBERG_ASSIGN_OR_RAISE(auto data_spec, DataSpec()); + if (data_spec->IsUnpartitioned()) { + ICEBERG_RETURN_UNEXPECTED(DeleteByRowFilter(Expressions::AlwaysTrue())); + } + + auto result = MergingSnapshotUpdate::Apply(metadata_to_update, snapshot); + if (result.has_value()) { + return result; + } + // Translate append-only conflicts to the ReplacePartitions error. + constexpr std::string_view kFailAnyDeletePrefix = + "Operation would delete existing data: "; + const Error& error = result.error(); + if (error.kind == ErrorKind::kValidationFailed && + error.message.starts_with(kFailAnyDeletePrefix)) { + std::string_view partition_path{error.message}; + partition_path.remove_prefix(kFailAnyDeletePrefix.size()); + return ValidationFailed( + "Cannot commit file that conflicts with existing partition: {}", partition_path); + } + return result; +} + +Status ReplacePartitions::Validate(const TableMetadata& current_metadata, + const std::shared_ptr& snapshot) { + ICEBERG_ASSIGN_OR_RAISE(auto data_spec, DataSpec()); + + if (snapshot == nullptr) { + return {}; + } + + auto io = ctx_->table->io(); + if (validate_conflicting_data_) { + if (data_spec->IsUnpartitioned()) { + ICEBERG_RETURN_UNEXPECTED(ValidateAddedDataFiles( + current_metadata, starting_snapshot_id_, Expressions::AlwaysTrue(), snapshot, + io, IsCaseSensitive())); + } else { + ICEBERG_RETURN_UNEXPECTED(ValidateAddedDataFiles( + current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + } + } + if (validate_conflicting_deletes_) { + // Check deleted data files before newly added delete files. + if (data_spec->IsUnpartitioned()) { + ICEBERG_RETURN_UNEXPECTED(ValidateDeletedDataFiles( + current_metadata, starting_snapshot_id_, Expressions::AlwaysTrue(), snapshot, + io, IsCaseSensitive())); + ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( + current_metadata, starting_snapshot_id_, Expressions::AlwaysTrue(), snapshot, + io, IsCaseSensitive())); + } else { + ICEBERG_RETURN_UNEXPECTED(ValidateDeletedDataFiles( + current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( + current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + } + } + return {}; +} + +} // namespace iceberg diff --git a/src/iceberg/update/replace_partitions.h b/src/iceberg/update/replace_partitions.h new file mode 100644 index 000000000..40048ba5c --- /dev/null +++ b/src/iceberg/update/replace_partitions.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/update/replace_partitions.h + +#include +#include +#include +#include +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" +#include "iceberg/update/merging_snapshot_update.h" +#include "iceberg/util/partition_value_util.h" + +namespace iceberg { + +/// \brief API for overwriting files in a table by partition. +/// +/// \note This is provided to implement SQL compatible with Hive table +/// operations but is not recommended. Instead, use OverwriteFiles to +/// explicitly overwrite data. +/// +/// The default validation mode is idempotent, meaning the overwrite is correct +/// and should be committed regardless of other concurrent changes to the table. +/// Alternatively, call ValidateNoConflictingDeletes() and +/// ValidateNoConflictingData() to ensure that no conflicting delete files or +/// data files, respectively, have been written since the snapshot passed to +/// ValidateFromSnapshot(). +/// +/// This API accumulates file additions and produces a new Snapshot by replacing +/// all files in partitions containing new data with the new additions. This +/// operation implements dynamic partition replacement. +/// +/// When committing, these changes are applied to the latest table snapshot. +/// Commit conflicts are resolved by applying the changes to the new latest +/// snapshot and reattempting the commit. +class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { + public: + /// \brief Create a new ReplacePartitions instance. + /// + /// \param table_name The name of the table + /// \param ctx The transaction context + /// \return A Result containing the ReplacePartitions instance or an error + static Result> Make( + std::string table_name, std::shared_ptr ctx); + + /// \brief Add a data file to the table. + /// + /// \param file A data file with partition_spec_id set + /// \return Reference to this for method chaining + ReplacePartitions& AddFile(const std::shared_ptr& file); + + /// \brief Validate that no partitions will be replaced and the operation is + /// append-only. + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateAppendOnly(); + + /// \brief Set the snapshot ID used in validations for this operation. + /// + /// All validations check changes after this snapshot ID. If this is not + /// called, validation occurs from the beginning of the table's history. + /// + /// This method should be called before this operation is committed. If a + /// concurrent operation committed a data or delete file or removed a data + /// file after the given snapshot ID that might contain rows matching a + /// partition marked for deletion, validation detects the conflict and fails. + /// + /// \param snapshot_id A snapshot ID set to when this operation started to + /// read the table + /// \return Reference to this for method chaining + ReplacePartitions& ValidateFromSnapshot(int64_t snapshot_id); + + /// \brief Enable validation that data added concurrently does not conflict + /// with this operation. + /// + /// Validating concurrent data files is required during non-idempotent replace + /// partition operations. This checks whether a concurrent operation inserted + /// data in any partition being overwritten and aborts the replace to avoid + /// removing rows added concurrently. + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateNoConflictingData(); + + /// \brief Enable validation that deletes that happened concurrently do not + /// conflict with this operation. + /// + /// Validating concurrent deletes is required during non-idempotent replace + /// partition operations. This checks whether a concurrent operation deleted + /// data in any partition being overwritten and aborts the replace to avoid + /// undeleting rows removed concurrently. + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateNoConflictingDeletes(); + + std::string operation() override; + + protected: + Status Validate(const TableMetadata& current_metadata, + const std::shared_ptr& snapshot) override; + + /// \brief Apply changes, translating the fail-any-delete error. + /// + /// Rewrites the delete error raised when ValidateAppendOnly() is set into the + /// partition-conflict message. + Result> Apply( + const TableMetadata& current_metadata, + const std::shared_ptr& snapshot) override; + + private: + explicit ReplacePartitions(std::string table_name, + std::shared_ptr ctx); + + std::optional starting_snapshot_id_; + bool validate_conflicting_data_{false}; + bool validate_conflicting_deletes_{false}; + PartitionSet replaced_partitions_; +}; + +} // namespace iceberg