awong/iceberg v2/equality deletes

src/v/iceberg/BUILD

@@ -444,6 +444,32 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "manifest_utils",
+    srcs = [
+        "manifest_utils.cc",
+    ],
+    hdrs = [
+        "manifest_utils.h",
+    ],
+    implementation_deps = [
+        ":compatibility",
+        ":logger",
+        ":manifest",
+        ":partition_key_type",
+        ":values_bytes",
+    ],
+    deps = [
+        ":action",
+        ":manifest_entry",
+        ":manifest_io",
+        ":manifest_list",
+        ":schema",
+        ":table_metadata",
+        "//src/v/base",
+    ],
+)
+
 redpanda_cc_library(
     name = "merge_append_action",
     srcs = [
@@ -453,9 +479,7 @@ redpanda_cc_library(
         "merge_append_action.h",
     ],
     implementation_deps = [
-        ":compatibility",
         ":logger",
-        ":manifest",
         ":manifest_file_packer",
         ":snapshot",
         ":table_requirement",
@@ -466,9 +490,37 @@ redpanda_cc_library(
         ":manifest_entry",
         ":manifest_io",
         ":manifest_list",
+        ":manifest_utils",
+        ":schema",
+        ":table_metadata",
+        "//src/v/base",
+    ],
+)
+
+redpanda_cc_library(
+    name = "row_delta_action",
+    srcs = [
+        "row_delta_action.cc",
+    ],
+    hdrs = [
+        "row_delta_action.h",
+    ],
+    implementation_deps = [
+        ":logger",
+        ":manifest_file_packer",
+        ":snapshot",
+        ":table_requirement",
+        "//src/v/random:generators",
+    ],
+    deps = [
+        ":action",
+        ":manifest_entry",
+        ":manifest_io",
+        ":manifest_list",
+        ":manifest_utils",
+        ":merge_append_action",
         ":schema",
         ":table_metadata",
-        ":values_bytes",
         "//src/v/base",
     ],
 )
@@ -554,6 +606,30 @@ redpanda_cc_library(
     ],
 )
 
+redpanda_cc_library(
+    name = "equality_delete_file",
+    srcs = [
+        "equality_delete_file.cc",
+    ],
+    hdrs = [
+        "equality_delete_file.h",
+    ],
+    implementation_deps = [
+        "//src/v/bytes:iostream",
+        "//src/v/serde/parquet:writer",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        ":datatypes",
+        ":manifest_entry",
+        "//src/v/bytes:iobuf",
+        "//src/v/container:chunked_vector",
+        "//src/v/serde/parquet:schema",
+        "//src/v/serde/parquet:value",
+        "@seastar",
+    ],
+)
+
 redpanda_cc_library(
     name = "partition_key_type",
     srcs = [

src/v/iceberg/equality_delete_file.cc

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+
+#include "iceberg/equality_delete_file.h"
+
+#include "bytes/iostream.h"
+#include "serde/parquet/writer.h"
+
+namespace iceberg {
+
+ss::future<delete_file_result> write_equality_delete_file(
+  equality_delete_options opts,
+  chunked_vector<serde::parquet::group_value> rows) {
+    iobuf file;
+    serde::parquet::writer w(
+      {.schema = std::move(opts.parquet_schema), .compress = opts.compress},
+      make_iobuf_ref_output_stream(file));
+    co_await w.init();
+
+    auto record_count = rows.size();
+    for (auto& row : rows) {
+        co_await w.write_row(std::move(row));
+    }
+
+    co_await w.close();
+
+    data_file df{
+      .content_type = data_file_content_type::equality_deletes,
+      .file_path = uri{},
+      .file_format = data_file_format::parquet,
+      .partition = partition_key{},
+      .record_count = record_count,
+      .file_size_bytes = file.size_bytes(),
+      .equality_ids = std::move(opts.equality_field_ids),
+    };
+
+    co_return delete_file_result{
+      .file_data = std::move(file),
+      .manifest_entry = std::move(df),
+    };
+}
+
+} // namespace iceberg

src/v/iceberg/equality_delete_file.h

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2026 Redpanda Data, Inc.
+ *
+ * Licensed as a Redpanda Enterprise file under the Redpanda Community
+ * License (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md
+ */
+#pragma once
+
+#include "bytes/iobuf.h"
+#include "container/chunked_vector.h"
+#include "iceberg/datatypes.h"
+#include "iceberg/manifest_entry.h"
+#include "serde/parquet/schema.h"
+#include "serde/parquet/value.h"
+
+#include <seastar/core/future.hh>
+
+namespace iceberg {
+
+struct delete_file_result {
+    iobuf file_data;
+    data_file manifest_entry;
+};
+
+struct equality_delete_options {
+    serde::parquet::schema_element parquet_schema;
+    chunked_vector<nested_field::id_t> equality_field_ids;
+    bool compress = false;
+};
+
+/// \brief Write an equality delete file in parquet format from key column
+/// values.
+///
+/// The schema must match the equality columns from the table schema. The
+/// resulting data_file has content_type=equality_deletes and equality_ids
+/// populated from the provided field IDs.
+ss::future<delete_file_result> write_equality_delete_file(
+  equality_delete_options opts,
+  chunked_vector<serde::parquet::group_value> rows);
+
+} // namespace iceberg