Merge pull request #29196 from dotnwat/ct-l0-reader

ct: a few simplifications in level zero reader

Author: dotnwat

src/v/cloud_topics/batch_cache/batch_cache.cc

@@ -47,6 +47,10 @@ ss::future<> batch_cache::stop() {
 }
 
 void batch_cache::put(const model::ntp& ntp, const model::record_batch& b) {
+    vassert(
+      b.term() > model::term_id{-1},
+      "Batch without term in the cache: {}",
+      b.header());
     if (_lm == nullptr) {
         return;
     }
@@ -81,6 +85,17 @@ batch_cache::get(const model::ntp& ntp, model::offset o) {
     if (auto it = _index.find(ntp); it != _index.end()) {
         auto rb = it->second->get(o);
         if (rb.has_value()) {
+            vassert(
+              rb->term() > model::term_id{-1},
+              "Batch without term in the cache: {}",
+              rb->header());
+            vassert(
+              rb->base_offset() <= o && o <= rb->last_offset(),
+              "Unexpected batch for {}, got range: [{},{}] for offset {}",
+              ntp,
+              rb->base_offset(),
+              rb->last_offset(),
+              o);
             _probe.register_get(rb->size_bytes());
         } else {
             _probe.register_miss();

src/v/cloud_topics/level_one/frontend_reader/level_one_reader.cc

@@ -257,7 +257,7 @@ level_one_log_reader_impl::read_batches(l1::object_reader& reader) {
                 set_end_of_stream();
                 break;
             }
-            _config.bytes_consumed += batch_size;
+            _bytes_consumed += batch_size;
 
             batches.push_back(std::move(batch));
 
@@ -355,8 +355,8 @@ bool level_one_log_reader_impl::is_end_of_stream() const {
 }
 
 bool level_one_log_reader_impl::is_over_limit_with_bytes(size_t size) const {
-    return (_config.strict_max_bytes || _config.bytes_consumed > 0)
-           && (_config.bytes_consumed + size) > _config.max_bytes;
+    return (_config.strict_max_bytes || _bytes_consumed > 0)
+           && (_bytes_consumed + size) > _config.max_bytes;
 }
 
 } // namespace cloud_topics

src/v/cloud_topics/level_one/frontend_reader/level_one_reader.h

@@ -135,6 +135,7 @@ class level_one_log_reader_impl : public model::record_batch_reader::impl {
     l1::metastore* _metastore;
     l1::io* _io;
     prefix_logger _log;
+    size_t _bytes_consumed{0};
 };
 
 } // namespace cloud_topics

src/v/cloud_topics/level_zero/frontend_reader/level_zero_reader.cc

@@ -32,6 +32,7 @@ level_zero_log_reader_impl::level_zero_log_reader_impl(
   ss::lw_shared_ptr<cluster::partition> ctp,
   data_plane_api* ct_api)
   : _config(cfg)
+  , _next_offset(_config.start_offset)
   , _ctp(std::move(ctp))
   , _ct_api(ct_api)
   , _log(cd_log, fmt::format("[{}/{}]", fmt::ptr(this), _ctp->ntp())) {
@@ -52,10 +53,8 @@ level_zero_log_reader_impl::do_load_slice(
     // the 'empty' state. It doesn't make any difference if the reader is in
     // the 'materialized' state. If we're in 'ready' state we risk to go out
     // of sync with cached metadata so it's safer to hydrate.
-    if (cache_enabled()) {
-        if (auto cached = maybe_load_slices_from_cache()) {
-            co_return std::move(cached.value());
-        }
+    if (auto cached = maybe_read_batches_from_cache(); !cached.empty()) {
+        co_return cached;
     }
 
     chunked_circular_buffer<model::record_batch> res;
@@ -88,63 +87,54 @@ level_zero_log_reader_impl::do_load_slice(
     co_return res;
 }
 
-std::optional<chunked_circular_buffer<model::record_batch>>
-level_zero_log_reader_impl::maybe_load_slices_from_cache() {
+chunked_circular_buffer<model::record_batch>
+level_zero_log_reader_impl::maybe_read_batches_from_cache() {
     chunked_circular_buffer<model::record_batch> ret;
-    auto current = _config.start_offset;
-    while (current <= _config.max_offset) {
+    if (!cache_enabled()) {
+        return ret;
+    }
+
+    /*
+     * Fetch batches from the cache starting at `_next_offset` until we hit a
+     * gap or a control batch and must then fetch the data from object storage.
+     */
+    while (_next_offset <= _config.max_offset) {
         auto batch = _ct_api->cache_get(
-          _ctp->ntp(), kafka::offset_cast(current));
+          _ctp->ntp(), kafka::offset_cast(_next_offset));
         if (!batch.has_value()) {
-            // We hit a gap in the cache and have to download objects
-            // from S3.
-            //
-            // NOTE: this can also happen when transactions are used and
-            // we would encounter reading a control batch from the local log.
             break;
         }
+
         vlog(
           _log.trace,
           "Loaded batch from cache for {}: {} @ term {}",
-          current,
+          _next_offset,
           batch.value().base_offset(),
           batch.value().term());
-        vassert(
-          batch.value().term() > model::term_id{-1},
-          "Batch without term in the cache: {}",
-          batch.value().header());
+
         auto batch_size = batch.value().size_bytes();
         if (is_over_limit(batch_size)) {
             break;
         }
-        vassert(
-          batch->base_offset() <= kafka::offset_cast(current)
-            && kafka::offset_cast(current) <= batch->last_offset(),
-          "Unexpected batch for {}, got range: [{},{}] for offset {}",
-          _ctp->ntp(),
-          batch->base_offset(),
-          batch->last_offset(),
-          current);
+
         ret.push_back(std::move(batch.value()));
-        _config.bytes_consumed += batch_size;
-        current = model::offset_cast(
+        _bytes_consumed += batch_size;
+        _next_offset = model::offset_cast(
           model::next_offset(ret.back().last_offset()));
     }
-    _config.start_offset = current;
-    if (_config.start_offset > _config.max_offset) {
+
+    if (_next_offset > _config.max_offset) {
         vlog(
           _log.debug,
           "reached end of stream, start offset: {}, max offset: {}, "
-          "current: {}",
+          "next offset: {}",
           _config.start_offset,
           _config.max_offset,
-          current);
+          _next_offset);
         _current = state::end_of_stream_state;
     }
-    if (!ret.empty()) {
-        return ret;
-    }
-    return std::nullopt;
+
+    return ret;
 }
 
 storage::local_log_reader_config level_zero_log_reader_impl::ctp_read_config() {
@@ -155,7 +145,7 @@ storage::local_log_reader_config level_zero_log_reader_impl::ctp_read_config() {
      */
     auto ot_state = _ctp->get_offset_translator_state();
     auto start_offset = ot_state->to_log_offset(
-      kafka::offset_cast(_config.start_offset));
+      kafka::offset_cast(_next_offset));
     auto max_offset = ot_state->to_log_offset(
       kafka::offset_cast(_config.max_offset));
 
@@ -324,7 +314,7 @@ ss::future<> level_zero_log_reader_impl::materialize_batches(
                   hydrated_batch_size);
                 break;
             }
-            _config.bytes_consumed += hydrated_batch_size;
+            _bytes_consumed += hydrated_batch_size;
             if (
               auto* meta = std::get_if<cloud_topics::extent_meta>(
                 &unhydrated_it->data)) {
@@ -453,7 +443,7 @@ void level_zero_log_reader_impl::consume_materialized_batches(
       _hydrated.size(),
       _unhydrated.size());
     *dest = std::exchange(_hydrated, {});
-    _config.start_offset = model::offset_cast(
+    _next_offset = model::offset_cast(
       model::next_offset(dest->back().last_offset()));
     _current = _unhydrated.empty() ? state::empty_state : state::ready_state;
 }
@@ -467,7 +457,7 @@ bool level_zero_log_reader_impl::is_end_of_stream() const {
 }
 
 bool level_zero_log_reader_impl::is_over_limit(size_t size) const {
-    return (_config.strict_max_bytes || _config.bytes_consumed > 0)
-           && (_config.bytes_consumed + size) > _config.max_bytes;
+    return (_config.strict_max_bytes || _bytes_consumed > 0)
+           && (_bytes_consumed + size) > _config.max_bytes;
 }
 } // namespace cloud_topics

src/v/cloud_topics/level_zero/frontend_reader/level_zero_reader.h

@@ -101,8 +101,8 @@ class level_zero_log_reader_impl : public model::record_batch_reader::impl {
     // Return data from the record batch cache.
     // This method could change state of the reader to end_of_stream_state
     // when it reaches committed offset.
-    std::optional<chunked_circular_buffer<model::record_batch>>
-    maybe_load_slices_from_cache();
+    chunked_circular_buffer<model::record_batch>
+    maybe_read_batches_from_cache();
 
     // If adding a batch of `size` would cause this to go over the bytes limit.
     bool is_over_limit(size_t size) const;
@@ -136,9 +136,11 @@ class level_zero_log_reader_impl : public model::record_batch_reader::impl {
     chunked_circular_buffer<model::record_batch> _hydrated;
 
     cloud_topic_log_reader_config _config;
+    kafka::offset _next_offset;
     ss::lw_shared_ptr<cluster::partition> _ctp;
     data_plane_api* _ct_api;
     prefix_logger _log;
+    size_t _bytes_consumed{0};
 };
 
 } // namespace cloud_topics

src/v/cloud_topics/log_reader_config.cc

@@ -20,7 +20,7 @@ fmt::iterator cloud_topic_log_reader_config::format_to(fmt::iterator it) const {
       it,
       "start_offset:{}, max_offset:{}, min_bytes:{}, max_bytes:{}, "
       "strict_max_bytes:{}, type_filter: {}, first_timestamp:{}, "
-      "bytes_consumed:{}, skip_cache:{}, abortable:{}, "
+      "skip_cache:{}, abortable:{}, "
       "client_address:{}",
       start_offset,
       max_offset,
@@ -29,7 +29,6 @@ fmt::iterator cloud_topic_log_reader_config::format_to(fmt::iterator it) const {
       strict_max_bytes,
       type_filter,
       first_timestamp,
-      bytes_consumed,
       skip_cache,
       abort_source.has_value(),
       client_address);

src/v/cloud_topics/log_reader_config.h

@@ -80,9 +80,6 @@ struct cloud_topic_log_reader_config {
 
     model::opt_client_address_t client_address;
 
-    // used by log reader
-    size_t bytes_consumed{0};
-
     // do not let the lower level readers go over budget even when that means
     // that the reader will return no batches.
     bool strict_max_bytes{false};