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/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 8315d9d52..0f700e780 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/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 <memory>
+#include <string>
+#include <vector>
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#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<DataFile> MakeDataFile(const std::string& path, int64_t partition_x) {
+    auto f = std::make_shared<DataFile>();
+    f->content = DataFile::Content::kData;
+    f->file_path = table_location_ + path;
+    f->file_format = FileFormatType::kParquet;
+    f->partition = PartitionValues(std::vector<Literal>{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<DataFile> MakeUnpartitionedFile(const std::string& path,
+                                                  int32_t spec_id) {
+    auto f = std::make_shared<DataFile>();
+    f->content = DataFile::Content::kData;
+    f->file_path = table_location_ + path;
+    f->file_format = FileFormatType::kParquet;
+    f->partition = PartitionValues(std::vector<Literal>{});
+    f->file_size_in_bytes = 1024;
+    f->record_count = 100;
+    f->partition_spec_id = spec_id;
+    return f;
+  }
+
+  std::shared_ptr<DataFile> 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<PartitionSpec>(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<PartitionSpec>(std::move(spec)));
+    ASSERT_THAT(table_->metadata()->PartitionSpecById(spec_id), IsOk());
+  }
+
+  Result<std::unique_ptr<ReplacePartitions>> 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<DataFile>& 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<PartitionSpec> spec_;
+  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<DataFile> file_a_;
+  std::shared_ptr<DataFile> 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/type_fwd.h b/src/iceberg/type_fwd.h
index 91e18b24c..53253dd45 100644
--- a/src/iceberg/type_fwd.h
+++ b/src/iceberg/type_fwd.h
@@ -243,6 +243,7 @@ class TransactionContext;
 class DeleteFiles;
 class ExpireSnapshots;
 class FastAppend;
+class ReplacePartitions;
 class MergeAppend;
 class OverwriteFiles;
 class PendingUpdate;
diff --git a/src/iceberg/update/merging_snapshot_update.h b/src/iceberg/update/merging_snapshot_update.h
index 1c08a0667..299680f4c 100644
--- a/src/iceberg/update/merging_snapshot_update.h
+++ b/src/iceberg/update/merging_snapshot_update.h
@@ -307,6 +307,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;
@@ -345,8 +353,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();
 
   /// \brief Write new data manifests for staged data files; caches the result.
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..c8c1287db
--- /dev/null
+++ b/src/iceberg/update/replace_partitions.cc
@@ -0,0 +1,173 @@
+/*
+ * 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 <algorithm>
+#include <string_view>
+
+#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"
+
+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));
+  // 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) {
+             return f.transform()->transform_type() == TransformType::kVoid;
+           });
+  };
+  if (is_unpartitioned(*spec)) {
+    // 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 {
+    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; }
+
+Result<std::vector<ManifestFile>> ReplacePartitions::Apply(
+    const TableMetadata& metadata_to_update, const std::shared_ptr<Snapshot>& 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: <partition>"; 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>& snapshot) {
+  // 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 {};
+  }
+
+  auto io = ctx_->table->io();
+  if (validate_conflicting_data_) {
+    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_) {
+    // 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(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..a9a51c5ae
--- /dev/null
+++ b/src/iceberg/update/replace_partitions.h
@@ -0,0 +1,143 @@
+/*
+ * 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 <vector>
+
+#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.
+///
+/// \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.
+  ///
+  /// \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 deletes occurred concurrently.
+  ///
+  /// 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();
+
+  std::string operation() override;
+
+ protected:
+  Status Validate(const TableMetadata& current_metadata,
+                  const std::shared_ptr<Snapshot>& 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<std::vector<ManifestFile>> Apply(
+      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};
+  // True when replacement uses an AlwaysTrue row filter.
+  bool replace_by_row_filter_{false};
+  // 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_;
+};
+
+}  // namespace iceberg