Merge pull request #30157 from redpanda-data/l1-reader-cache

ct/l1: Remove L1 stream cache and add reader cache

Author: wdberkeley

src/v/cloud_topics/app.cc

@@ -26,6 +26,7 @@
 #include "cloud_topics/topic_manifest_upload_manager.h"
 #include "cluster/controller.h"
 #include "cluster/utils/partition_change_notifier_impl.h"
+#include "config/configuration.h"
 #include "config/node_config.h"
 #include "resource_mgmt/cpu_scheduling.h"
 #include "ssx/future-util.h"
@@ -70,6 +71,17 @@ ss::future<> app::construct(
 
     co_await construct_service(_l1_reader_probe);
 
+    co_await construct_service(
+      _l1_reader_cache,
+      ss::sharded_parameter([] {
+          return config::shard_local_cfg()
+            .cloud_topics_l1_reader_cache_eviction_timeout_ms.bind();
+      }),
+      ss::sharded_parameter([] {
+          return config::shard_local_cfg()
+            .cloud_topics_l1_reader_cache_max_size.bind();
+      }));
+
     co_await construct_service(
       l1_io,
       config::node().l1_staging_path(),
@@ -102,8 +114,6 @@ ss::future<> app::construct(
       ss::sharded_parameter([&remote] { return std::ref(remote->local()); }),
       bucket);
 
-    co_await construct_service(l1_reader_cache_);
-
     co_await construct_service(
       rr_snapshot_manager_,
       config::node().l1_staging_path(),
@@ -131,7 +141,7 @@ ss::future<> app::construct(
       ss::sharded_parameter(
         [&metadata_cache] { return &metadata_cache->local(); }),
       ss::sharded_parameter([this] { return &_l1_reader_probe.local(); }),
-      ss::sharded_parameter([this] { return &l1_reader_cache_.local(); }),
+      ss::sharded_parameter([this] { return &_l1_reader_cache.local(); }),
       ss::sharded_parameter([this] { return &rr_metadata_manager_.local(); }),
       ss::sharded_parameter([this] { return &rr_snapshot_manager_.local(); }));
 

src/v/cloud_topics/app.h

@@ -102,8 +102,8 @@ class app : public ssx::sharded_service_container {
 
     ss::sstring _logger_name;
     ss::sharded<level_one_reader_probe> _l1_reader_probe;
+    ss::sharded<l1_reader_cache> _l1_reader_cache;
     std::unique_ptr<data_plane_api> data_plane;
-    ss::sharded<l1_reader_cache> l1_reader_cache_;
     ss::sharded<state_accessors> state;
     ss::sharded<l1::file_io> l1_io;
     ss::sharded<l1::replicated_metastore> replicated_metastore;

src/v/cloud_topics/frontend/BUILD

@@ -26,6 +26,7 @@ redpanda_cc_library(
         "//src/v/cloud_storage:types",
         "//src/v/cloud_topics:logger",
         "//src/v/cloud_topics:topic_id_partition",
+        "//src/v/cloud_topics/level_one/frontend_reader:l1_reader_cache",
         "//src/v/cloud_topics/level_one/metastore",
         "//src/v/cloud_topics/level_zero/common:producer_queue",
         "//src/v/cloud_topics/level_zero/stm:ctp_stm",

src/v/cloud_topics/frontend/frontend.cc

@@ -12,6 +12,7 @@
 #include "cloud_storage/types.h"
 #include "cloud_topics/data_plane_api.h"
 #include "cloud_topics/frontend/errc.h"
+#include "cloud_topics/level_one/frontend_reader/l1_reader_cache.h"
 #include "cloud_topics/level_one/frontend_reader/level_one_reader.h"
 #include "cloud_topics/level_one/metastore/metastore.h"
 #include "cloud_topics/level_zero/common/extent_meta.h"
@@ -287,8 +288,7 @@ frontend::make_reader(cloud_topic_log_reader_config cfg) {
         if (!tidp) {
             throw topic_config_not_found_exception(ntp());
         }
-        co_return storage::translating_reader{
-          model::record_batch_reader(make_l1_reader(cfg, *tidp))};
+        co_return storage::translating_reader{make_l1_reader(cfg, *tidp)};
     }
     co_return storage::translating_reader{
       model::record_batch_reader(make_l0_reader(cfg)),
@@ -387,23 +387,28 @@ ss::future<size_t> frontend::size_bytes() {
     co_return l0_size + l1_size_res.value().size;
 }
 
-std::unique_ptr<model::record_batch_reader::impl> frontend::make_l1_reader(
+model::record_batch_reader frontend::make_l1_reader(
   const cloud_topic_log_reader_config& cfg,
   model::topic_id_partition tidp) const {
     auto ct_state = _partition->get_cloud_topics_state();
+    auto* cache = ct_state->local().get_l1_reader_cache();
+    if (cache) {
+        if (auto hit = cache->get_reader(tidp, cfg); hit) {
+            return std::move(*hit);
+        }
+    }
+
     auto l1_metastore = ct_state->local().get_l1_metastore();
     auto l1_io = ct_state->local().get_l1_io();
     auto l1_reader_probe = ct_state->local().get_l1_reader_probe();
-    auto l1_cache = ct_state->local().get_l1_reader_cache();
 
-    return std::make_unique<level_one_log_reader_impl>(
-      cfg,
-      _partition->ntp(),
-      tidp,
-      l1_metastore,
-      l1_io,
-      l1_reader_probe,
-      l1_cache);
+    auto reader = std::make_unique<level_one_log_reader_impl>(
+      cfg, _partition->ntp(), tidp, l1_metastore, l1_io, l1_reader_probe);
+
+    if (cache) {
+        return cache->put(std::move(reader));
+    }
+    return model::record_batch_reader(std::move(reader));
 }
 
 ss::future<std::optional<storage::timequery_result>>

src/v/cloud_topics/frontend/frontend.h

@@ -238,7 +238,7 @@ class frontend final {
     std::unique_ptr<model::record_batch_reader::impl>
     make_l0_reader(const cloud_topic_log_reader_config& cfg) const;
 
-    std::unique_ptr<model::record_batch_reader::impl> make_l1_reader(
+    model::record_batch_reader make_l1_reader(
       const cloud_topic_log_reader_config& cfg,
       model::topic_id_partition tidp) const;
 

src/v/cloud_topics/level_one/compaction/source.cc

@@ -282,7 +282,7 @@ ss::future<ss::stop_iteration> compaction_source::deduplication_iteration(
             config, _ntp, _tp, _metastore, _io, _l1_reader_probe));
 
         co_await sink.prepare_iteration(start_offset);
-        auto stats = co_await rdr.consume(
+        auto stats = co_await std::move(rdr).consume(
           compaction_filter{sink, *_map, _ntp, _removable_tombstone_ranges},
           model::no_timeout);
         co_await sink.finish_iteration(start_offset, last_offset);

src/v/cloud_topics/level_one/frontend_reader/BUILD

@@ -26,12 +26,13 @@ redpanda_cc_library(
     ],
     visibility = ["//visibility:public"],
     deps = [
-        "//src/v/base",
+        ":reader",
+        "//src/v/cloud_topics:log_reader_config",
         "//src/v/cloud_topics:logger",
-        "//src/v/cloud_topics/level_one/common:object",
-        "//src/v/cloud_topics/level_one/common:object_id",
+        "//src/v/config",
         "//src/v/container:intrusive",
         "//src/v/model",
+        "//src/v/random:time_jitter",
         "//src/v/ssx:future_util",
         "@seastar",
     ],
@@ -47,7 +48,6 @@ redpanda_cc_library(
     ],
     visibility = ["//visibility:public"],
     deps = [
-        ":l1_reader_cache",
         "//src/v/bytes:iostream",
         "//src/v/cloud_topics:log_reader_config",
         "//src/v/cloud_topics:logger",