diff --git a/src/v/cloud_topics/level_one/maintenance/compaction/BUILD b/src/v/cloud_topics/level_one/maintenance/compaction/BUILD
index 09375dbd5b009..59fc207fb5abe 100644
--- a/src/v/cloud_topics/level_one/maintenance/compaction/BUILD
+++ b/src/v/cloud_topics/level_one/maintenance/compaction/BUILD
@@ -29,6 +29,7 @@ redpanda_cc_library(
     deps = [
         "//src/v/cloud_topics/level_one/metastore:offset_interval_set",
         "//src/v/compaction:filter",
+        "//src/v/compaction:key",
         "//src/v/compaction:key_offset_map",
         "//src/v/compaction:utils",
         "//src/v/model",
diff --git a/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.cc b/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.cc
index da6991c26cac9..452fefe579404 100644
--- a/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.cc
+++ b/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.cc
@@ -10,9 +10,11 @@
 
 #include "cloud_topics/level_one/maintenance/compaction/compaction_filter.h"
 
+#include "compaction/key.h"
 #include "compaction/utils.h"
 #include "model/fundamental.h"
 #include "model/record.h"
+#include "model/record_utils.h"
 
 #include <seastar/core/coroutine.hh>
 #include <seastar/core/future.hh>
@@ -32,27 +34,28 @@ compaction_filter::compaction_filter(
   , _removable_tombstone_ranges(removable_tombstone_ranges) {}
 
 ss::future<bool> compaction_filter::should_keep(
-  const model::record_batch& b, const model::record& r) const {
-    if (r.is_tombstone()) {
-        auto o = model::offset_cast(
-          b.base_offset() + model::offset_delta(r.offset_delta()));
-        if (_removable_tombstone_ranges.contains(o)) {
+  model::offset o,
+  bool is_tombstone,
+  const compaction::compaction_key& key) const {
+    if (is_tombstone) {
+        if (_removable_tombstone_ranges.contains(model::offset_cast(o))) {
             ++_stats.expired_tombstones_discarded;
             co_return false;
         }
     }
 
-    auto keep = co_await compaction::is_latest_record_for_key(_map, b, r);
-
-    co_return keep;
+    co_return co_await compaction::is_latest_record_for_key(_map, o, key);
 }
 
 ss::future<> compaction_filter::maybe_index_offset_delta(
   const model::record_batch& b,
-  const model::record& r,
+  model::record_key_metadata rec,
   chunked_vector<int32_t>& offset_deltas) const {
-    if (co_await should_keep(b, r)) {
-        offset_deltas.push_back(r.offset_delta());
+    const auto o = b.base_offset() + model::offset_delta(rec.offset_delta);
+    if (
+      co_await should_keep(
+        o, rec.is_tombstone, compaction::compaction_key{std::move(rec.key)})) {
+        offset_deltas.push_back(rec.offset_delta);
     }
 }
 
@@ -62,9 +65,9 @@ compaction_filter::compute_offset_deltas_to_keep(
     chunked_vector<int32_t> offset_deltas;
     offset_deltas.reserve(b.record_count());
 
-    co_await b.for_each_record_async(
-      [this, &b, &offset_deltas](const model::record& r) {
-          return maybe_index_offset_delta(b, r, offset_deltas);
+    co_await b.for_each_record_key_async(
+      [this, &b, &offset_deltas](model::record_key_metadata rec) {
+          return maybe_index_offset_delta(b, std::move(rec), offset_deltas);
       });
 
     co_return offset_deltas;
diff --git a/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.h b/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.h
index 0941a4b9b0ae5..638bd21f6bfe6 100644
--- a/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.h
+++ b/src/v/cloud_topics/level_one/maintenance/compaction/compaction_filter.h
@@ -12,7 +12,9 @@
 
 #include "cloud_topics/level_one/metastore/offset_interval_set.h"
 #include "compaction/filter.h"
+#include "compaction/key.h"
 #include "compaction/key_offset_map.h"
+#include "model/record_utils.h"
 
 namespace cloud_topics::l1 {
 
@@ -25,12 +27,14 @@ class compaction_filter : public compaction::filter {
       const offset_interval_set&);
 
 private:
-    ss::future<bool>
-    should_keep(const model::record_batch&, const model::record&) const;
+    ss::future<bool> should_keep(
+      model::offset,
+      bool is_tombstone,
+      const compaction::compaction_key&) const;
 
     ss::future<> maybe_index_offset_delta(
       const model::record_batch&,
-      const model::record&,
+      model::record_key_metadata,
       chunked_vector<int32_t>&) const;
 
     ss::future<chunked_vector<int32_t>>
diff --git a/src/v/cloud_topics/level_one/maintenance/compaction/compaction_source.cc b/src/v/cloud_topics/level_one/maintenance/compaction/compaction_source.cc
index 798bc8255f168..da275660ed2b3 100644
--- a/src/v/cloud_topics/level_one/maintenance/compaction/compaction_source.cc
+++ b/src/v/cloud_topics/level_one/maintenance/compaction/compaction_source.cc
@@ -10,6 +10,7 @@
 
 #include "cloud_topics/level_one/maintenance/compaction/compaction_source.h"
 
+#include "bytes/bytes.h"
 #include "cloud_io/admission_control_types.h"
 #include "cloud_topics/level_one/frontend_reader/level_one_reader.h"
 #include "cloud_topics/level_one/maintenance/compaction/compaction_filter.h"
@@ -23,6 +24,7 @@
 #include "model/batch_compression.h"
 #include "model/fundamental.h"
 #include "model/record_batch_reader.h"
+#include "model/record_utils.h"
 #include "model/timeout_clock.h"
 #include "utils/prefix_logger.h"
 
@@ -49,13 +51,14 @@ class map_building_reducer {
             b = co_await model::decompress_batch(b);
         }
 
-        co_await b.for_each_record_async(
+        co_await b.for_each_record_key_async(
           [this, base_offset = b.base_offset()](
-            const model::record& r) -> ss::future<ss::stop_iteration> {
-              if (r.is_tombstone()) {
+            model::record_key_metadata rec) -> ss::future<ss::stop_iteration> {
+              if (rec.is_tombstone) {
                   _range_has_tombstones = true;
               }
-              return maybe_index_record_in_map(r, base_offset);
+              return maybe_index_record_in_map(
+                rec.offset_delta, std::move(rec.key), base_offset);
           });
 
         if (_map_is_full) {
@@ -71,14 +74,14 @@ class map_building_reducer {
 
 private:
     ss::future<ss::stop_iteration> maybe_index_record_in_map(
-      const model::record& r, model::offset base_offset) {
-        auto offset = base_offset + model::offset_delta(r.offset_delta());
+      int32_t offset_delta, bytes key_bytes, model::offset base_offset) {
+        auto offset = base_offset + model::offset_delta(offset_delta);
 
         if (offset < _start_offset) {
             co_return ss::stop_iteration::no;
         }
 
-        auto key = compaction::compaction_key{iobuf_to_bytes(r.key())};
+        auto key = compaction::compaction_key{std::move(key_bytes)};
         bool inserted = co_await _map.put(key, offset);
 
         if (inserted) {
diff --git a/src/v/compaction/utils.cc b/src/v/compaction/utils.cc
index 2ea0878fe0535..34aa894d34984 100644
--- a/src/v/compaction/utils.cc
+++ b/src/v/compaction/utils.cc
@@ -81,12 +81,7 @@ model::record_batch make_placeholder_batch(model::record_batch_header& hdr) {
 }
 
 ss::future<bool> is_latest_record_for_key(
-  const key_offset_map& map,
-  const model::record_batch& b,
-  const model::record& r) {
-    const auto o = b.base_offset() + model::offset_delta(r.offset_delta());
-    auto key = compaction_key{iobuf_to_bytes(r.key())};
-
+  const key_offset_map& map, model::offset o, const compaction_key& key) {
     auto latest_offset_indexed = co_await map.get(key);
     // If the map hasn't indexed the given key, we should keep the
     // key.
@@ -98,6 +93,15 @@ ss::future<bool> is_latest_record_for_key(
     co_return o >= latest_offset_indexed.value();
 }
 
+ss::future<bool> is_latest_record_for_key(
+  const key_offset_map& map,
+  const model::record_batch& b,
+  const model::record& r) {
+    const auto o = b.base_offset() + model::offset_delta(r.offset_delta());
+    co_return co_await is_latest_record_for_key(
+      map, o, compaction_key{iobuf_to_bytes(r.key())});
+}
+
 bool log_needs_compaction(
   double dirty_ratio,
   double min_cleanable_dirty_ratio,
diff --git a/src/v/compaction/utils.h b/src/v/compaction/utils.h
index 5ba4cc1742dbe..2bea71d2db04b 100644
--- a/src/v/compaction/utils.h
+++ b/src/v/compaction/utils.h
@@ -65,6 +65,12 @@ ss::future<bool> is_latest_record_for_key(
   const model::record_batch& b,
   const model::record& r);
 
+// As above, but for callers that have already extracted the record's absolute
+// offset and key (e.g. from a key-only parse), avoiding full record
+// materialization.
+ss::future<bool> is_latest_record_for_key(
+  const key_offset_map& map, model::offset o, const compaction_key& key);
+
 // A log is eligible for compaction if at least one of the following
 // is true:
 // 1. It's dirty enough: the dirty ratio is at least the minimum
diff --git a/src/v/model/record.h b/src/v/model/record.h
index b2baac54dae28..5afd47689c624 100644
--- a/src/v/model/record.h
+++ b/src/v/model/record.h
@@ -985,6 +985,43 @@ class record_batch
         }
     }
 
+    /**
+     * Iterate over records parsing only their key, offset delta and
+     * tombstone flag, skipping the value and headers.
+     */
+    template<typename Func>
+    void for_each_record_key(Func f) const {
+        iobuf_const_parser parser(_records);
+        for (int32_t i = 0; i < _header.record_count; ++i) {
+            f(model::parse_record_key_from_buffer(parser));
+        }
+    }
+
+    /**
+     * Futurized `for_each_record_key`. Like for_each_record_async, the callback
+     * may return ss::stop_iteration to halt early.
+     */
+    template<typename Func>
+    ss::future<> for_each_record_key_async(Func f) const {
+        iobuf_const_parser parser(_records);
+        for (int32_t i = 0; i < _header.record_count; ++i) {
+            auto key_view = model::parse_record_key_from_buffer(parser);
+            if constexpr (
+              std::is_same_v<
+                ss::futurize_t<
+                  std::invoke_result_t<Func, model::record_key_metadata>>,
+                ss::future<ss::stop_iteration>>) {
+                ss::stop_iteration s = co_await ss::futurize_invoke(
+                  f, std::move(key_view));
+                if (s == ss::stop_iteration::yes) {
+                    co_return;
+                }
+            } else {
+                co_await ss::futurize_invoke(f, std::move(key_view));
+            }
+        }
+    }
+
     /**
      * Materialize records.
      *
diff --git a/src/v/model/record_utils.cc b/src/v/model/record_utils.cc
index 6fc86332b5388..f2c482f89f48a 100644
--- a/src/v/model/record_utils.cc
+++ b/src/v/model/record_utils.cc
@@ -260,6 +260,35 @@ void append_record_to_buffer(iobuf& a, const model::record& r) {
     }
 }
 
+model::record_key_metadata parse_record_key_from_buffer(iobuf_const_parser& p) {
+    [[maybe_unused]] auto [record_size, attr] = parse_record_meta_from_buffer(
+      p);
+    [[maybe_unused]] auto [timestamp_delta, tv] = p.read_varlong();
+    auto [offset_delta, ov] = p.read_varlong();
+    auto [key_length, kv] = p.read_varlong();
+    bytes key;
+    if (key_length > 0) {
+        key = p.read_bytes(static_cast<size_t>(key_length));
+    }
+    auto [value_length, vv] = p.read_varlong();
+    const bool is_tombstone = value_length < 0;
+    // record_size covers attributes(1) + the four varints + key + value +
+    // headers; we've consumed up to and including value_length, so the rest is
+    // the value bytes plus the (skipped) headers.
+    const int64_t header_and_key_bytes = 1 + tv + ov + kv
+                                         + std::max<int64_t>(key_length, 0)
+                                         + vv;
+    if (record_size < header_and_key_bytes) [[unlikely]] {
+        throw std::out_of_range(
+          fmt::format(
+            "Record size {} smaller than parsed header+key bytes {}",
+            record_size,
+            header_and_key_bytes));
+    }
+    p.skip(static_cast<size_t>(record_size - header_and_key_bytes));
+    return {static_cast<int32_t>(offset_delta), is_tombstone, std::move(key)};
+}
+
 model::record_metadata parse_record_metadata_from_buffer(
   iobuf_const_parser& p, bool fully_parse_record) {
     auto [record_size, attr] = parse_record_meta_from_buffer(p);
diff --git a/src/v/model/record_utils.h b/src/v/model/record_utils.h
index 3369593c5d769..5536f6f778b79 100644
--- a/src/v/model/record_utils.h
+++ b/src/v/model/record_utils.h
@@ -40,4 +40,17 @@ void append_record_to_buffer(iobuf& a, const model::record& r);
 model::record_metadata parse_record_metadata_from_buffer(
   iobuf_const_parser& p, bool fully_parse_record);
 
+/// A record's identity fields, without materializing its value or headers.
+struct record_key_metadata {
+    int32_t offset_delta;
+    // true when the record has a null value
+    bool is_tombstone;
+    // empty when the record has a null/empty key
+    bytes key;
+};
+
+/// \brief Parses a record's header and key, skipping the value and headers.
+/// Consumes exactly one record from `p`.
+model::record_key_metadata parse_record_key_from_buffer(iobuf_const_parser& p);
+
 } // namespace model
diff --git a/src/v/model/tests/BUILD b/src/v/model/tests/BUILD
index 76720a77ae18e..f574afe8cfaa9 100644
--- a/src/v/model/tests/BUILD
+++ b/src/v/model/tests/BUILD
@@ -164,6 +164,8 @@ redpanda_cc_gtest(
     cpu = 1,
     deps = [
         ":random",
+        "//src/v/bytes",
+        "//src/v/bytes:iobuf_parser",
         "//src/v/container:chunked_vector",
         "//src/v/model",
         "//src/v/model:batch_compression",
diff --git a/src/v/model/tests/record_batch_test.cc b/src/v/model/tests/record_batch_test.cc
index 7a290bb7be2a7..9afefaef21626 100644
--- a/src/v/model/tests/record_batch_test.cc
+++ b/src/v/model/tests/record_batch_test.cc
@@ -7,15 +7,22 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0
 
+#include "bytes/bytes.h"
+#include "bytes/iobuf_parser.h"
 #include "container/chunked_vector.h"
 #include "model/batch_compression.h"
 #include "model/record.h"
+#include "model/record_batch_types.h"
 #include "model/record_utils.h"
 #include "model/tests/random_batch.h"
 #include "model/timestamp.h"
 
 #include <gtest/gtest.h>
 
+#include <optional>
+#include <string_view>
+#include <vector>
+
 class RecordBatchTest : public ::testing::Test {};
 
 class RecordBatchAttributesTest
@@ -260,3 +267,155 @@ INSTANTIATE_TEST_SUITE_P(
     model::compression::snappy,
     model::compression::zstd,
     model::compression::lz4));
+
+namespace {
+
+// Serialize records into a raft_data batch (deterministic; for key-parse
+// tests).
+model::record_batch batch_from_records(std::vector<model::record> recs) {
+    iobuf body;
+    const auto count = static_cast<int32_t>(recs.size());
+    for (const auto& r : recs) {
+        model::append_record_to_buffer(body, r);
+    }
+    model::record_batch_header hdr{};
+    hdr.type = model::record_batch_type::raft_data;
+    hdr.base_offset = model::offset(0);
+    hdr.record_count = count;
+    hdr.last_offset_delta = count - 1;
+    hdr.first_timestamp = model::timestamp(1234);
+    hdr.max_timestamp = model::timestamp(1234);
+    hdr.reset_size_checksum_metadata(body);
+    return model::record_batch(
+      hdr, std::move(body), model::record_batch::tag_ctor_ng{});
+}
+
+model::record make_record(
+  int32_t offset_delta,
+  std::optional<iobuf> key,
+  std::optional<iobuf> value,
+  chunked_vector<model::record_header> headers = {}) {
+    return model::record(
+      model::record_attributes{},
+      /*timestamp_delta=*/0,
+      offset_delta,
+      std::move(key),
+      std::move(value),
+      std::move(headers));
+}
+
+} // namespace
+
+// parse_record_key_from_buffer must return the key/offset/tombstone flag and
+// advance the parser past the (skipped) value and headers, so the *next* record
+// parses from the correct position. The final bytes_left() == 0 is the check
+// that value+headers were skipped by exactly the right amount.
+TEST_F(RecordBatchTest, ParseRecordKeySkipsValueAndHeaders) {
+    std::vector<model::record> recs;
+    {
+        chunked_vector<model::record_header> h;
+        h.emplace_back(2, iobuf::from("hk"), 3, iobuf::from("hdr"));
+        recs.push_back(make_record(
+          0,
+          iobuf::from("key-0"),
+          iobuf::from("a-biggish-value"),
+          std::move(h)));
+    }
+    // Tombstone: null value, non-empty key.
+    recs.push_back(make_record(1, iobuf::from("key-1"), std::nullopt));
+    auto batch = batch_from_records(std::move(recs));
+
+    iobuf_const_parser p(batch.data());
+
+    auto k0 = model::parse_record_key_from_buffer(p);
+    EXPECT_EQ(k0.offset_delta, 0);
+    EXPECT_FALSE(k0.is_tombstone);
+    EXPECT_EQ(k0.key, iobuf_to_bytes(iobuf::from("key-0")));
+
+    auto k1 = model::parse_record_key_from_buffer(p);
+    EXPECT_EQ(k1.offset_delta, 1);
+    EXPECT_TRUE(k1.is_tombstone);
+    EXPECT_EQ(k1.key, iobuf_to_bytes(iobuf::from("key-1")));
+
+    EXPECT_EQ(p.bytes_left(), 0u);
+}
+
+// A null key yields an empty key view; is_tombstone reflects the value, not the
+// key (here: null key but a present value -> not a tombstone).
+TEST_F(RecordBatchTest, ParseRecordKeyNullKeyIsNotTombstone) {
+    std::vector<model::record> recs;
+    recs.push_back(make_record(0, std::nullopt, iobuf::from("v")));
+    auto batch = batch_from_records(std::move(recs));
+
+    iobuf_const_parser p(batch.data());
+    auto k = model::parse_record_key_from_buffer(p);
+    EXPECT_TRUE(k.key.empty());
+    EXPECT_FALSE(k.is_tombstone);
+    EXPECT_EQ(p.bytes_left(), 0u);
+}
+
+// for_each_record_key must yield exactly the same key/offset/tombstone info as
+// full materialization via for_each_record, across normal records, a tombstone,
+// a null-key record, and a record with headers.
+TEST_F(RecordBatchTest, ForEachRecordKeyMatchesFullParse) {
+    std::vector<model::record> recs;
+    recs.push_back(make_record(0, iobuf::from("k0"), iobuf::from("v0")));
+    recs.push_back(
+      make_record(1, iobuf::from("k1"), std::nullopt)); // tombstone
+    recs.push_back(make_record(2, std::nullopt, iobuf::from("v2"))); // null key
+    {
+        chunked_vector<model::record_header> h;
+        h.emplace_back(1, iobuf::from("a"), 1, iobuf::from("b"));
+        recs.push_back(
+          make_record(3, iobuf::from("k3"), iobuf::from("v3"), std::move(h)));
+    }
+    auto batch = batch_from_records(std::move(recs));
+
+    struct row {
+        int32_t offset_delta;
+        bool is_tombstone;
+        bytes key;
+        bool operator==(const row&) const = default;
+    };
+
+    // Ground truth: full record materialization.
+    std::vector<row> full;
+    batch.for_each_record([&full](model::record r) {
+        full.push_back(
+          {r.offset_delta(), r.is_tombstone(), iobuf_to_bytes(r.key())});
+    });
+
+    // Key-only iteration.
+    std::vector<row> key_only;
+    batch.for_each_record_key([&key_only](model::record_key_metadata kv) {
+        key_only.push_back(
+          {kv.offset_delta, kv.is_tombstone, std::move(kv.key)});
+    });
+
+    ASSERT_EQ(full.size(), 4u);
+    EXPECT_EQ(key_only, full);
+}
+
+// for_each_record_key_async honours ss::stop_iteration returned by the
+// callback.
+TEST_F(RecordBatchTest, ForEachRecordKeyAsyncStops) {
+    std::vector<model::record> recs;
+    for (int32_t i = 0; i < 5; ++i) {
+        recs.push_back(make_record(i, iobuf::from("k"), iobuf::from("v")));
+    }
+    auto batch = batch_from_records(std::move(recs));
+
+    std::vector<int32_t> seen;
+    batch
+      .for_each_record_key_async(
+        [&seen](
+          this auto,
+          model::record_key_metadata kv) -> ss::future<ss::stop_iteration> {
+            seen.push_back(kv.offset_delta);
+            co_return kv.offset_delta == 2 ? ss::stop_iteration::yes
+                                           : ss::stop_iteration::no;
+        })
+      .get();
+
+    EXPECT_EQ(seen, (std::vector<int32_t>{0, 1, 2}));
+}
diff --git a/src/v/storage/compaction_reducers.cc b/src/v/storage/compaction_reducers.cc
index 8b9e2f5b5dddc..f387a149cfdd0 100644
--- a/src/v/storage/compaction_reducers.cc
+++ b/src/v/storage/compaction_reducers.cc
@@ -118,11 +118,11 @@ compacted_offset_list_reducer::operator()(compacted_index::entry&& e) {
 
 ss::future<> copy_data_segment_reducer::maybe_keep_offset(
   const model::record_batch& batch,
-  const model::record& r,
+  model::record_key_metadata r,
   bool is_last_record_in_batch,
   chunked_vector<int32_t>& offset_deltas) {
     if (co_await _should_keep_fn(batch, r, is_last_record_in_batch)) {
-        offset_deltas.push_back(r.offset_delta());
+        offset_deltas.push_back(r.offset_delta);
         co_return;
     }
 }
@@ -171,11 +171,15 @@ copy_data_segment_reducer::filter(model::record_batch batch) {
     offset_deltas.reserve(batch.record_count());
 
     int32_t records_seen = 0;
-    co_await batch.for_each_record_async(
-      [this, &batch, &offset_deltas, &records_seen](const model::record& r) {
+    co_await batch.for_each_record_key_async(
+      [this, &batch, &offset_deltas, &records_seen](
+        model::record_key_metadata r) {
           ++records_seen;
           return maybe_keep_offset(
-            batch, r, batch.record_count() == records_seen, offset_deltas);
+            batch,
+            std::move(r),
+            batch.record_count() == records_seen,
+            offset_deltas);
       });
 
     if (offset_deltas.empty() && _compaction_placeholder_enabled) {
@@ -363,15 +367,15 @@ ss::future<ss::stop_iteration> copy_data_segment_reducer::filter_and_append(
         ++_stats.non_compactible_batches;
     }
     if (_compacted_idx && compactible_batch) {
-        co_await model::for_each_record(
-          batch, [&batch, this](const model::record& r) {
+        co_await batch.for_each_record_key_async(
+          [&batch, this](model::record_key_metadata r) {
               auto& hdr = batch.header();
               return _compacted_idx->index(
                 hdr.type,
                 hdr.attrs.is_control(),
-                r.key(),
+                std::move(r.key),
                 batch.base_offset(),
-                r.offset_delta());
+                r.offset_delta);
           });
     }
     using result = filtered_batch::result;
@@ -469,17 +473,14 @@ index_rebuilder_reducer::operator()(model::record_batch b) {
     co_return stop_t::no;
 }
 
-ss::future<> index_rebuilder_reducer::do_index(model::record_batch&& b) {
-    return ss::do_with(std::move(b), [this](model::record_batch& b) {
-        return model::for_each_record(
-          b,
-          [this,
-           bt = b.header().type,
-           ctrl = b.header().attrs.is_control(),
-           o = b.base_offset()](model::record& r) {
-              return _w->index(bt, ctrl, r.key(), o, r.offset_delta());
-          });
-    });
+ss::future<> index_rebuilder_reducer::do_index(model::record_batch b) {
+    co_await b.for_each_record_key_async(
+      [this,
+       bt = b.header().type,
+       ctrl = b.header().attrs.is_control(),
+       o = b.base_offset()](model::record_key_metadata r) {
+          return _w->index(bt, ctrl, std::move(r.key), o, r.offset_delta);
+      });
 }
 
 void tx_reducer::refresh_ongoing_aborted_txs(const model::record_batch& b) {
@@ -547,18 +548,17 @@ ss::future<ss::stop_iteration> tx_reducer::operator()(model::record_batch&& b) {
 }
 
 ss::future<ss::stop_iteration> map_building_reducer::maybe_index_record_in_map(
-  const model::record& r,
+  model::record_key_metadata record,
   model::offset base_offset,
   model::record_batch_type type,
   bool is_control,
   bool& fully_indexed_batch) {
-    auto offset = base_offset + model::offset_delta(r.offset_delta());
+    auto offset = base_offset + model::offset_delta(record.offset_delta);
     if (offset < _start_offset) {
         co_return ss::stop_iteration::no;
     }
 
-    auto key_view = iobuf_to_bytes(r.key());
-    auto key = enhance_key(type, is_control, key_view);
+    auto key = enhance_key(type, is_control, record.key);
     bool success = co_await _map->put(key, offset);
 
     if (success) {
@@ -583,15 +583,15 @@ map_building_reducer::operator()(model::record_batch batch) {
         batch = co_await model::decompress_batch(batch);
     }
     // is_control must be false below due to above `is_compactible()` check.
-    co_await batch.for_each_record_async(
+    co_await batch.for_each_record_key_async(
       [this,
        &fully_indexed_batch,
        base_offset = batch.base_offset(),
        type = header.type,
        is_control = false](
-        const model::record& r) -> ss::future<ss::stop_iteration> {
+        model::record_key_metadata r) -> ss::future<ss::stop_iteration> {
           return maybe_index_record_in_map(
-            r, base_offset, type, is_control, fully_indexed_batch);
+            std::move(r), base_offset, type, is_control, fully_indexed_batch);
       });
 
     if (fully_indexed_batch) {
diff --git a/src/v/storage/compaction_reducers.h b/src/v/storage/compaction_reducers.h
index bb6d76b519ecb..ce187fd7d5766 100644
--- a/src/v/storage/compaction_reducers.h
+++ b/src/v/storage/compaction_reducers.h
@@ -21,6 +21,7 @@
 #include "hashing/xx.h"
 #include "model/fundamental.h"
 #include "model/record_batch_reader.h"
+#include "model/record_utils.h"
 #include "storage/compacted_index.h"
 #include "storage/compacted_index_writer.h"
 #include "storage/compacted_offset_list.h"
@@ -125,7 +126,7 @@ class compacted_offset_list_reducer : public compaction_reducer {
 class copy_data_segment_reducer : public compaction_reducer {
 public:
     using filter_t = ss::noncopyable_function<ss::future<bool>(
-      const model::record_batch&, const model::record&, bool)>;
+      const model::record_batch&, const model::record_key_metadata&, bool)>;
 
     struct idx_and_stats {
         index_state new_idx;
@@ -190,7 +191,7 @@ class copy_data_segment_reducer : public compaction_reducer {
 
     ss::future<> maybe_keep_offset(
       const model::record_batch&,
-      const model::record&,
+      model::record_key_metadata,
       bool,
       chunked_vector<int32_t>&);
 
@@ -247,7 +248,7 @@ class index_rebuilder_reducer : public compaction_reducer {
     void end_of_stream() {}
 
 private:
-    ss::future<> do_index(model::record_batch&&);
+    ss::future<> do_index(model::record_batch);
 
     compacted_index_writer* _w;
 };
@@ -363,7 +364,7 @@ class map_building_reducer : public compaction_reducer {
 
 private:
     ss::future<ss::stop_iteration> maybe_index_record_in_map(
-      const model::record& r,
+      model::record_key_metadata record,
       model::offset base_offset,
       model::record_batch_type type,
       bool is_control,
diff --git a/src/v/storage/segment_deduplication_utils.cc b/src/v/storage/segment_deduplication_utils.cc
index bc7ca55c669db..0104aaa225dbc 100644
--- a/src/v/storage/segment_deduplication_utils.cc
+++ b/src/v/storage/segment_deduplication_utils.cc
@@ -53,11 +53,10 @@ ss::future<ss::stop_iteration> put_entry(
 ss::future<bool> is_latest_record_for_enhanced_key(
   const compaction::key_offset_map& map,
   const model::record_batch& b,
-  const model::record& r) {
-    const auto o = b.base_offset() + model::offset_delta(r.offset_delta());
-    auto key_view = compaction::compaction_key{iobuf_to_bytes(r.key())};
+  const model::record_key_metadata& r) {
+    const auto o = b.base_offset() + model::offset_delta(r.offset_delta);
     auto key = enhance_key(
-      b.header().type, b.header().attrs.is_control(), key_view);
+      b.header().type, b.header().attrs.is_control(), r.key);
 
     auto latest_offset_indexed = co_await map.get(key);
     // If the map hasn't indexed the given key, we should keep the
@@ -219,9 +218,10 @@ ss::future<index_state> deduplicate_segment(
     bool may_have_transaction_control_batches = false;
     bool may_have_transaction_data_or_fence_batches = false;
 
-    auto is_latest_record = [&map](
-                              const model::record_batch& b,
-                              const model::record& r) -> ss::future<bool> {
+    auto is_latest_record =
+      [&map](
+        const model::record_batch& b,
+        const model::record_key_metadata& r) -> ss::future<bool> {
         return is_latest_record_for_enhanced_key(map, b, r);
     };
 
@@ -238,7 +238,7 @@ ss::future<index_state> deduplicate_segment(
                           &may_have_transaction_data_or_fence_batches,
                           tx_batch_compaction_enabled](
                            const model::record_batch& b,
-                           const model::record& r,
+                           const model::record_key_metadata& r,
                            bool is_last_record_in_batch) {
         return internal::should_keep(
           b,
diff --git a/src/v/storage/segment_utils.cc b/src/v/storage/segment_utils.cc
index d548f3d499249..99c7c6d943e9e 100644
--- a/src/v/storage/segment_utils.cc
+++ b/src/v/storage/segment_utils.cc
@@ -404,8 +404,8 @@ ss::future<storage::index_state> do_copy_segment_data(
     auto offset_in_compacted_list =
       [compacted_offsets = std::move(compacted_offsets)](
         const model::record_batch& b,
-        const model::record& r) -> ss::future<bool> {
-        const auto o = b.base_offset() + model::offset_delta(r.offset_delta());
+        const model::record_key_metadata& r) -> ss::future<bool> {
+        const auto o = b.base_offset() + model::offset_delta(r.offset_delta);
         const auto keep = compacted_offsets.contains(o);
         return ss::make_ready_future<bool>(keep);
     };
@@ -423,7 +423,7 @@ ss::future<storage::index_state> do_copy_segment_data(
                           &may_have_transaction_data_or_fence_batches,
                           tx_batch_compaction_enabled](
                            const model::record_batch& b,
-                           const model::record& r,
+                           const model::record_key_metadata& r,
                            bool is_last_record_in_batch) {
         return internal::should_keep(
           b,
diff --git a/src/v/storage/segment_utils.h b/src/v/storage/segment_utils.h
index 7874eff0b4852..f0f363d2276e5 100644
--- a/src/v/storage/segment_utils.h
+++ b/src/v/storage/segment_utils.h
@@ -15,6 +15,7 @@
 #include "model/record.h"
 #include "model/record_batch_reader.h"
 #include "model/record_batch_types.h"
+#include "model/record_utils.h"
 #include "storage/compacted_index.h"
 #include "storage/compacted_index_reader.h"
 #include "storage/compacted_index_writer.h"
@@ -329,7 +330,7 @@ auto with_segment_reader_handle(segment_reader_handle handle, Func func) {
 // In all other cases, return `false`.
 inline bool can_discard(
   const model::record_batch& b,
-  const model::record& r,
+  const model::record_key_metadata& r,
   const model::ntp& ntp,
   bool past_tombstone_delete_horizon,
   bool past_tx_delete_horizon,
@@ -341,7 +342,7 @@ inline bool can_discard(
     }
 
     // Deal with tombstone record removal
-    if (r.is_tombstone() && past_tombstone_delete_horizon) {
+    if (r.is_tombstone && past_tombstone_delete_horizon) {
         return true;
     }
 
@@ -358,7 +359,7 @@ inline bool can_discard(
 template<typename Func>
 ss::future<bool> should_keep(
   const model::record_batch& b,
-  const model::record& r,
+  const model::record_key_metadata& r,
   const model::ntp& ntp,
   bool is_last_record_in_batch,
   Func&& is_latest_key,
@@ -380,7 +381,7 @@ ss::future<bool> should_keep(
          && b.header().attrs.is_transactional());
     const auto is_tx_fence_batch = b.header().type
                                    == model::record_batch_type::tx_fence;
-    const auto is_tombstone = r.is_tombstone();
+    const auto is_tombstone = r.is_tombstone;
     // once compaction placeholder feature is enabled, we are not
     // worried about empty batches as the reducer then installs a
     // placeholder batch if all the records are compacted away.
@@ -389,8 +390,10 @@ ss::future<bool> should_keep(
       && (is_last_batch && is_last_record_in_batch)) {
         vlog(
           gclog.trace,
-          "retaining last record: {} of segment from batch: {}",
-          r,
+          "retaining last record: offset_delta {} (tombstone={}) of segment "
+          "from batch: {}",
+          r.offset_delta,
+          is_tombstone,
           b.header());
         if (is_tombstone) {
             may_have_tombstone_records = true;
diff --git a/src/v/storage/tests/segment_deduplication_test.cc b/src/v/storage/tests/segment_deduplication_test.cc
index c3302867426b4..0882be291b4ae 100644
--- a/src/v/storage/tests/segment_deduplication_test.cc
+++ b/src/v/storage/tests/segment_deduplication_test.cc
@@ -653,9 +653,10 @@ TEST(CopyReducerCompressionReuseTest, ReusesCompressedPayloadWhenUnchanged) {
           return reducer.end_of_stream().reducer_stats;
       };
 
-    auto keep_all = [](const model::record_batch&, const model::record&, bool) {
-        return ss::make_ready_future<bool>(true);
-    };
+    auto keep_all =
+      [](const model::record_batch&, const model::record_key_metadata&, bool) {
+          return ss::make_ready_future<bool>(true);
+      };
 
     auto make_batch = [](bool compress) {
         auto batch = model::test::make_random_batch(
@@ -685,12 +686,14 @@ TEST(CopyReducerCompressionReuseTest, ReusesCompressedPayloadWhenUnchanged) {
         auto batch = make_batch(/*compress=*/true);
         ASSERT_TRUE(batch.compressed());
         bool drop = true;
-        auto drop_first =
-          [&](const model::record_batch&, const model::record&, bool) mutable {
-              // Keep every record except the first: the predicate returns
-              // whether to keep, so negate the one-shot drop flag.
-              return ss::make_ready_future<bool>(!std::exchange(drop, false));
-          };
+        auto drop_first = [&](
+                            const model::record_batch&,
+                            const model::record_key_metadata&,
+                            bool) mutable {
+            // Keep every record except the first: the predicate returns
+            // whether to keep, so negate the one-shot drop flag.
+            return ss::make_ready_future<bool>(!std::exchange(drop, false));
+        };
         auto stats = run_one(segs[1], std::move(batch), std::move(drop_first));
         EXPECT_EQ(stats.records_discarded, 1);
         EXPECT_EQ(stats.compressed_batches_reused, 0);