compaction: Reduce record copying in compaction read paths

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

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;

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

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) {

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,

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

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.
      *

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);

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

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