From 12dc230a6a387bf1b6020e4958a808875014fa36 Mon Sep 17 00:00:00 2001 From: Xinli shang Date: Tue, 23 Jun 2026 17:05:21 -0700 Subject: [PATCH 1/6] feat: add ReplacePartitions core class (PR1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds iceberg::ReplacePartitions — the dynamic partition overwrite operation. Each AddFile() registers the file's partition for replacement of all existing data and delete files in that partition. Produces an overwrite snapshot with "replace-partitions=true" in the summary. Unpartitioned tables replace all existing data files. Extends MergingSnapshotUpdate (matching Java's BaseReplacePartitions) so the full data+delete manifest pipeline, custom-summary-property handling, and conflict-validation helpers are inherited. AddFile() unconditionally calls DropPartition(spec_id, file->partition) — for unpartitioned specs the partition value is empty and the filter manager matches every file in that spec, so no separate AlwaysTrue path is needed. Touched partitions are tracked in a PartitionSet; Validate() uses the partition-scoped overloads of ValidateAddedDataFiles / ValidateNoNewDeleteFiles, or skips entirely when no partitions were staged. Changes: * iceberg::ReplacePartitions extending MergingSnapshotUpdate with builder API (AddFile, ValidateAppendOnly, ValidateFromSnapshot, ValidateNoConflictingData, ValidateNoConflictingDeletes) and Validate() override. * SnapshotSummaryFields::kReplacePartitions = "replace-partitions". * MergingSnapshotUpdate::SetSummaryProperty promoted from private to protected so subclasses can stash custom summary entries that survive commit retry via the cached-rebuild path. * Forward declaration in type_fwd.h. * CMake + Meson source registration. Public API wiring (Table::NewReplacePartitions(), Transaction::NewReplacePartitions()) and end-to-end tests are deferred to PR2. Tracking: #775 Related: #637 --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/meson.build | 1 + src/iceberg/snapshot.h | 2 + src/iceberg/type_fwd.h | 1 + src/iceberg/update/merging_snapshot_update.h | 10 +- src/iceberg/update/replace_partitions.cc | 112 +++++++++++++++++ src/iceberg/update/replace_partitions.h | 126 +++++++++++++++++++ 7 files changed, 251 insertions(+), 2 deletions(-) create mode 100644 src/iceberg/update/replace_partitions.cc create mode 100644 src/iceberg/update/replace_partitions.h diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 9a0dc68b7..39a4b66c1 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -104,6 +104,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/pending_update.cc update/row_delta.cc diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build index ab514be87..47aa41074 100644 --- a/src/iceberg/meson.build +++ b/src/iceberg/meson.build @@ -131,6 +131,7 @@ iceberg_sources = files( 'update/merge_append.cc', 'update/merging_snapshot_update.cc', 'update/pending_update.cc', + 'update/replace_partitions.cc', 'update/row_delta.cc', 'update/set_snapshot.cc', 'update/snapshot_manager.cc', diff --git a/src/iceberg/snapshot.h b/src/iceberg/snapshot.h index f3e7ffb85..73768f2f6 100644 --- a/src/iceberg/snapshot.h +++ b/src/iceberg/snapshot.h @@ -251,6 +251,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/type_fwd.h b/src/iceberg/type_fwd.h index f29bc4a1a..0fcb42dd4 100644 --- a/src/iceberg/type_fwd.h +++ b/src/iceberg/type_fwd.h @@ -241,6 +241,7 @@ class TransactionContext; class DeleteFiles; class ExpireSnapshots; class FastAppend; +class ReplacePartitions; class MergeAppend; class PendingUpdate; class RowDelta; diff --git a/src/iceberg/update/merging_snapshot_update.h b/src/iceberg/update/merging_snapshot_update.h index fc3987ee1..0f447c5e5 100644 --- a/src/iceberg/update/merging_snapshot_update.h +++ b/src/iceberg/update/merging_snapshot_update.h @@ -296,6 +296,14 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { const std::shared_ptr& parent, std::shared_ptr io) const; + /// \brief Record a caller-supplied summary entry that survives commit retry. + /// + /// MergingSnapshotUpdate clears summary_ at the start of every Apply() and + /// rebuilds it from the cached sub-builders, so subclasses must route any + /// custom property through this hook rather than calling summary_.Set() + /// directly. Stored entries are re-merged in Summary(). + void SetSummaryProperty(const std::string& property, const std::string& value) override; + private: struct PendingDeleteFile { std::shared_ptr file; @@ -334,8 +342,6 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate { Status ManagersReady() const; - void SetSummaryProperty(const std::string& property, const std::string& value) override; - Result> MergeDVs() const; /// \brief Write new data manifests for staged data files; caches the result. diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc new file mode 100644 index 000000000..cc627c7b7 --- /dev/null +++ b/src/iceberg/update/replace_partitions.cc @@ -0,0 +1,112 @@ +/* + * 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 "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)) { + SetSummaryProperty(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_ASSIGN_OR_RETURN(auto spec, base().PartitionSpecById(spec_id)); + + ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(file)); + // DropPartition(spec_id, partition) registers the (spec_id, partition_values) + // tuple with both data and delete filter managers. For an unpartitioned spec + // the partition values are empty and naturally match every file under that + // spec — no separate AlwaysTrue path is needed, and validation stays scoped + // to the spec rather than the whole table. + ICEBERG_BUILDER_RETURN_IF_ERROR(DropPartition(spec_id, file->partition)); + replaced_partitions_.add(spec_id, file->partition); + 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; } + +Status ReplacePartitions::Validate(const TableMetadata& current_metadata, + const std::shared_ptr& snapshot) { + if (snapshot == nullptr) { + return {}; + } + // No-op update: no partitions were staged, so there is nothing to conflict + // with. Calling the validators with AlwaysTrue here would turn an empty + // builder into a full-table conflict check. + if (replaced_partitions_.empty()) { + return {}; + } + + auto io = ctx_->table->io(); + if (validate_conflicting_data_) { + ICEBERG_RETURN_UNEXPECTED(ValidateAddedDataFiles( + current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + } + if (validate_conflicting_deletes_) { + 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..33689722a --- /dev/null +++ b/src/iceberg/update/replace_partitions.h @@ -0,0 +1,126 @@ +/* + * 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 "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 Replaces partitions in a table with new data files. +/// +/// ReplacePartitions dynamically identifies which partitions to overwrite based +/// on the data files added via AddFile(). All existing data files in each +/// touched partition are marked DELETED, and the new files are written as the +/// sole data in those partitions. Partitions not referenced by any added file +/// are left unchanged. +/// +/// This operation produces a snapshot with operation="overwrite" and +/// "replace-partitions"="true" in the summary. For unpartitioned tables, all +/// existing files are replaced. +/// +/// When committing, these changes are applied to the latest table snapshot. +/// Commit conflicts are resolved by re-applying 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 and mark its partition for replacement. + /// + /// Each call registers the file's partition so all existing data files in + /// that partition are replaced. Duplicate files (same path) are ignored. + /// + /// \param file The data file to add (must have partition_spec_id set) + /// \return Reference to this for method chaining + ReplacePartitions& AddFile(const std::shared_ptr& file); + + /// \brief Fail the commit if any existing data file would be deleted. + /// + /// This validation is useful to ensure the operation is only applied to + /// tables where no data currently exists in the affected partitions. + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateAppendOnly(); + + /// \brief Set the snapshot ID used as the baseline for conflict validation. + /// + /// Validations check changes that occurred after this snapshot ID. If not + /// set, all ancestor snapshots through the initial snapshot are validated. + /// + /// \param snapshot_id A snapshot ID + /// \return Reference to this for method chaining + ReplacePartitions& ValidateFromSnapshot(int64_t snapshot_id); + + /// \brief Enable validation that no conflicting data files were added concurrently. + /// + /// Fails the commit if a concurrent operation added a data file in any of + /// the partitions being replaced after the snapshot set by + /// ValidateFromSnapshot(). + /// + /// \return Reference to this for method chaining + ReplacePartitions& ValidateNoConflictingData(); + + /// \brief Enable validation that no conflicting delete files were added concurrently. + /// + /// Fails the commit if a concurrent operation added a delete file covering + /// any of the partitions being replaced after the snapshot set by + /// ValidateFromSnapshot(). + /// + /// \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; + + 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}; + // Partitions touched by AddFile(); used to scope conflict validation to the + // overwritten partitions instead of the whole table. For unpartitioned specs + // the partition values are empty, and DropPartition(spec_id, {}) already + // matches every file in that spec — no separate "whole table" path is needed. + PartitionSet replaced_partitions_; +}; + +} // namespace iceberg From 90ae61d8ec8bb349e4e51b0511d912cd6fde2005 Mon Sep 17 00:00:00 2001 From: Xinli Shang Date: Wed, 24 Jun 2026 11:53:15 -0700 Subject: [PATCH 2/6] address review: table-wide unpartitioned replace + ValidateDeletedDataFiles - AddFile() now uses DeleteByRowFilter(AlwaysTrue()) for unpartitioned specs instead of DropPartition with empty partition values, matching Java BaseReplacePartitions which treats unpartitioned tables as a table-wide replace. - Validate() now also calls ValidateDeletedDataFiles when ValidateNoConflictingDeletes is enabled, mirroring Java where validateNewDeletes gates both checks. This rejects concurrent overwrite/delete commits in the replaced partitions. - New replace_by_row_filter_ flag drives the AlwaysTrue path in Validate() for the unpartitioned case. --- src/iceberg/update/replace_partitions.cc | 56 +++++++++++++++++------- src/iceberg/update/replace_partitions.h | 12 +++-- 2 files changed, 49 insertions(+), 19 deletions(-) diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc index cc627c7b7..f0ea49e1c 100644 --- a/src/iceberg/update/replace_partitions.cc +++ b/src/iceberg/update/replace_partitions.cc @@ -53,13 +53,17 @@ ReplacePartitions& ReplacePartitions::AddFile(const std::shared_ptr& f ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, base().PartitionSpecById(spec_id)); ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(file)); - // DropPartition(spec_id, partition) registers the (spec_id, partition_values) - // tuple with both data and delete filter managers. For an unpartitioned spec - // the partition values are empty and naturally match every file under that - // spec — no separate AlwaysTrue path is needed, and validation stays scoped - // to the spec rather than the whole table. - ICEBERG_BUILDER_RETURN_IF_ERROR(DropPartition(spec_id, file->partition)); - replaced_partitions_.add(spec_id, file->partition); + if (spec->fields().empty()) { + // Unpartitioned spec: Java's BaseReplacePartitions treats this as a + // table-wide replace rather than a spec-scoped DropPartition with empty + // partition values. Mirror that so every existing data file is dropped + // and conflict validation runs against AlwaysTrue. + ICEBERG_BUILDER_RETURN_IF_ERROR(DeleteByRowFilter(Expressions::AlwaysTrue())); + replace_by_row_filter_ = true; + } else { + ICEBERG_BUILDER_RETURN_IF_ERROR(DropPartition(spec_id, file->partition)); + replaced_partitions_.add(spec_id, file->partition); + } return *this; } @@ -90,21 +94,43 @@ Status ReplacePartitions::Validate(const TableMetadata& current_metadata, if (snapshot == nullptr) { return {}; } - // No-op update: no partitions were staged, so there is nothing to conflict - // with. Calling the validators with AlwaysTrue here would turn an empty - // builder into a full-table conflict check. - if (replaced_partitions_.empty()) { + // No-op update: no partitions were staged and no table-wide replace was + // requested, so there is nothing to conflict with. Calling the validators + // with AlwaysTrue here would turn an empty builder into a full-table check. + if (!replace_by_row_filter_ && replaced_partitions_.empty()) { return {}; } auto io = ctx_->table->io(); if (validate_conflicting_data_) { - ICEBERG_RETURN_UNEXPECTED(ValidateAddedDataFiles( - current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + if (replace_by_row_filter_) { + 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_) { - ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( - current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + // Java's BaseReplacePartitions.validate gates both ValidateNoNewDeleteFiles + // and ValidateDeletedDataFiles on the same validateNewDeletes flag. The + // second check rejects concurrent overwrite/delete commits in the replaced + // partitions; without it a concurrent delete in a replaced partition would + // commit silently. + if (replace_by_row_filter_) { + ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( + current_metadata, starting_snapshot_id_, Expressions::AlwaysTrue(), snapshot, + io, IsCaseSensitive())); + ICEBERG_RETURN_UNEXPECTED(ValidateDeletedDataFiles( + current_metadata, starting_snapshot_id_, Expressions::AlwaysTrue(), snapshot, + io, IsCaseSensitive())); + } else { + ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( + current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + ICEBERG_RETURN_UNEXPECTED(ValidateDeletedDataFiles( + current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); + } } return {}; } diff --git a/src/iceberg/update/replace_partitions.h b/src/iceberg/update/replace_partitions.h index 33689722a..27c69dee9 100644 --- a/src/iceberg/update/replace_partitions.h +++ b/src/iceberg/update/replace_partitions.h @@ -116,10 +116,14 @@ class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { std::optional starting_snapshot_id_; bool validate_conflicting_data_{false}; bool validate_conflicting_deletes_{false}; - // Partitions touched by AddFile(); used to scope conflict validation to the - // overwritten partitions instead of the whole table. For unpartitioned specs - // the partition values are empty, and DropPartition(spec_id, {}) already - // matches every file in that spec — no separate "whole table" path is needed. + // True once an AddFile() call has staged a file whose partition spec is + // unpartitioned. Java's BaseReplacePartitions treats this case as a + // table-wide replace (DeleteByRowFilter(AlwaysTrue())) and runs conflict + // validation against AlwaysTrue rather than a partition set — mirror that. + bool replace_by_row_filter_{false}; + // Partitions touched by AddFile() in partitioned specs. Used to scope + // conflict validation to the overwritten partitions in the partitioned + // case; ignored when replace_by_row_filter_ is true. PartitionSet replaced_partitions_; }; From c2a4a5f71309653d70d66a44feec65795e287c21 Mon Sep 17 00:00:00 2001 From: Xinli shang Date: Thu, 25 Jun 2026 09:36:56 -0700 Subject: [PATCH 3/6] address review: meson, Hive-compat note, reject no-data-files in Validate - Add replace_partitions.h to src/iceberg/update/meson.build (manuzhang) - Add Java-style note recommending OverwriteFiles for non-Hive use (zhjwpku) - Validate(): call DataSpec() to reject no-data-files replace, matching Java BaseReplacePartitions which throws when no file was added (manuzhang) --- src/iceberg/update/meson.build | 1 + src/iceberg/update/replace_partitions.cc | 5 +++++ src/iceberg/update/replace_partitions.h | 4 ++++ 3 files changed, 10 insertions(+) diff --git a/src/iceberg/update/meson.build b/src/iceberg/update/meson.build index 4f594a06e..e00def71b 100644 --- a/src/iceberg/update/meson.build +++ b/src/iceberg/update/meson.build @@ -23,6 +23,7 @@ install_headers( 'merge_append.h', 'merging_snapshot_update.h', 'pending_update.h', + 'replace_partitions.h', 'row_delta.h', 'set_snapshot.h', 'snapshot_manager.h', diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc index f0ea49e1c..3744ae179 100644 --- a/src/iceberg/update/replace_partitions.cc +++ b/src/iceberg/update/replace_partitions.cc @@ -91,6 +91,11 @@ std::string ReplacePartitions::operation() { return DataOperation::kOverwrite; } Status ReplacePartitions::Validate(const TableMetadata& current_metadata, const std::shared_ptr& snapshot) { + // Java's BaseReplacePartitions requires at least one staged data file: + // dataSpec() throws if none was added. Call DataSpec() up front to surface + // the same error here, instead of silently committing an empty snapshot. + ICEBERG_ASSIGN_OR_RAISE(auto data_spec, DataSpec()); + if (snapshot == nullptr) { return {}; } diff --git a/src/iceberg/update/replace_partitions.h b/src/iceberg/update/replace_partitions.h index 27c69dee9..70515054b 100644 --- a/src/iceberg/update/replace_partitions.h +++ b/src/iceberg/update/replace_partitions.h @@ -49,6 +49,10 @@ namespace iceberg { /// When committing, these changes are applied to the latest table snapshot. /// Commit conflicts are resolved by re-applying to the new latest snapshot /// and reattempting the commit. +/// +/// \note This is provided to implement SQL compatible with Hive table +/// operations but is not recommended. Instead, use OverwriteFiles to +/// explicitly overwrite data. class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { public: /// \brief Create a new ReplacePartitions instance. From d000927fdef29c70355fde337eedf6eca232c381 Mon Sep 17 00:00:00 2001 From: Xinli shang Date: Thu, 25 Jun 2026 09:45:24 -0700 Subject: [PATCH 4/6] fix: use AddFile flags to gate empty-replace check, not DataSpec() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DataSpec() also errors on multi-spec stages, which would silently break a ReplacePartitions that touches files from more than one spec. Use the flags AddFile() sets (replace_by_row_filter_ / replaced_partitions_) instead — the same condition that already gated the no-op early-return below, now promoted to an error to match Java BaseReplacePartitions. --- src/iceberg/update/replace_partitions.cc | 17 +++++++---------- 1 file changed, 7 insertions(+), 10 deletions(-) diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc index 3744ae179..d47b21a92 100644 --- a/src/iceberg/update/replace_partitions.cc +++ b/src/iceberg/update/replace_partitions.cc @@ -91,20 +91,17 @@ std::string ReplacePartitions::operation() { return DataOperation::kOverwrite; } Status ReplacePartitions::Validate(const TableMetadata& current_metadata, const std::shared_ptr& snapshot) { - // Java's BaseReplacePartitions requires at least one staged data file: - // dataSpec() throws if none was added. Call DataSpec() up front to surface - // the same error here, instead of silently committing an empty snapshot. - ICEBERG_ASSIGN_OR_RAISE(auto data_spec, DataSpec()); + // Match Java BaseReplacePartitions: require at least one staged data file. + // Use the flags AddFile() sets — `DataSpec()` would also error on multi-spec + // stages and is not the guard we want here. + if (!replace_by_row_filter_ && replaced_partitions_.empty()) { + return InvalidArgument( + "ReplacePartitions requires at least one data file; call AddFile() first"); + } if (snapshot == nullptr) { return {}; } - // No-op update: no partitions were staged and no table-wide replace was - // requested, so there is nothing to conflict with. Calling the validators - // with AlwaysTrue here would turn an empty builder into a full-table check. - if (!replace_by_row_filter_ && replaced_partitions_.empty()) { - return {}; - } auto io = ctx_->table->io(); if (validate_conflicting_data_) { From b03d7909cfbb44ab06480802c706c59c6a857949 Mon Sep 17 00:00:00 2001 From: Xinli shang Date: Sat, 11 Jul 2026 13:22:19 -0700 Subject: [PATCH 5/6] Address review: void-transform unpartitioned check and DataSpec single-spec enforcement --- src/iceberg/update/replace_partitions.cc | 27 ++++++++++++++++++------ 1 file changed, 20 insertions(+), 7 deletions(-) diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc index d47b21a92..1a30a74c8 100644 --- a/src/iceberg/update/replace_partitions.cc +++ b/src/iceberg/update/replace_partitions.cc @@ -19,12 +19,15 @@ #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/transform.h" #include "iceberg/util/error_collector.h" #include "iceberg/util/macros.h" @@ -53,7 +56,17 @@ ReplacePartitions& ReplacePartitions::AddFile(const std::shared_ptr& f ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, base().PartitionSpecById(spec_id)); ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(file)); - if (spec->fields().empty()) { + // A spec is effectively unpartitioned if it has no fields, or every field + // uses the void transform. Java's PartitionSpec.isUnpartitioned() covers + // both cases; mirror that so an all-void spec triggers the table-wide + // replace path instead of a DropPartition with void-valued partition keys. + auto is_unpartitioned = [](const PartitionSpec& s) { + return s.fields().empty() || + std::all_of(s.fields().begin(), s.fields().end(), [](const PartitionField& f) { + return f.transform()->transform_type() == TransformType::kVoid; + }); + }; + if (is_unpartitioned(*spec)) { // Unpartitioned spec: Java's BaseReplacePartitions treats this as a // table-wide replace rather than a spec-scoped DropPartition with empty // partition values. Mirror that so every existing data file is dropped @@ -91,12 +104,12 @@ std::string ReplacePartitions::operation() { return DataOperation::kOverwrite; } Status ReplacePartitions::Validate(const TableMetadata& current_metadata, const std::shared_ptr& snapshot) { - // Match Java BaseReplacePartitions: require at least one staged data file. - // Use the flags AddFile() sets — `DataSpec()` would also error on multi-spec - // stages and is not the guard we want here. - if (!replace_by_row_filter_ && replaced_partitions_.empty()) { - return InvalidArgument( - "ReplacePartitions requires at least one data file; call AddFile() first"); + // Match Java BaseReplacePartitions.validate: require at least one staged data + // file and that all staged files share exactly one partition spec. + // DataSpec() enforces both invariants; ignore the returned spec here since + // replace_by_row_filter_ / replaced_partitions_ already scope validation. + if (!replace_by_row_filter_) { + ICEBERG_RETURN_UNEXPECTED(DataSpec()); } if (snapshot == nullptr) { From 638d2d4e401a1d4fe9e982283ed745b3d1f5da40 Mon Sep 17 00:00:00 2001 From: Xinli shang Date: Thu, 23 Jul 2026 13:40:19 -0700 Subject: [PATCH 6/6] Address wgtmac review: unconditional DataSpec, error translation, tests - Call DataSpec() unconditionally in Validate so a replace mixing an unpartitioned/all-void file with files from another spec is rejected instead of committing a table-wide replace. - Override Apply() to translate the fail-any-delete error into Java's "Cannot commit file that conflicts with existing partition" wording. - Run ValidateDeletedDataFiles before ValidateNoNewDeleteFiles in both branches to match Java's deterministic failure ordering. - Trim Java-comparison/implementation-history comments to focus on the rule; align the ValidateNoConflictingDeletes contract with Java. - Add direct ReplacePartitions tests: partitioned and unpartitioned replacement, empty and mixed specs, append-only validation, and same/different-partition data and delete concurrency. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/iceberg/test/CMakeLists.txt | 3 +- src/iceberg/test/replace_partitions_test.cc | 328 ++++++++++++++++++++ src/iceberg/update/replace_partitions.cc | 68 ++-- src/iceberg/update/replace_partitions.h | 31 +- 4 files changed, 394 insertions(+), 36 deletions(-) create mode 100644 src/iceberg/test/replace_partitions_test.cc diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 7a1b45b70..e6c3410a8 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -244,7 +244,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/replace_partitions_test.cc b/src/iceberg/test/replace_partitions_test.cc new file mode 100644 index 000000000..c70e954c5 --- /dev/null +++ b/src/iceberg/test/replace_partitions_test.cc @@ -0,0 +1,328 @@ +/* + * 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/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/test/matchers.h" +#include "iceberg/test/update_test_base.h" +#include "iceberg/transaction.h" +#include "iceberg/transform.h" +#include "iceberg/update/fast_append.h" +#include "iceberg/update/row_delta.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; + } + + // 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()); + } + + 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"); + // file_b in partition 2 plus the new file in partition 1. + EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kTotalDataFiles), "2"); +} + +// 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 fails when data already exists in the target partition and +// the error is translated into Java's partition-conflict wording. +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")); +} + +// 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()); +} + +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")); +} + +} // namespace iceberg diff --git a/src/iceberg/update/replace_partitions.cc b/src/iceberg/update/replace_partitions.cc index 1a30a74c8..c8c1287db 100644 --- a/src/iceberg/update/replace_partitions.cc +++ b/src/iceberg/update/replace_partitions.cc @@ -20,6 +20,7 @@ #include "iceberg/update/replace_partitions.h" #include +#include #include "iceberg/expression/expressions.h" #include "iceberg/partition_spec.h" @@ -56,10 +57,7 @@ ReplacePartitions& ReplacePartitions::AddFile(const std::shared_ptr& f ICEBERG_BUILDER_ASSIGN_OR_RETURN(auto spec, base().PartitionSpecById(spec_id)); ICEBERG_BUILDER_RETURN_IF_ERROR(AddDataFile(file)); - // A spec is effectively unpartitioned if it has no fields, or every field - // uses the void transform. Java's PartitionSpec.isUnpartitioned() covers - // both cases; mirror that so an all-void spec triggers the table-wide - // replace path instead of a DropPartition with void-valued partition keys. + // Specs with no non-void transforms are unpartitioned. auto is_unpartitioned = [](const PartitionSpec& s) { return s.fields().empty() || std::all_of(s.fields().begin(), s.fields().end(), [](const PartitionField& f) { @@ -67,10 +65,8 @@ ReplacePartitions& ReplacePartitions::AddFile(const std::shared_ptr& f }); }; if (is_unpartitioned(*spec)) { - // Unpartitioned spec: Java's BaseReplacePartitions treats this as a - // table-wide replace rather than a spec-scoped DropPartition with empty - // partition values. Mirror that so every existing data file is dropped - // and conflict validation runs against AlwaysTrue. + // Unpartitioned: replace the whole table and validate conflicts against an + // AlwaysTrue row filter instead of a spec-scoped partition set. ICEBERG_BUILDER_RETURN_IF_ERROR(DeleteByRowFilter(Expressions::AlwaysTrue())); replace_by_row_filter_ = true; } else { @@ -102,15 +98,39 @@ ReplacePartitions& ReplacePartitions::ValidateNoConflictingDeletes() { std::string ReplacePartitions::operation() { return DataOperation::kOverwrite; } +Result> ReplacePartitions::Apply( + const TableMetadata& metadata_to_update, const std::shared_ptr& snapshot) { + auto result = MergingSnapshotUpdate::Apply(metadata_to_update, snapshot); + if (result.has_value()) { + return result; + } + // Translate the fail-any-delete error raised when ValidateAppendOnly() is set, + // matching Java BaseReplacePartitions.apply(). The base reports "Operation + // would delete existing data: "; surface the partition-conflict + // wording Java callers expect. + 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) { - // Match Java BaseReplacePartitions.validate: require at least one staged data - // file and that all staged files share exactly one partition spec. - // DataSpec() enforces both invariants; ignore the returned spec here since - // replace_by_row_filter_ / replaced_partitions_ already scope validation. - if (!replace_by_row_filter_) { - ICEBERG_RETURN_UNEXPECTED(DataSpec()); - } + // Require at least one staged data file, all sharing exactly one partition + // spec. DataSpec() enforces both invariants. Call it unconditionally, even on + // the row-filter path: if an unpartitioned/all-void file is staged first and a + // later file comes from a different spec, this rejects the mixed-spec replace + // instead of committing a table-wide replace. The returned spec is unused + // because replace_by_row_filter_ / replaced_partitions_ already scope + // validation. + ICEBERG_RETURN_UNEXPECTED(DataSpec()); if (snapshot == nullptr) { return {}; @@ -128,23 +148,23 @@ Status ReplacePartitions::Validate(const TableMetadata& current_metadata, } } if (validate_conflicting_deletes_) { - // Java's BaseReplacePartitions.validate gates both ValidateNoNewDeleteFiles - // and ValidateDeletedDataFiles on the same validateNewDeletes flag. The - // second check rejects concurrent overwrite/delete commits in the replaced - // partitions; without it a concurrent delete in a replaced partition would - // commit silently. + // Run ValidateDeletedDataFiles before ValidateNoNewDeleteFiles, matching + // Java, so the reported failure is deterministic when both conflict types + // are present. The first rejects concurrent removals of data in the + // replaced partitions (which would otherwise resurrect deleted rows); the + // second rejects delete files added concurrently in those partitions. if (replace_by_row_filter_) { - ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( + ICEBERG_RETURN_UNEXPECTED(ValidateDeletedDataFiles( current_metadata, starting_snapshot_id_, Expressions::AlwaysTrue(), snapshot, io, IsCaseSensitive())); - ICEBERG_RETURN_UNEXPECTED(ValidateDeletedDataFiles( + ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( current_metadata, starting_snapshot_id_, Expressions::AlwaysTrue(), snapshot, io, IsCaseSensitive())); } else { - ICEBERG_RETURN_UNEXPECTED(ValidateNoNewDeleteFiles( - current_metadata, starting_snapshot_id_, replaced_partitions_, snapshot, io)); 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 {}; diff --git a/src/iceberg/update/replace_partitions.h b/src/iceberg/update/replace_partitions.h index 70515054b..a9a51c5ae 100644 --- a/src/iceberg/update/replace_partitions.h +++ b/src/iceberg/update/replace_partitions.h @@ -25,6 +25,7 @@ #include #include #include +#include #include "iceberg/iceberg_export.h" #include "iceberg/result.h" @@ -98,11 +99,13 @@ class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { /// \return Reference to this for method chaining ReplacePartitions& ValidateNoConflictingData(); - /// \brief Enable validation that no conflicting delete files were added concurrently. + /// \brief Enable validation that no conflicting deletes occurred concurrently. /// - /// Fails the commit if a concurrent operation added a delete file covering - /// any of the partitions being replaced after the snapshot set by - /// ValidateFromSnapshot(). + /// Fails the commit if, after the snapshot set by ValidateFromSnapshot(), a + /// concurrent operation either added a delete file or removed a data file in + /// any of the partitions being replaced. Rejecting concurrent data-file + /// removals prevents this replace from undeleting rows that the concurrent + /// commit deleted. /// /// \return Reference to this for method chaining ReplacePartitions& ValidateNoConflictingDeletes(); @@ -113,6 +116,15 @@ class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { Status Validate(const TableMetadata& current_metadata, const std::shared_ptr& snapshot) override; + /// \brief Apply changes, translating the fail-any-delete error. + /// + /// Delegates to MergingSnapshotUpdate::Apply and, when ValidateAppendOnly() + /// triggers a delete, rewrites the error into the partition-conflict message + /// that Java's BaseReplacePartitions.apply() produces. + Result> Apply( + const TableMetadata& current_metadata, + const std::shared_ptr& snapshot) override; + private: explicit ReplacePartitions(std::string table_name, std::shared_ptr ctx); @@ -120,14 +132,11 @@ class ICEBERG_EXPORT ReplacePartitions : public MergingSnapshotUpdate { std::optional starting_snapshot_id_; bool validate_conflicting_data_{false}; bool validate_conflicting_deletes_{false}; - // True once an AddFile() call has staged a file whose partition spec is - // unpartitioned. Java's BaseReplacePartitions treats this case as a - // table-wide replace (DeleteByRowFilter(AlwaysTrue())) and runs conflict - // validation against AlwaysTrue rather than a partition set — mirror that. + // True when replacement uses an AlwaysTrue row filter. bool replace_by_row_filter_{false}; - // Partitions touched by AddFile() in partitioned specs. Used to scope - // conflict validation to the overwritten partitions in the partitioned - // case; ignored when replace_by_row_filter_ is true. + // Partitions touched by AddFile() for partitioned specs; scopes conflict + // validation to the overwritten partitions. Ignored when + // replace_by_row_filter_ is true. PartitionSet replaced_partitions_; };