feat: add ReplacePartitions core class (PR1)

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

Author: shangxinli

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

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',

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.

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;

src/iceberg/update/merging_snapshot_update.h

@@ -296,6 +296,14 @@ class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
                           const std::shared_ptr<Snapshot>& parent,
                           std::shared_ptr<FileIO> 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<DataFile> 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<std::vector<PendingDeleteFile>> MergeDVs() const;
 
   /// \brief Write new data manifests for staged data files; caches the result.

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<std::unique_ptr<ReplacePartitions>> ReplacePartitions::Make(
+    std::string table_name, std::shared_ptr<TransactionContext> 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<ReplacePartitions>(
+      new ReplacePartitions(std::move(table_name), std::move(ctx)));
+}
+
+ReplacePartitions::ReplacePartitions(std::string table_name,
+                                     std::shared_ptr<TransactionContext> ctx)
+    : MergingSnapshotUpdate(std::move(table_name), std::move(ctx)) {
+  SetSummaryProperty(SnapshotSummaryFields::kReplacePartitions, "true");
+}
+
+ReplacePartitions& ReplacePartitions::AddFile(const std::shared_ptr<DataFile>& 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>& 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

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 <cstdint>
+#include <memory>
+#include <optional>
+#include <string>
+
+#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<std::unique_ptr<ReplacePartitions>> Make(
+      std::string table_name, std::shared_ptr<TransactionContext> 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<DataFile>& 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>& snapshot) override;
+
+ private:
+  explicit ReplacePartitions(std::string table_name,
+                             std::shared_ptr<TransactionContext> ctx);
+
+  std::optional<int64_t> 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