lsm/io: read cloud SSTs in chunks

The cloud cache LSM persistence was previously reading SSTs whole. At
scale, this meant that when the LSM grew to include some L3 or L4
objects (O(GiBs) each), readers could be hit with minutes of latency.
Worse yet, often times the reads only need to read some few MiBs
(footers, bloom filters, indexes) to actually serve the incoming
metastore request.

This adds the new chunked file reader to the cloud_cache_persistence,
though it still attempts to read the full file from the cache if it
exists (e.g. if the object was there from previous versions of Redpanda,
or if it was written to the cache on the SST upload path).

The default chunk size is 24MiB, 1.5x the default SST size of L0, so
that by default L0 files are read whole (empirically they didn't seem
problematic being downloaded whole).

Author: andrwng

src/v/cloud_topics/level_one/domain/tests/db_domain_manager_test.cc

@@ -232,16 +232,22 @@ void flush_as_manifest(
   term_state_update_t new_terms) {
     auto domain_prefix = cloud_storage_clients::object_key{
       domain_cloud_prefix(uuid)};
-    auto cloud_db = lsm::database::open(
-                      {.database_epoch = db_epoch},
-                      lsm::io::persistence{
-                        .data = lsm::io::open_cloud_cache_data_persistence(
-                                  cache, remote, bucket, domain_prefix)
-                                  .get(),
-                        .metadata = lsm::io::open_cloud_metadata_persistence(
-                                      remote, bucket, domain_prefix)
-                                      .get()})
-                      .get();
+    auto cloud_db
+      = lsm::database::open(
+          {.database_epoch = db_epoch},
+          lsm::io::persistence{
+            .data = lsm::io::open_cloud_cache_data_persistence(
+                      cache,
+                      remote,
+                      bucket,
+                      domain_prefix,
+                      config::shard_local_cfg()
+                        .cloud_topics_metastore_sst_chunk_size.bind())
+                      .get(),
+            .metadata = lsm::io::open_cloud_metadata_persistence(
+                          remote, bucket, domain_prefix)
+                          .get()})
+          .get();
 
     // Pre-register the object IDs before building the add_objects rows.
     preregister_objects_db_update prereg_update;

src/v/cloud_topics/level_one/metastore/lsm/replicated_db.cc

@@ -115,7 +115,12 @@ replicated_database::open(
 
     auto data_persist_fut = co_await ss::coroutine::as_future(
       lsm::io::open_cloud_cache_data_persistence(
-        cache, remote, bucket, domain_prefix, cloud_io::group_id::metastore));
+        cache,
+        remote,
+        bucket,
+        domain_prefix,
+        config::shard_local_cfg().cloud_topics_metastore_sst_chunk_size.bind(),
+        cloud_io::group_id::metastore));
     if (data_persist_fut.failed()) {
         co_return std::unexpected(wrap_failed_future(
           data_persist_fut.get_exception(), "Failed to open data persistence"));

src/v/cloud_topics/level_one/metastore/lsm/tests/replicated_db_test.cc

@@ -252,7 +252,9 @@ class ReplicatedDatabaseTest
                           &test_cache.local(),
                           &sr->remote.local(),
                           bucket_name,
-                          domain_prefix)
+                          domain_prefix,
+                          config::shard_local_cfg()
+                            .cloud_topics_metastore_sst_chunk_size.bind())
                           .get(),
                 .metadata = lsm::io::open_cloud_metadata_persistence(
                               &sr->remote.local(), bucket_name, domain_prefix)

src/v/cloud_topics/read_replica/snapshot_manager.cc

@@ -138,7 +138,12 @@ ss::future<> database_refresher::open_or_refresh() {
     cloud_storage_clients::object_key domain_prefix{
       domain_cloud_prefix(domain_uuid_)};
     auto data_persist = co_await lsm::io::open_cloud_cache_data_persistence(
-      cache_, remote_, bucket_, domain_prefix, cloud_io::group_id::metastore);
+      cache_,
+      remote_,
+      bucket_,
+      domain_prefix,
+      config::shard_local_cfg().cloud_topics_metastore_sst_chunk_size.bind(),
+      cloud_io::group_id::metastore);
     auto meta_persist = co_await lsm::io::open_cloud_metadata_persistence(
       remote_, bucket_, domain_prefix, cloud_io::group_id::metastore);
     lsm::io::persistence io{

src/v/cloud_topics/read_replica/tests/db_utils.h

@@ -17,6 +17,7 @@
 #include "cloud_topics/level_one/metastore/lsm/state_reader.h"
 #include "cloud_topics/level_one/metastore/lsm/state_update.h"
 #include "cloud_topics/level_one/metastore/lsm/write_batch_row.h"
+#include "config/configuration.h"
 #include "container/chunked_hash_map.h"
 #include "lsm/io/cloud_cache_persistence.h"
 #include "lsm/lsm.h"
@@ -49,7 +50,11 @@ inline ss::future<lsm::database*> get_or_create_writer_db(
     cloud_storage_clients::object_key domain_prefix{cloud_prefix};
 
     auto data_persist = co_await lsm::io::open_cloud_cache_data_persistence(
-      cache, remote, bucket, domain_prefix);
+      cache,
+      remote,
+      bucket,
+      domain_prefix,
+      config::shard_local_cfg().cloud_topics_metastore_sst_chunk_size.bind());
     auto meta_persist = co_await lsm::io::open_cloud_metadata_persistence(
       remote, bucket, domain_prefix);
 

src/v/lsm/io/BUILD

@@ -52,9 +52,9 @@ redpanda_cc_library(
     srcs = ["cloud_cache_persistence.cc"],
     hdrs = ["cloud_cache_persistence.h"],
     implementation_deps = [
+        ":chunked_remote_file_reader",
         ":file_io",
         "//src/v/cloud_io:io_result",
-        "//src/v/config",
         "//src/v/lsm/core:exceptions",
         "//src/v/lsm/core/internal:files",
         "//src/v/ssx:future_util",
@@ -66,6 +66,7 @@ redpanda_cc_library(
         "//src/v/cloud_io:cache",
         "//src/v/cloud_io:remote",
         "//src/v/cloud_storage_clients",
+        "//src/v/config",
         "@seastar",
     ],
 )

src/v/lsm/io/cloud_cache_persistence.cc

@@ -15,6 +15,7 @@
 #include "config/configuration.h"
 #include "lsm/core/exceptions.h"
 #include "lsm/core/internal/files.h"
+#include "lsm/io/chunked_remote_file_reader.h"
 #include "lsm/io/file_io.h"
 #include "ssx/future-util.h"
 #include "utils/retry_chain_node.h"
@@ -33,11 +34,14 @@ namespace {
 
 static constexpr auto reservation_timeout = std::chrono::seconds(30);
 
+static constexpr auto cloud_op_timeout = std::chrono::seconds(10);
+
+std::chrono::milliseconds cloud_op_backoff() {
+    return config::shard_local_cfg().cloud_storage_initial_backoff_ms.value();
+}
+
 retry_chain_node make_cloud_rtc(ss::abort_source& as) {
-    constexpr auto timeout = std::chrono::seconds(10);
-    auto backoff
-      = config::shard_local_cfg().cloud_storage_initial_backoff_ms.value();
-    return retry_chain_node{as, timeout, backoff};
+    return retry_chain_node{as, cloud_op_timeout, cloud_op_backoff()};
 }
 
 bool check_cloud_result(cloud_io::download_result result) {
@@ -239,53 +243,50 @@ class cloud_cache_data_persistence : public data_persistence {
       cloud_io::remote* remote,
       cloud_storage_clients::bucket_name bucket,
       cloud_storage_clients::object_key prefix,
+      config::binding<size_t> chunk_size,
       cloud_io::group_id gid)
       : _cache(cache)
       , _remote(remote)
       , _bucket(std::move(bucket))
       , _prefix(std::move(prefix))
+      , _chunk_size(std::move(chunk_size))
       , _gid(gid) {}
 
     ss::future<optional_pointer<random_access_file_reader>>
     open_random_access_reader(
-      internal::file_handle h, uint64_t /*file_size*/) override {
+      internal::file_handle h, uint64_t file_size) override {
         _as.check();
         auto _ = _gate.hold();
         auto filename = internal::sst_file_name(h);
-        auto key = cache_key(filename);
 
-        auto root = make_cloud_rtc(_as);
-        while (true) {
-            auto reader = co_await open_cached_reader(key);
-            if (reader) {
-                co_return reader;
-            }
+        auto cached = co_await open_cached_reader(cache_key(filename));
+        if (cached) {
+            // The whole object is already local -- e.g. just written through
+            // by the cache-staged writer. Serve it directly rather than
+            // re-fetching it from the cloud in chunks.
+            co_return cached;
+        }
 
-            auto dl_fut = co_await ss::coroutine::as_future(
-              _remote->download_stream(
-                {
-                  .bucket = _bucket,
-                  .key = cloud_key(filename),
-                  .parent_rtc = root,
-                },
-                [this,
-                 &key](uint64_t content_length, ss::input_stream<char> stream) {
-                    return save_to_cache(
-                      content_length, std::move(stream), key);
-                },
-                "SST file download",
-                /*acquire_hydration_units=*/true,
-                /*byte_range=*/std::nullopt,
-                /*throttle_metric_ms_cb=*/{},
-                _gid));
-            if (dl_fut.failed()) {
-                throw_as_lsm_ex(
-                  dl_fut.get_exception(), "error downloading file");
-            }
-            if (!check_cloud_result(dl_fut.get())) {
-                co_return std::nullopt;
-            }
+        // Hydrate chunks on demand. Chunk entries live in a sibling
+        // subdirectory so they can never collide with the whole-file key
+        // probed above.
+        auto reader = co_await chunked_remote_file_reader::open(
+          _cache,
+          _remote,
+          _bucket,
+          cloud_key(filename),
+          chunk_cache_prefix(filename),
+          file_size,
+          _chunk_size(),
+          config::shard_local_cfg().cloud_storage_hydration_timeout_ms(),
+          cloud_op_backoff(),
+          _as,
+          _gid);
+        if (!reader) {
+            co_return std::nullopt;
         }
+        std::unique_ptr<random_access_file_reader> ptr = std::move(*reader);
+        co_return ptr;
     }
 
     ss::future<std::unique_ptr<sequential_file_writer>>
@@ -403,9 +404,9 @@ class cloud_cache_data_persistence : public data_persistence {
                 co_return std::nullopt;
             }
             auto local_path = _cache->get_local_path(key);
-            std::unique_ptr<random_access_file_reader> ptr;
-            ptr = std::make_unique<disk_file_reader>(
-              std::move(local_path), std::move(item->body));
+            std::unique_ptr<random_access_file_reader> ptr
+              = std::make_unique<disk_file_reader>(
+                std::move(local_path), std::move(item->body));
             co_return ptr;
         } catch (const std::system_error& e) {
             if (e.code() == std::errc::no_such_file_or_directory) {
@@ -423,29 +424,16 @@ class cloud_cache_data_persistence : public data_persistence {
         }
     }
 
-    ss::future<uint64_t> save_to_cache(
-      uint64_t content_length,
-      ss::input_stream<char> input_stream,
-      const std::filesystem::path& key) {
-        std::exception_ptr ex;
-        try {
-            auto reservation = co_await _cache->reserve_space(
-              content_length, 1);
-            co_await _cache->put(key, input_stream, reservation);
-        } catch (...) {
-            ex = std::current_exception();
-        }
-        co_await input_stream.close();
-        if (ex) {
-            std::rethrow_exception(ex);
-        }
-        co_return content_length;
-    }
-
     std::filesystem::path cache_key(std::string_view name) {
         return std::filesystem::path("lsm") / _prefix() / name;
     }
 
+    // Where the chunked reader caches its chunks for `name`: a sibling of the
+    // whole-file key so the two never share a path.
+    std::filesystem::path chunk_cache_prefix(std::string_view name) {
+        return cache_key(fmt::format("{}.chunks", name));
+    }
+
     cloud_storage_clients::object_key cloud_key(std::string_view name) {
         return join_cloud_key(_prefix, name);
     }
@@ -454,6 +442,7 @@ class cloud_cache_data_persistence : public data_persistence {
     cloud_io::remote* _remote;
     cloud_storage_clients::bucket_name _bucket;
     cloud_storage_clients::object_key _prefix;
+    config::binding<size_t> _chunk_size;
     cloud_io::group_id _gid;
     ss::abort_source _as;
     ss::gate _gate;
@@ -600,9 +589,15 @@ ss::future<std::unique_ptr<data_persistence>> open_cloud_cache_data_persistence(
   cloud_io::remote* remote,
   cloud_storage_clients::bucket_name bucket,
   cloud_storage_clients::object_key prefix,
+  config::binding<size_t> sst_chunk_size,
   cloud_io::group_id gid) {
     co_return std::make_unique<cloud_cache_data_persistence>(
-      cache, remote, std::move(bucket), std::move(prefix), gid);
+      cache,
+      remote,
+      std::move(bucket),
+      std::move(prefix),
+      std::move(sst_chunk_size),
+      gid);
 }
 
 ss::future<std::unique_ptr<metadata_persistence>>

src/v/lsm/io/cloud_cache_persistence.h

@@ -13,18 +13,22 @@
 #include "cloud_io/cache_service.h"
 #include "cloud_io/remote.h"
 #include "cloud_storage_clients/types.h"
+#include "config/property.h"
 #include "lsm/io/persistence.h"
 
 namespace lsm::io {
 
 /// Open a data persistence backed by the cloud cache and cloud storage.
-/// `gid` tags every cloud operation this persistence issues for the
-/// cloud_io scheduler; defaults to default_group.
+/// SST files are read from object storage in chunks of `sst_chunk_size`
+/// bytes, hydrated into the cache on demand; the binding is read per open so
+/// the size can change at runtime. `gid` tags every cloud operation this
+/// persistence issues for the cloud_io scheduler; defaults to default_group.
 ss::future<std::unique_ptr<data_persistence>> open_cloud_cache_data_persistence(
   cloud_io::cache* cache,
   cloud_io::remote* remote,
   cloud_storage_clients::bucket_name bucket,
   cloud_storage_clients::object_key prefix,
+  config::binding<size_t> sst_chunk_size,
   cloud_io::group_id gid = cloud_io::group_id::default_group);
 
 /// Open a metadata persistence backed by cloud storage. `gid` tags every

src/v/lsm/io/tests/persistence_test.cc

@@ -68,7 +68,8 @@ TEST_P(PersistenceTest, CanWriteAndReadAFile) {
         w->append(iobuf::from("world")).get();
     }
     {
-        auto maybe_r = persistence->open_random_access_reader({}, 0).get();
+        auto maybe_r
+          = persistence->open_random_access_reader({}, /*file_size=*/10).get();
         ASSERT_TRUE(bool(maybe_r));
         auto r = std::move(*maybe_r);
         auto _ = ss::defer([&r] { r->close().get(); });
@@ -114,7 +115,8 @@ TEST_P(PersistenceTest, DuplicateWritePreservesOriginal) {
     } catch (...) {
         // Expected for backends that use O_EXCL.
     }
-    auto maybe_r = persistence->open_random_access_reader({}, 0).get();
+    auto maybe_r
+      = persistence->open_random_access_reader({}, /*file_size=*/13).get();
     ASSERT_TRUE(bool(maybe_r));
     auto r = std::move(*maybe_r);
     auto _ = ss::defer([&r] { r->close().get(); });
@@ -124,7 +126,10 @@ TEST_P(PersistenceTest, DuplicateWritePreservesOriginal) {
 }
 
 TEST_P(PersistenceTest, ReadNonExisting) {
-    auto maybe_r = persistence->open_random_access_reader({}, 0).get();
+    // A nonzero size is required by backends that probe the object on open;
+    // the file does not exist, so the result is still nullopt.
+    auto maybe_r
+      = persistence->open_random_access_reader({}, /*file_size=*/16).get();
     EXPECT_FALSE(bool(maybe_r));
 }
 
@@ -156,7 +161,7 @@ TEST_P(PersistenceTest, RandomAccessReaderComprehensive) {
     }
 
     // Open reader for all tests
-    auto maybe_r = persistence->open_random_access_reader({}, 0).get();
+    auto maybe_r = persistence->open_random_access_reader({}, file_size).get();
     ASSERT_TRUE(bool(maybe_r));
     auto r = std::move(*maybe_r);
 
@@ -359,7 +364,8 @@ INSTANTIATE_TEST_SUITE_P(
           &cache->local(),
           &sr->remote.local(),
           fixture->bucket_name,
-          cloud_storage_clients::object_key("test-prefix"));
+          cloud_storage_clients::object_key("test-prefix"),
+          config::mock_binding<size_t>(1_MiB));
 
         co_return std::make_unique<mock_cloud_cache_data_persistence>(
           std::move(fixture),