Merge pull request #30114 from vbotbuildovich/backport-pr-30108-v26.1.x-353

[v26.1.x] `ct/l1`: support timestamp lookup by L1 index in footer

Author: WillemKauf

src/v/cloud_topics/frontend/frontend.cc

@@ -518,16 +518,17 @@ ss::future<std::optional<storage::timequery_result>>
 frontend::refine_timequery_result(
   coarse_grained_timequery_result input,
   model::opt_abort_source_t abort_source) {
+    // Pass the timestamp so the L1 reader can use the footer's timestamp
+    // index to seek directly to the relevant position, avoiding unnecessary
+    // cloud IO. In the case of L0, we should only need to materialize a
+    // single batch here, because the local log is correct to the granularity of
+    // a batch (but not within a batch due to placeholders).
     cloud_topic_log_reader_config reader_cfg(
       /*start_offset=*/input.start_offset,
       /*max_offset=*/input.last_offset,
-      /*as=*/abort_source);
-    // TODO(perf): In the case of L0, we should only need to materialize a
-    // single batch here, because the local log is correct to the granularity of
-    // a batch (but not within a batch due to placeholders). For L1, we could be
-    // giving the reader a timestamp so it uses the L1 object indexes to seek
-    // to the correct spot within the index, this would allow us to optimize IO
-    // against the cloud.
+      /*first_timestamp=*/input.time,
+      /*as=*/abort_source,
+      /*client_addr=*/std::nullopt);
     auto reader = co_await make_reader(reader_cfg);
     auto generator = std::move(reader.reader).generator(model::no_timeout);
     auto query_interval = model::bounded_offset_interval::checked(

src/v/cloud_topics/level_one/common/object.cc

@@ -237,7 +237,7 @@ footer::seek_result footer::file_position_before_kafka_offset(
 }
 
 footer::seek_result footer::file_position_before_max_timestamp(
-  const model::topic_id_partition& tidp, model::timestamp target) {
+  const model::topic_id_partition& tidp, model::timestamp target) const {
     auto [begin, end] = partitions.equal_range(tidp);
     auto filtered = std::views::filter(
       std::ranges::subrange{begin, end}, [&target](const auto& entry) {
@@ -256,20 +256,20 @@ footer::seek_result footer::file_position_before_max_timestamp(
       index, target, std::less<>{}, [](const auto& entry) {
           return entry.max_timestamp;
       });
-    // If we're past all index entries, but still within the recorded file
-    // bounds, the best we can do is start at the last well known offset (the
-    // last index entry).
-    if (it == index.end()) {
-        --it;
-    }
-    // If at the first entry, we must start at the file beginning, because the
+    // If we're at the first entry, we must start at the file beginning because
     // the max is inclusive of those entries.
     if (it == index.begin()) {
         return {
           .file_position = partition.file_position,
           .length = partition.length,
         };
     }
+    // If we're past all index entries, but still within the recorded file
+    // bounds, the best we can do is start at the last well known offset (the
+    // last index entry).
+    if (it == index.end()) {
+        --it;
+    }
     auto delta = it->file_position - partition.file_position;
     return {
       .file_position = it->file_position,

src/v/cloud_topics/level_one/common/object.h

@@ -198,7 +198,7 @@ struct footer
     // While a search for timestamp 25 or 40 would yield batch[4] and the
     // timestamp 50 would yield `npos`.
     seek_result file_position_before_max_timestamp(
-      const model::topic_id_partition&, model::timestamp);
+      const model::topic_id_partition&, model::timestamp) const;
 
     // Read the footer using the suffix of an L1 object.
     //

src/v/cloud_topics/level_one/common/tests/object_test.cc

@@ -197,6 +197,36 @@ TEST(L1ObjectsIndex, TimestampSearch) {
     }
 }
 
+// Regression test: file_position_before_max_timestamp must not crash when the
+// partition's index is empty. This can happen when the partition data is
+// smaller than the indexing interval.
+TEST(L1ObjectsIndex, TimestampSearchEmptyIndex) {
+    footer index;
+    auto tidp = model::topic_id_partition{
+      model::topic_id(uuid_t::create()), model::partition_id(0)};
+    index.partitions.emplace(
+      tidp,
+      footer::partition{
+        .file_position = 0,
+        .length = 200,
+        .indexes = {},
+        .first_offset = 0_o,
+        .last_offset = 10_o,
+        .max_timestamp = 1000_t,
+      });
+
+    // Any timestamp within range should return the partition start.
+    auto result = index.file_position_before_max_timestamp(tidp, 500_t);
+    EXPECT_EQ(result, (footer::seek_result{.file_position = 0, .length = 200}));
+
+    result = index.file_position_before_max_timestamp(tidp, 1000_t);
+    EXPECT_EQ(result, (footer::seek_result{.file_position = 0, .length = 200}));
+
+    // Timestamp beyond the partition max should return npos.
+    result = index.file_position_before_max_timestamp(tidp, 1001_t);
+    EXPECT_EQ(result, footer::npos);
+}
+
 TEST(L1Objects, OffsetSearch) {
     auto test_topic_id = model::topic_id(uuid_t::create());
     auto specs_by_tidp = std::vector<batches_by_tidp>{

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

@@ -208,7 +208,8 @@ ss::future<ss::stop_iteration> compaction_source::map_building_iteration() {
         const auto& start_offset = extent.base_offset;
         const auto& max_offset = extent.last_offset;
 
-        cloud_topic_log_reader_config config(start_offset, max_offset, _as);
+        cloud_topic_log_reader_config config(
+          start_offset, max_offset, std::nullopt, _as);
         auto rdr = model::record_batch_reader(
           std::make_unique<level_one_log_reader_impl>(
             config, _ntp, _tp, _metastore, _io, _l1_reader_probe));
@@ -276,7 +277,8 @@ ss::future<ss::stop_iteration> compaction_source::deduplication_iteration(
     if (should_compact_extent(extent, _min_compaction_lag_ms)) {
         kafka::offset start_offset{extent.base_offset};
         kafka::offset last_offset{extent.last_offset};
-        cloud_topic_log_reader_config config(start_offset, last_offset, _as);
+        cloud_topic_log_reader_config config(
+          start_offset, last_offset, std::nullopt, _as);
         auto rdr = model::record_batch_reader(
           std::make_unique<level_one_log_reader_impl>(
             config, _ntp, _tp, _metastore, _io, _l1_reader_probe));

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

@@ -424,8 +424,28 @@ level_one_log_reader_impl::materialize_batches_from_object_offset(
   const object_info& object,
   kafka::offset offset,
   model::timeout_clock::time_point /*deadline*/) {
-    auto seek_res = object.footer.file_position_before_kafka_offset(
-      _tidp, offset);
+    // When a timestamp hint is available, use the footer's timestamp index
+    // to narrow the seek position. Both offset and timestamp constraints
+    // must hold, so we start at whichever position is further into the
+    // file.
+    auto seek_res = [&] {
+        auto offset_seek = object.footer.file_position_before_kafka_offset(
+          _tidp, offset);
+        if (!_config.first_timestamp) {
+            return offset_seek;
+        }
+        auto time_seek = object.footer.file_position_before_max_timestamp(
+          _tidp, *_config.first_timestamp);
+        if (time_seek == l1::footer::npos) {
+            return offset_seek;
+        }
+        if (offset_seek == l1::footer::npos) {
+            return time_seek;
+        }
+        return time_seek.file_position > offset_seek.file_position
+                 ? time_seek
+                 : offset_seek;
+    }();
     if (seek_res == l1::footer::npos) {
         // Perhaps this object spans offsets in the metastore but has
         // no data because of compaction.

src/v/cloud_topics/log_reader_config.h

@@ -30,7 +30,7 @@ struct cloud_topic_log_reader_config {
       size_t min_bytes,
       size_t max_bytes,
       std::optional<model::record_batch_type> type_filter,
-      std::optional<model::timestamp> time,
+      std::optional<model::timestamp> first_timestamp,
       model::opt_abort_source_t as,
       model::opt_client_address_t client_addr = std::nullopt,
       bool strict_max_bytes = false)
@@ -39,7 +39,7 @@ struct cloud_topic_log_reader_config {
       , min_bytes(min_bytes)
       , max_bytes(max_bytes)
       , type_filter(type_filter)
-      , first_timestamp(time)
+      , first_timestamp(first_timestamp)
       , abort_source(as)
       , client_address(std::move(client_addr))
       , strict_max_bytes(strict_max_bytes) {}
@@ -50,6 +50,7 @@ struct cloud_topic_log_reader_config {
     cloud_topic_log_reader_config(
       kafka::offset start_offset,
       kafka::offset max_offset,
+      std::optional<model::timestamp> first_timestamp = std::nullopt,
       model::opt_abort_source_t as = std::nullopt,
       model::opt_client_address_t client_addr = std::nullopt)
       : cloud_topic_log_reader_config(
@@ -58,7 +59,7 @@ struct cloud_topic_log_reader_config {
           0,
           std::numeric_limits<size_t>::max(),
           std::nullopt,
-          std::nullopt,
+          first_timestamp,
           as,
           std::move(client_addr),
           false) {}