feat: add ReplacePartitions core class (PR1 of 2 for #637)

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

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

src/iceberg/partition_spec.cc

@@ -66,6 +66,12 @@ int32_t PartitionSpec::spec_id() const { return spec_id_; }
 
 std::span<const PartitionField> PartitionSpec::fields() const { return fields_; }
 
+bool PartitionSpec::IsUnpartitioned() const {
+  return std::ranges::all_of(fields_, [](const PartitionField& field) {
+    return field.transform()->transform_type() == TransformType::kVoid;
+  });
+}
+
 Result<std::unique_ptr<StructType>> PartitionSpec::PartitionType(
     const Schema& schema) const {
   if (fields_.empty()) {

src/iceberg/partition_spec.h

@@ -61,6 +61,9 @@ class ICEBERG_EXPORT PartitionSpec : public util::Formattable {
   /// \brief Get a list view of the partition fields.
   std::span<const PartitionField> fields() const;
 
+  /// \brief Return whether this spec has no non-void partition fields.
+  bool IsUnpartitioned() const;
+
   /// \brief Get the partition type binding to the input schema.
   Result<std::unique_ptr<StructType>> PartitionType(const Schema& schema) const;
 

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.

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

src/iceberg/test/partition_spec_test.cc

@@ -60,6 +60,20 @@ TEST(PartitionSpecTest, Basics) {
   }
 }
 
+TEST(PartitionSpecTest, IsUnpartitioned) {
+  ICEBERG_UNWRAP_OR_FAIL(auto empty_spec, PartitionSpec::Make(100, {}));
+  EXPECT_TRUE(empty_spec->IsUnpartitioned());
+
+  PartitionField void_field(5, 1000, "void", Transform::Void());
+  ICEBERG_UNWRAP_OR_FAIL(auto all_void_spec, PartitionSpec::Make(101, {void_field}));
+  EXPECT_TRUE(all_void_spec->IsUnpartitioned());
+
+  PartitionField identity_field(7, 1001, "identity", Transform::Identity());
+  ICEBERG_UNWRAP_OR_FAIL(auto partitioned_spec,
+                         PartitionSpec::Make(102, {void_field, identity_field}));
+  EXPECT_FALSE(partitioned_spec->IsUnpartitioned());
+}
+
 TEST(PartitionSpecTest, Equality) {
   SchemaField field1(5, "ts", timestamp(), true);
   SchemaField field2(7, "bar", string(), true);