Add DeleteFiles table update API

src/iceberg/json_serde.cc

@@ -196,6 +196,7 @@ constexpr std::string_view kActionRemoveStatistics = "remove-statistics";
 constexpr std::string_view kActionSetPartitionStatistics = "set-partition-statistics";
 constexpr std::string_view kActionRemovePartitionStatistics =
     "remove-partition-statistics";
+constexpr std::string_view kActionDeleteFiles = "delete-files";
 
 // TableUpdate field constants
 constexpr std::string_view kUUID = "uuid";
@@ -209,6 +210,7 @@ constexpr std::string_view kSnapshotIds = "snapshot-ids";
 constexpr std::string_view kRefName = "ref-name";
 constexpr std::string_view kUpdates = "updates";
 constexpr std::string_view kRemovals = "removals";
+constexpr std::string_view kFiles = "files";
 
 // TableRequirement type constants
 constexpr std::string_view kRequirementAssertDoesNotExist = "assert-create";
@@ -1471,6 +1473,12 @@ nlohmann::json ToJson(const TableUpdate& update) {
       json[kSnapshotId] = u.snapshot_id();
       break;
     }
+    case TableUpdate::Kind::kDeleteFiles: {
+      const auto& u = internal::checked_cast<const table::DeleteFiles&>(update);
+      json[kAction] = kActionDeleteFiles;
+      json[kFiles] = u.files();
+      break;
+    }
   }
   return json;
 }
@@ -1672,6 +1680,11 @@ Result<std::unique_ptr<TableUpdate>> TableUpdateFromJson(const nlohmann::json& j
     ICEBERG_ASSIGN_OR_RAISE(auto snapshot_id, GetJsonValue<int64_t>(json, kSnapshotId));
     return std::make_unique<table::RemovePartitionStatistics>(snapshot_id);
   }
+  if (action == kActionDeleteFiles) {
+    ICEBERG_ASSIGN_OR_RAISE(auto files,
+                            GetJsonValue<std::vector<std::string>>(json, kFiles));
+    return std::make_unique<table::DeleteFiles>(std::move(files));
+  }
 
   return JsonParseError("Unknown table update action: {}", action);
 }

src/iceberg/table_update.cc

@@ -565,4 +565,26 @@ std::unique_ptr<TableUpdate> RemovePartitionStatistics::Clone() const {
   return std::make_unique<RemovePartitionStatistics>(snapshot_id_);
 }
 
+// DeleteFiles
+
+void DeleteFiles::ApplyTo(TableMetadataBuilder& builder) const {
+  // DeleteFiles does not alter table metadata.
+}
+
+void DeleteFiles::GenerateRequirements(TableUpdateContext& context) const {
+  // DeleteFiles does not generate any requirements.
+}
+
+bool DeleteFiles::Equals(const TableUpdate& other) const {
+  if (other.kind() != Kind::kDeleteFiles) {
+    return false;
+  }
+  const auto& other_delete = internal::checked_cast<const DeleteFiles&>(other);
+  return files_ == other_delete.files_;
+}
+
+std::unique_ptr<TableUpdate> DeleteFiles::Clone() const {
+  return std::make_unique<DeleteFiles>(files_);
+}
+
 }  // namespace iceberg::table

src/iceberg/table_update.h

@@ -63,6 +63,7 @@ class ICEBERG_EXPORT TableUpdate {
     kRemoveStatistics,
     kSetPartitionStatistics,
     kRemovePartitionStatistics,
+    kDeleteFiles,
   };
 
   virtual ~TableUpdate();
@@ -608,6 +609,31 @@ class ICEBERG_EXPORT RemovePartitionStatistics : public TableUpdate {
   int64_t snapshot_id_;
 };
 
+/// \brief Represents deleting one or more physical files.
+///
+/// Similar to Iceberg's Java DeleteFiles API, file paths are treated as exact
+/// metadata values. Equivalent but differently-normalized URIs are not
+/// considered matches.
+class ICEBERG_EXPORT DeleteFiles : public TableUpdate {
+ public:
+  explicit DeleteFiles(std::vector<std::string> files) : files_(std::move(files)) {}
+
+  const std::vector<std::string>& files() const { return files_; }
+
+  void ApplyTo(TableMetadataBuilder& builder) const override;
+
+  void GenerateRequirements(TableUpdateContext& context) const override;
+
+  Kind kind() const override { return Kind::kDeleteFiles; }
+
+  bool Equals(const TableUpdate& other) const override;
+
+  std::unique_ptr<TableUpdate> Clone() const override;
+
+ private:
+  std::vector<std::string> files_;
+};
+
 }  // namespace table
 
 }  // namespace iceberg

src/iceberg/test/json_serde_test.cc

@@ -649,6 +649,23 @@ TEST(JsonInternalTest, TableUpdateRemovePartitionStatistics) {
       update);
 }
 
+TEST(JsonInternalTest, TableUpdateDeleteFiles) {
+  table::DeleteFiles update({"s3://bucket/warehouse/table/metadata/old-1.json",
+                             "s3://bucket/warehouse/table/metadata/old-2.json"});
+  nlohmann::json expected = R"({
+    "action":"delete-files",
+    "files":[
+      "s3://bucket/warehouse/table/metadata/old-1.json",
+      "s3://bucket/warehouse/table/metadata/old-2.json"
+    ]
+  })"_json;
+
+  EXPECT_EQ(ToJson(update), expected);
+  auto parsed = TableUpdateFromJson(expected);
+  ASSERT_THAT(parsed, IsOk());
+  EXPECT_EQ(*internal::checked_cast<table::DeleteFiles*>(parsed.value().get()), update);
+}
+
 TEST(JsonInternalTest, TableUpdateUnknownAction) {
   nlohmann::json json = R"({"action":"unknown-action"})"_json;
   auto result = TableUpdateFromJson(json);