feat: add ReplacePartitions core class and tests (PR1)

Adds iceberg::ReplacePartitions — the dynamic partition overwrite
operation. Each AddFile() registers the file's partition for
replacement of all existing data 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)
rather than SnapshotUpdate. This is load-bearing: MergingSnapshotUpdate
handles both data and delete manifest pipelines, preserves custom
summary properties across commit retry, and exposes the validator
helpers needed for conflict detection. An earlier draft extended
SnapshotUpdate directly and reimplemented the manifest pipeline by
hand; that draft dropped parent delete manifests and lost custom
summary entries — both bugs disappear under the merging-update design.

Validation is partition-scoped via DropPartition(spec_id, partition):
unpartitioned specs naturally use an empty partition value, so the
filter manager matches every file in that spec without needing a
separate AlwaysTrue path. Validate() uses the PartitionSet overloads
of ValidateAddedDataFiles / ValidateNoNewDeleteFiles when partitions
are tracked and skips entirely when nothing was staged.

Tests (replace_partitions_test.cc, parameterized over format v2/v3 and
main/feature branch):
  * AddFile null / no-spec-id rejection
  * operation() == "overwrite"
  * Empty-table replace writes overwrite snapshot with replace-partitions=true
  * Replace overwrites only the touched partition (existing file in
    untouched partition survives; existing files in touched partition
    marked deleted; new file added)
  * ValidateAppendOnly fails when target partition has data, succeeds
    when target partition is empty

API surface added:
  * Table::NewReplacePartitions() / StaticTable rejects
  * Transaction::NewReplacePartitions()
  * SnapshotSummaryFields::kReplacePartitions = "replace-partitions"
  * MergingSnapshotUpdate::SetSummaryProperty promoted from private to
    protected so subclasses can stash custom summary entries through
    the cached-rebuild path.

Closes #637.
Tracking: #775

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/table.cc

@@ -35,6 +35,7 @@
 #include "iceberg/update/expire_snapshots.h"
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/merge_append.h"
+#include "iceberg/update/replace_partitions.h"
 #include "iceberg/update/row_delta.h"
 #include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/snapshot_manager.h"
@@ -238,6 +239,12 @@ Result<std::shared_ptr<RowDelta>> Table::NewRowDelta() {
   return RowDelta::Make(name().name, std::move(ctx));
 }
 
+Result<std::shared_ptr<ReplacePartitions>> Table::NewReplacePartitions() {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
+  return ReplacePartitions::Make(name().name, std::move(ctx));
+}
+
 Result<std::shared_ptr<UpdateStatistics>> Table::NewUpdateStatistics() {
   ICEBERG_ASSIGN_OR_RAISE(
       auto ctx, TransactionContext::Make(shared_from_this(), TransactionKind::kUpdate));
@@ -349,6 +356,10 @@ Result<std::shared_ptr<RowDelta>> StaticTable::NewRowDelta() {
   return NotSupported("Cannot create a row delta for a static table");
 }
 
+Result<std::shared_ptr<ReplacePartitions>> StaticTable::NewReplacePartitions() {
+  return NotSupported("Cannot create a replace partitions for a static table");
+}
+
 Result<std::shared_ptr<SnapshotManager>> StaticTable::NewSnapshotManager() {
   return NotSupported("Cannot create a snapshot manager for a static table");
 }

src/iceberg/table.h

@@ -185,6 +185,10 @@ class ICEBERG_EXPORT Table : public std::enable_shared_from_this<Table> {
   /// \brief Create a new RowDelta to add rows and row-level deletes.
   virtual Result<std::shared_ptr<RowDelta>> NewRowDelta();
 
+  /// \brief Create a new ReplacePartitions to dynamically overwrite partitions
+  /// touched by added data files.
+  virtual Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions();
+
   /// \brief Create a new SnapshotManager to manage snapshots and snapshot references.
   virtual Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();
 
@@ -257,6 +261,7 @@ class ICEBERG_EXPORT StaticTable : public Table {
   Result<std::shared_ptr<DeleteFiles>> NewDeleteFiles() override;
 
   Result<std::shared_ptr<RowDelta>> NewRowDelta() override;
+  Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions() override;
 
   Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager() override;
 

src/iceberg/test/CMakeLists.txt

@@ -230,6 +230,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    merge_append_test.cc
                    merging_snapshot_update_test.cc
                    name_mapping_update_test.cc
+                   replace_partitions_test.cc
                    row_delta_test.cc
                    snapshot_manager_test.cc
                    transaction_test.cc

src/iceberg/test/replace_partitions_test.cc

@@ -0,0 +1,275 @@
+/*
+ * 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 <format>
+#include <memory>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/avro/avro_register.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/manifest/manifest_list.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/partition_spec.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/test_resource.h"
+#include "iceberg/test/update_test_base.h"
+#include "iceberg/transaction.h"
+#include "iceberg/update/fast_append.h"
+#include "iceberg/util/uuid.h"
+
+namespace iceberg {
+
+class ReplacePartitionsTestBase : public UpdateTestBase {
+ protected:
+  static void SetUpTestSuite() { avro::RegisterAll(); }
+
+  std::string TableName() const override { return "minimal_table"; }
+
+  void SetUp() override {
+    table_ident_ = TableIdentifier{.name = TableName()};
+    table_location_ = "/warehouse/" + TableName();
+
+    InitializeFileIO();
+    RegisterMinimalTable(format_version());
+
+    ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec());
+    ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema());
+
+    file_p1_a_ = MakeDataFile("/data/p1_a.parquet", /*partition_x=*/1L);
+    file_p1_b_ = MakeDataFile("/data/p1_b.parquet", /*partition_x=*/1L);
+    file_p2_a_ = MakeDataFile("/data/p2_a.parquet", /*partition_x=*/2L);
+    file_p1_new_ = MakeDataFile("/data/p1_new.parquet", /*partition_x=*/1L);
+  }
+
+  void RegisterMinimalTable(int8_t format_version) {
+    auto metadata_location = std::format("{}/metadata/00001-{}.metadata.json",
+                                         table_location_, Uuid::GenerateV7().ToString());
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto metadata, ReadTableMetadataFromResource("TableMetadataV2ValidMinimal.json"));
+    metadata->format_version = format_version;
+    metadata->location = table_location_;
+    metadata->next_row_id = TableMetadata::kInitialRowId;
+
+    ASSERT_THAT(TableMetadataUtil::Write(*file_io_, metadata_location, *metadata),
+                IsOk());
+    ICEBERG_UNWRAP_OR_FAIL(table_,
+                           catalog_->RegisterTable(table_ident_, metadata_location));
+  }
+
+  virtual int8_t format_version() const {
+    return TableMetadata::kDefaultTableFormatVersion;
+  }
+
+  virtual std::string branch() const { return std::string(SnapshotRef::kMainBranch); }
+
+  std::shared_ptr<DataFile> MakeDataFile(const std::string& path, int64_t partition_x) {
+    return MakeDataFile(path, spec_, {Literal::Long(partition_x)});
+  }
+
+  std::shared_ptr<DataFile> MakeDataFile(const std::string& path,
+                                         std::shared_ptr<PartitionSpec> spec,
+                                         std::vector<Literal> partition_values) {
+    auto file = std::make_shared<DataFile>();
+    file->content = DataFile::Content::kData;
+    file->file_path = table_location_ + path;
+    file->file_format = FileFormatType::kParquet;
+    file->partition = PartitionValues(std::move(partition_values));
+    file->file_size_in_bytes = 1024;
+    file->record_count = 100;
+    file->partition_spec_id = spec->spec_id();
+    return file;
+  }
+
+  Result<std::shared_ptr<ReplacePartitions>> NewBranchReplacePartitions() {
+    ICEBERG_ASSIGN_OR_RAISE(auto replace, table_->NewReplacePartitions());
+    replace->SetTargetBranch(branch());
+    return replace;
+  }
+
+  Result<std::shared_ptr<FastAppend>> NewBranchFastAppend() {
+    ICEBERG_ASSIGN_OR_RAISE(auto append, table_->NewFastAppend());
+    append->SetTargetBranch(branch());
+    return append;
+  }
+
+  Result<std::shared_ptr<Snapshot>> CurrentSnapshot() {
+    const auto& refs = table_->metadata()->refs;
+    auto it = refs.find(branch());
+    if (it == refs.end()) {
+      return NotFound("No snapshot ref for branch '{}'", branch());
+    }
+    return table_->SnapshotById(it->second->snapshot_id);
+  }
+
+  Result<std::vector<ManifestFile>> CurrentDataManifests() {
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot, CurrentSnapshot());
+    SnapshotCache snapshot_cache(snapshot.get());
+    ICEBERG_ASSIGN_OR_RAISE(auto manifests, snapshot_cache.DataManifests(file_io_));
+    return std::vector<ManifestFile>(manifests.begin(), manifests.end());
+  }
+
+  std::shared_ptr<PartitionSpec> spec_;
+  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<DataFile> file_p1_a_;
+  std::shared_ptr<DataFile> file_p1_b_;
+  std::shared_ptr<DataFile> file_p2_a_;
+  std::shared_ptr<DataFile> file_p1_new_;
+};
+
+class ReplacePartitionsTest
+    : public ReplacePartitionsTestBase,
+      public ::testing::WithParamInterface<std::tuple<int8_t, std::string>> {
+ protected:
+  int8_t format_version() const override { return std::get<0>(GetParam()); }
+  std::string branch() const override { return std::get<1>(GetParam()); }
+};
+
+TEST_F(ReplacePartitionsTestBase, AddNullFile) {
+  ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
+  replace->AddFile(nullptr);
+
+  auto result = replace->Commit();
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+}
+
+TEST_F(ReplacePartitionsTestBase, AddFileWithoutSpecIdRejected) {
+  auto file = MakeDataFile("/data/no_spec.parquet", /*partition_x=*/1L);
+  file->partition_spec_id = std::nullopt;
+
+  ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
+  replace->AddFile(file);
+
+  auto result = replace->Commit();
+  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
+}
+
+TEST_P(ReplacePartitionsTest, OperationIsOverwrite) {
+  ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
+  EXPECT_EQ(replace->operation(), DataOperation::kOverwrite);
+}
+
+TEST_P(ReplacePartitionsTest, EmptyTableReplaceWritesOverwriteSnapshot) {
+  EXPECT_THAT(CurrentSnapshot(), IsError(ErrorKind::kNotFound));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
+  replace->AddFile(file_p1_a_).AddFile(file_p2_a_);
+  EXPECT_THAT(replace->Commit(), IsOk());
+
+  EXPECT_THAT(table_->Refresh(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, CurrentSnapshot());
+  EXPECT_EQ(snapshot->Operation(), DataOperation::kOverwrite);
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kReplacePartitions), "true");
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "2");
+
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentDataManifests());
+  ASSERT_EQ(manifests.size(), 1U);
+}
+
+TEST_P(ReplacePartitionsTest, ReplaceOverwritesOnlyTouchedPartitions) {
+  // Seed the table with two partitions worth of data.
+  {
+    ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchFastAppend());
+    append->AppendFile(file_p1_a_).AppendFile(file_p1_b_).AppendFile(file_p2_a_);
+    EXPECT_THAT(append->Commit(), IsOk());
+    EXPECT_THAT(table_->Refresh(), IsOk());
+  }
+
+  // Now replace partition 1 only.
+  ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
+  replace->AddFile(file_p1_new_);
+  EXPECT_THAT(replace->Commit(), IsOk());
+  EXPECT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto snapshot, CurrentSnapshot());
+  EXPECT_EQ(snapshot->Operation(), DataOperation::kOverwrite);
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kReplacePartitions), "true");
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kAddedDataFiles), "1");
+  // file_p1_a_ and file_p1_b_ should be marked deleted; file_p2_a_ should not.
+  EXPECT_EQ(snapshot->summary.at(SnapshotSummaryFields::kDeletedDataFiles), "2");
+
+  // Verify the surviving live data file set: file_p2_a_ + file_p1_new_.
+  ICEBERG_UNWRAP_OR_FAIL(auto manifests, CurrentDataManifests());
+  std::unordered_set<std::string> live_paths;
+  for (const auto& manifest : manifests) {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto spec, table_->metadata()->PartitionSpecById(manifest.partition_spec_id));
+    ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                           ManifestReader::Make(manifest, file_io_, schema_, spec));
+    ICEBERG_UNWRAP_OR_FAIL(auto entries, reader->Entries());
+    for (const auto& entry : entries) {
+      if (entry.status != ManifestStatus::kDeleted) {
+        live_paths.insert(entry.data_file->file_path);
+      }
+    }
+  }
+  EXPECT_THAT(live_paths, ::testing::UnorderedElementsAre(file_p2_a_->file_path,
+                                                          file_p1_new_->file_path));
+}
+
+TEST_P(ReplacePartitionsTest, ValidateAppendOnlyFailsWhenPartitionHasData) {
+  // Seed partition 1 with a file.
+  {
+    ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchFastAppend());
+    append->AppendFile(file_p1_a_);
+    EXPECT_THAT(append->Commit(), IsOk());
+    EXPECT_THAT(table_->Refresh(), IsOk());
+  }
+
+  // ValidateAppendOnly should reject replacing partition 1 because it already
+  // contains data.
+  ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
+  replace->ValidateAppendOnly();
+  replace->AddFile(file_p1_new_);
+  EXPECT_THAT(replace->Commit(), IsError(ErrorKind::kValidationFailed));
+}
+
+TEST_P(ReplacePartitionsTest, ValidateAppendOnlyAllowsAddToEmptyPartition) {
+  // Seed partition 1 only.
+  {
+    ICEBERG_UNWRAP_OR_FAIL(auto append, NewBranchFastAppend());
+    append->AppendFile(file_p1_a_);
+    EXPECT_THAT(append->Commit(), IsOk());
+    EXPECT_THAT(table_->Refresh(), IsOk());
+  }
+
+  // Replacing the empty partition 2 with append-only validation should succeed.
+  ICEBERG_UNWRAP_OR_FAIL(auto replace, NewBranchReplacePartitions());
+  replace->ValidateAppendOnly();
+  replace->AddFile(file_p2_a_);
+  EXPECT_THAT(replace->Commit(), IsOk());
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    FormatVersionAndBranch, ReplacePartitionsTest,
+    ::testing::Combine(::testing::Values(2, 3),
+                       ::testing::Values(std::string(SnapshotRef::kMainBranch),
+                                         std::string("feature"))));
+
+}  // namespace iceberg

src/iceberg/transaction.cc

@@ -37,6 +37,7 @@
 #include "iceberg/update/fast_append.h"
 #include "iceberg/update/merge_append.h"
 #include "iceberg/update/pending_update.h"
+#include "iceberg/update/replace_partitions.h"
 #include "iceberg/update/row_delta.h"
 #include "iceberg/update/set_snapshot.h"
 #include "iceberg/update/snapshot_manager.h"
@@ -513,6 +514,13 @@ Result<std::shared_ptr<RowDelta>> Transaction::NewRowDelta() {
   return row_delta;
 }
 
+Result<std::shared_ptr<ReplacePartitions>> Transaction::NewReplacePartitions() {
+  ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<ReplacePartitions> replace_partitions,
+                          ReplacePartitions::Make(ctx_->table->name().name, ctx_));
+  ICEBERG_RETURN_UNEXPECTED(AddUpdate(replace_partitions));
+  return replace_partitions;
+}
+
 Result<std::shared_ptr<UpdateStatistics>> Transaction::NewUpdateStatistics() {
   ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<UpdateStatistics> update_statistics,
                           UpdateStatistics::Make(ctx_));

src/iceberg/transaction.h

@@ -115,6 +115,10 @@ class ICEBERG_EXPORT Transaction : public std::enable_shared_from_this<Transacti
   /// \brief Create a new RowDelta to add rows and row-level deletes.
   Result<std::shared_ptr<RowDelta>> NewRowDelta();
 
+  /// \brief Create a new ReplacePartitions to dynamically overwrite partitions
+  /// touched by added data files.
+  Result<std::shared_ptr<ReplacePartitions>> NewReplacePartitions();
+
   /// \brief Create a new SnapshotManager to manage snapshots.
   Result<std::shared_ptr<SnapshotManager>> NewSnapshotManager();