awong/iceberg v2/equality deletes

src/v/iceberg/BUILD

@@ -554,6 +554,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

src/v/iceberg/tests/BUILD

@@ -727,6 +727,21 @@ redpanda_cc_bench(
     ],
 )
 
+redpanda_cc_gtest(
+    name = "equality_delete_file_test",
+    timeout = "short",
+    srcs = [
+        "equality_delete_file_test.cc",
+    ],
+    deps = [
+        "//src/v/iceberg:equality_delete_file",
+        "//src/v/serde/parquet:reader",
+        "//src/v/serde/parquet:value",
+        "//src/v/test_utils:gtest",
+        "@googletest//:gtest",
+    ],
+)
+
 redpanda_cc_bench(
     name = "iceberg_compatibility_rpbench",
     srcs = [

src/v/iceberg/tests/equality_delete_file_test.cc

@@ -0,0 +1,134 @@
+/*
+ * 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 "serde/parquet/reader.h"
+#include "serde/parquet/value.h"
+
+#include <gtest/gtest.h>
+
+namespace iceberg {
+
+namespace {
+
+serde::parquet::schema_element two_column_key_schema() {
+    using serde::parquet::byte_array_type;
+    using serde::parquet::field_repetition_type;
+    using serde::parquet::i32_type;
+    using serde::parquet::schema_element;
+    using serde::parquet::string_type;
+
+    chunked_vector<schema_element> children;
+    children.push_back(
+      schema_element{
+        .type = i32_type{},
+        .repetition_type = field_repetition_type::required,
+        .path = {ss::sstring("id")},
+        .field_id = 1,
+      });
+    children.push_back(
+      schema_element{
+        .type = byte_array_type{},
+        .repetition_type = field_repetition_type::required,
+        .path = {ss::sstring("name")},
+        .field_id = 2,
+        .logical_type = string_type{},
+      });
+    return schema_element{
+      .type = std::monostate{},
+      .repetition_type = field_repetition_type::required,
+      .path = {ss::sstring("equality_deletes")},
+      .children = std::move(children),
+    };
+}
+
+serde::parquet::group_value make_row(int32_t id, const ss::sstring& name) {
+    serde::parquet::group_value row;
+    row.push_back(
+      serde::parquet::group_member{serde::parquet::int32_value{id}});
+    row.push_back(
+      serde::parquet::group_member{
+        serde::parquet::byte_array_value{iobuf::from(name)}});
+    return row;
+}
+
+} // namespace
+
+// NOLINTBEGIN(*magic-number*)
+
+TEST(EqualityDeleteFile, RoundTrip) {
+    chunked_vector<serde::parquet::group_value> rows;
+    rows.push_back(make_row(1, "alice"));
+    rows.push_back(make_row(2, "bob"));
+    rows.push_back(make_row(3, "charlie"));
+
+    chunked_vector<nested_field::id_t> field_ids;
+    field_ids.push_back(nested_field::id_t{1});
+    field_ids.push_back(nested_field::id_t{2});
+
+    equality_delete_options opts{
+      .parquet_schema = two_column_key_schema(),
+      .equality_field_ids = std::move(field_ids),
+    };
+
+    auto result
+      = write_equality_delete_file(std::move(opts), std::move(rows)).get();
+
+    EXPECT_EQ(
+      result.manifest_entry.content_type,
+      data_file_content_type::equality_deletes);
+    EXPECT_EQ(result.manifest_entry.file_format, data_file_format::parquet);
+    EXPECT_EQ(result.manifest_entry.record_count, 3);
+    EXPECT_GT(result.manifest_entry.file_size_bytes, 0);
+
+    ASSERT_TRUE(result.manifest_entry.equality_ids.has_value());
+    auto& eq_ids = result.manifest_entry.equality_ids.value();
+    ASSERT_EQ(eq_ids.size(), 2);
+    EXPECT_EQ(eq_ids[0], nested_field::id_t{1});
+    EXPECT_EQ(eq_ids[1], nested_field::id_t{2});
+
+    auto records
+      = serde::parquet::read_file_as_records(std::move(result.file_data)).get();
+
+    ASSERT_EQ(records.size(), 3);
+    EXPECT_EQ(records[0], make_row(1, "alice"));
+    EXPECT_EQ(records[1], make_row(2, "bob"));
+    EXPECT_EQ(records[2], make_row(3, "charlie"));
+}
+
+TEST(EqualityDeleteFile, RoundTripCompressed) {
+    chunked_vector<serde::parquet::group_value> rows;
+    rows.push_back(make_row(42, "eve"));
+
+    chunked_vector<nested_field::id_t> field_ids;
+    field_ids.push_back(nested_field::id_t{1});
+    field_ids.push_back(nested_field::id_t{2});
+
+    equality_delete_options opts{
+      .parquet_schema = two_column_key_schema(),
+      .equality_field_ids = std::move(field_ids),
+      .compress = true,
+    };
+
+    auto result
+      = write_equality_delete_file(std::move(opts), std::move(rows)).get();
+
+    EXPECT_EQ(result.manifest_entry.record_count, 1);
+
+    auto records
+      = serde::parquet::read_file_as_records(std::move(result.file_data)).get();
+
+    ASSERT_EQ(records.size(), 1);
+    EXPECT_EQ(records[0], make_row(42, "eve"));
+}
+
+// NOLINTEND(*magic-number*)
+
+} // namespace iceberg