diff --git a/proto/redpanda/core/admin/internal/cloud_topics/v1/metastore.proto b/proto/redpanda/core/admin/internal/cloud_topics/v1/metastore.proto index 103c0298a2aed..b964e63a76e05 100644 --- a/proto/redpanda/core/admin/internal/cloud_topics/v1/metastore.proto +++ b/proto/redpanda/core/admin/internal/cloud_topics/v1/metastore.proto @@ -237,6 +237,34 @@ message MetadataValue { int64 compaction_epoch = 3; uint64 size = 4; uint64 num_extents = 5; + // True while the partition is mid tiered->cloud migration. + bool migrating = 6; +} + +// Where the backing tiered-storage segment of an imported object lives. An +// object property: the imported analog of a native object's id->bucket path. +message ImportedTsObjectLocation { + string ts_path = 1; +} + +// Whether an imported tiered-storage segment has an aborted-transaction (.tx) +// manifest, resolved at import time so the read path need not probe object +// storage. UNKNOWN (v1/v2 non-compacted) => probe and tolerate a missing .tx. +enum ImportedTsTxManifestState { + IMPORTED_TS_TX_MANIFEST_STATE_UNKNOWN = 0; + IMPORTED_TS_TX_MANIFEST_STATE_ABSENT = 1; + IMPORTED_TS_TX_MANIFEST_STATE_PRESENT = 2; +} + +// Properties of the data in an imported tiered-storage extent. An extent +// property, alongside the extent's Kafka offset bounds (base/last_offset). +message ImportedTsSegmentInfo { + int64 segment_term = 1; + // Offset-translation delta at the segment's base (source segment_meta's + // delta_offset). + int64 delta_base = 2; + // Whether the segment's .tx manifest exists (resolved at import time). + ImportedTsTxManifestState tx_state = 3; } // Value for an extent row. @@ -246,6 +274,7 @@ message ExtentValue { uint64 filepos = 3; uint64 len = 4; string object_id = 5; + optional ImportedTsSegmentInfo imported_ts_info = 6; } // Value for a term row. @@ -280,6 +309,7 @@ message ObjectValue { uint64 object_size = 4; int64 last_updated = 5; bool is_preregistration = 6; + optional ImportedTsObjectLocation imported_ts_location = 7; } // A typed value for a metastore row. diff --git a/src/v/cloud_storage/BUILD b/src/v/cloud_storage/BUILD index e5b1a1aebc49f..91e97d8e8afb6 100644 --- a/src/v/cloud_storage/BUILD +++ b/src/v/cloud_storage/BUILD @@ -86,6 +86,32 @@ redpanda_cc_library( ], ) +redpanda_cc_library( + name = "offset_index", + srcs = [ + "offset_index.cc", + ], + hdrs = [ + "offset_index.h", + ], + implementation_deps = [ + ":logger", + "//src/v/base", + "//src/v/bytes:iobuf_parser", + "//src/v/serde", + "//src/v/serde:iobuf", + "//src/v/serde:vector", + "@seastar", + ], + visibility = ["//visibility:public"], + deps = [ + "//src/v/bytes:iobuf", + "//src/v/model", + "//src/v/utils:delta_for", + "@abseil-cpp//absl/container:btree", + ], +) + redpanda_cc_library( name = "cloud_storage", srcs = [ @@ -183,6 +209,7 @@ redpanda_cc_library( visibility = ["//visibility:public"], deps = [ ":logger", + ":offset_index", ":remote_label", ":segment_meta_cstore", ":topic_mount_manifest_path", diff --git a/src/v/cloud_storage/offset_index.cc b/src/v/cloud_storage/offset_index.cc new file mode 100644 index 0000000000000..f72a8ea0dab56 --- /dev/null +++ b/src/v/cloud_storage/offset_index.cc @@ -0,0 +1,435 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/vectorizedio/redpanda/blob/master/licenses/rcl.md + */ + +#include "cloud_storage/offset_index.h" + +#include "base/vassert.h" +#include "bytes/iobuf_parser.h" +#include "cloud_storage/logger.h" +#include "serde/rw/envelope.h" +#include "serde/rw/iobuf.h" +#include "serde/rw/vector.h" + +#include + +namespace cloud_storage { + +offset_index::offset_index( + model::offset initial_rp, + kafka::offset initial_kaf, + int64_t initial_file_pos, + int64_t file_pos_step, + model::timestamp initial_time) + : _rp_offsets{} + , _kaf_offsets{} + , _file_offsets{} + , _time_offsets{} + , _pos{} + , _initial_rp(initial_rp) + , _initial_kaf(initial_kaf) + , _initial_file_pos(initial_file_pos) + , _initial_time(initial_time) + , _rp_index(initial_rp) + , _kaf_index(initial_kaf) + , _file_index(initial_file_pos, delta_delta_t(file_pos_step)) + , _time_index(initial_time.value()) + , _min_file_pos_step(file_pos_step) {} + +size_t offset_index::estimate_memory_use() const { + return _file_index.mem_use() + _rp_index.mem_use() + _kaf_index.mem_use() + + _time_index.mem_use(); +} + +void offset_index::add( + model::offset rp_offset, + kafka::offset kaf_offset, + int64_t file_offset, + model::timestamp max_timestamp) { + auto ix = index_mask & _pos++; + _rp_offsets.at(ix) = rp_offset(); + _kaf_offsets.at(ix) = kaf_offset(); + _file_offsets.at(ix) = file_offset; + _time_offsets.at(ix) = max_timestamp.value(); + try { + if ((_pos & index_mask) == 0) { + _rp_index.add(_rp_offsets); + _kaf_index.add(_kaf_offsets); + _file_index.add(_file_offsets); + _time_index.add(_time_offsets); + } + } catch (...) { + // Get rid of the corrupted state in the encoders. + // If the exception is thrown out of 'add' method + // the invariant of the index is broken. + _pos = 0; + _rp_offsets = {}; + _kaf_offsets = {}; + _file_offsets = {}; + _rp_index = encoder_t(_initial_rp); + _kaf_index = encoder_t(_initial_kaf); + _file_index = foffset_encoder_t( + _initial_file_pos, delta_delta_t(_min_file_pos_step)); + _time_index = encoder_t(_initial_time.value()); + throw; + } +} + +std:: + variant + offset_index::maybe_find_offset( + int64_t upper_bound, + deltafor_encoder& encoder, + const std::array& write_buffer) { + deltafor_decoder decoder( + encoder.get_initial_value(), encoder.get_row_count(), encoder.share()); + auto max_index = encoder.get_row_count() * details::FOR_buffer_depth - 1; + auto maybe_ix = _find_under(std::move(decoder), upper_bound); + if (!maybe_ix || maybe_ix->ix == max_index) { + auto ixend = _pos & index_mask; + std::optional candidate; + for (size_t i = 0; i < ixend; i++) { + if (write_buffer.at(i) < upper_bound) { + candidate = find_result{ + .rp_offset = model::offset(_rp_offsets.at(i)), + .kaf_offset = kafka::offset(_kaf_offsets.at(i)), + .file_pos = _file_offsets.at(i), + }; + } else { + break; + } + } + // maybe_ix can point to the last element of the compressed + // chunk if all elements inside it are less than the offset that + // we're looking for. In this case we should use it even if + // we can't find anything inside the buffer. + // maybe_ix will be null if the compressed chunk is empty. + if (candidate) { + return *candidate; + } + if (!maybe_ix) { + return std::monostate(); + } + } + // Invariant: maybe_ix here can't be nullopt + return *maybe_ix; +} + +std::optional +offset_index::find_rp_offset(model::offset upper_bound) { + auto search_result = maybe_find_offset(upper_bound, _rp_index, _rp_offsets); + + return ss::visit( + search_result, + [](std::monostate) -> std::optional { return std::nullopt; }, + [](find_result result) -> std::optional { return result; }, + [this](index_value index_result) -> std::optional { + find_result res{}; + + size_t ix = index_result.ix; + res.rp_offset = model::offset(index_result.value); + + decoder_t kaf_dec( + _kaf_index.get_initial_value(), + _kaf_index.get_row_count(), + _kaf_index.copy()); + auto kaf_offset = _fetch_ix(std::move(kaf_dec), ix); + vassert(kaf_offset.has_value(), "Inconsistent index state"); + res.kaf_offset = kafka::offset(*kaf_offset); + foffset_decoder_t file_dec( + _file_index.get_initial_value(), + _file_index.get_row_count(), + _file_index.copy(), + delta_delta_t(_min_file_pos_step)); + auto file_pos = _fetch_ix(std::move(file_dec), ix); + res.file_pos = *file_pos; + return res; + }); +} + +std::optional +offset_index::find_kaf_offset(kafka::offset upper_bound) { + auto search_result = maybe_find_offset( + upper_bound, _kaf_index, _kaf_offsets); + return ss::visit( + search_result, + [](std::monostate) -> std::optional { return std::nullopt; }, + [](find_result result) -> std::optional { return result; }, + [this](index_value index_result) -> std::optional { + find_result res{}; + + size_t ix = index_result.ix; + res.kaf_offset = kafka::offset(index_result.value); + + decoder_t rp_dec( + _rp_index.get_initial_value(), + _rp_index.get_row_count(), + _rp_index.copy()); + auto rp_offset = _fetch_ix(std::move(rp_dec), ix); + vassert(rp_offset.has_value(), "Inconsistent index state"); + res.rp_offset = model::offset(*rp_offset); + foffset_decoder_t file_dec( + _file_index.get_initial_value(), + _file_index.get_row_count(), + _file_index.copy(), + delta_delta_t(_min_file_pos_step)); + auto file_pos = _fetch_ix(std::move(file_dec), ix); + res.file_pos = *file_pos; + return res; + }); +} + +std::optional +offset_index::find_timestamp(model::timestamp upper_bound) { + if (_initial_time == model::timestamp::missing()) { + // Bail out early if this is a version 1 index + // that does not index timestamps. + return std::nullopt; + } + + auto search_result = maybe_find_offset( + upper_bound.value(), _time_index, _time_offsets); + + return ss::visit( + search_result, + [](std::monostate) -> std::optional { return std::nullopt; }, + [](find_result result) -> std::optional { return result; }, + [this](index_value index_result) -> std::optional { + size_t ix = index_result.ix; + + // Decode all offset indices to build up the result. + decoder_t rp_dec( + _rp_index.get_initial_value(), + _rp_index.get_row_count(), + _rp_index.copy()); + auto rp_offset = _fetch_ix(std::move(rp_dec), ix); + vassert(rp_offset.has_value(), "Inconsistent index state"); + + decoder_t kaf_dec( + _kaf_index.get_initial_value(), + _kaf_index.get_row_count(), + _kaf_index.copy()); + auto kaf_offset = _fetch_ix(std::move(kaf_dec), ix); + vassert(kaf_offset.has_value(), "Inconsistent index state"); + + foffset_decoder_t file_dec( + _file_index.get_initial_value(), + _file_index.get_row_count(), + _file_index.copy(), + delta_delta_t(_min_file_pos_step)); + auto file_pos = _fetch_ix(std::move(file_dec), ix); + vassert(file_pos.has_value(), "Inconsistent index state"); + + return offset_index::find_result{ + .rp_offset = model::offset(*rp_offset), + .kaf_offset = kafka::offset(*kaf_offset), + .file_pos = *file_pos}; + }); +} + +offset_index::coarse_index_t offset_index::build_coarse_index( + uint64_t step_size, std::string_view index_path) const { + vlog( + cst_log.trace, + "{}: building coarse index from file offset index with {} rows", + index_path, + _file_index.get_row_count()); + vassert( + step_size > static_cast(_min_file_pos_step), + "{}: step size {} cannot be less than or equal to index step size {}", + index_path, + step_size, + _min_file_pos_step); + + foffset_decoder_t file_dec( + _file_index.get_initial_value(), + _file_index.get_row_count(), + _file_index.copy(), + delta_delta_t(_min_file_pos_step)); + std::array file_row{}; + + decoder_t kaf_dec( + _kaf_index.get_initial_value(), + _kaf_index.get_row_count(), + _kaf_index.copy()); + std::array kafka_row{}; + + coarse_index_t index; + auto populate_index = [step_size, &index, index_path]( + const auto& file_offsets, + const auto& kafka_offsets, + auto& span_start, + auto& span_end) { + for (auto it = file_offsets.cbegin(), kit = kafka_offsets.cbegin(); + it != file_offsets.cend() && kit != kafka_offsets.cend(); + ++it, ++kit) { + span_end = *it; + auto delta = span_end - span_start + 1; + if (span_end > span_start && delta >= step_size) { + vlog( + cst_log.trace, + "{}: adding entry to coarse index, current file pos: {}, " + "step size: {}, span size: {}", + index_path, + span_end, + step_size, + delta); + index[kafka::offset{*kit}] = span_end; + span_start = span_end + 1; + } + } + }; + + size_t start{0}; + size_t end{0}; + while (file_dec.read(file_row) && kaf_dec.read(kafka_row)) { + populate_index(file_row, kafka_row, start, end); + file_row = {}; + kafka_row = {}; + } + + populate_index(_file_offsets, _kaf_offsets, start, end); + return index; +} + +struct offset_index_header + : serde::envelope< + offset_index_header, + serde::version<2>, + serde::compat_version<1>> { + int64_t min_file_pos_step; + uint64_t num_elements; + int64_t base_rp; + int64_t last_rp; + int64_t base_kaf; + int64_t last_kaf; + int64_t base_file; + int64_t last_file; + std::vector rp_write_buf; + std::vector kaf_write_buf; + std::vector file_write_buf; + iobuf rp_index; + iobuf kaf_index; + iobuf file_index; + + // Version 2 fields + int64_t base_time{model::timestamp::missing().value()}; + int64_t last_time{model::timestamp::missing().value()}; + std::vector time_write_buf; + iobuf time_index; + + auto serde_fields() { + return std::tie( + min_file_pos_step, + num_elements, + base_rp, + last_rp, + base_kaf, + last_kaf, + base_file, + last_file, + rp_write_buf, + kaf_write_buf, + file_write_buf, + rp_index, + kaf_index, + file_index, + base_time, + last_time, + time_write_buf, + time_index); + } +}; + +iobuf offset_index::to_iobuf() const { + offset_index_header hdr{ + .min_file_pos_step = _min_file_pos_step, + .num_elements = _pos, + .base_rp = _initial_rp, + .last_rp = _rp_index.get_last_value(), + .base_kaf = _initial_kaf, + .last_kaf = _kaf_index.get_last_value(), + .base_file = _initial_file_pos, + .last_file = _file_index.get_last_value(), + .rp_write_buf = std::vector( + _rp_offsets.begin(), _rp_offsets.end()), + .kaf_write_buf = std::vector( + _kaf_offsets.begin(), _kaf_offsets.end()), + .file_write_buf = std::vector( + _file_offsets.begin(), _file_offsets.end()), + .rp_index = _rp_index.copy(), + .kaf_index = _kaf_index.copy(), + .file_index = _file_index.copy(), + .base_time = _initial_time.value(), + .last_time = _time_index.get_last_value(), + .time_write_buf = std::vector( + _time_offsets.begin(), _time_offsets.end()), + .time_index = _time_index.copy()}; + return serde::to_iobuf(std::move(hdr)); +} + +void offset_index::from_iobuf(iobuf b) { + iobuf_parser parser(std::move(b)); + auto hdr = serde::read(parser); + auto num_rows = hdr.num_elements / buffer_depth; + _pos = hdr.num_elements; + _initial_rp = model::offset(hdr.base_rp); + _initial_kaf = kafka::offset(hdr.base_kaf); + _initial_file_pos = hdr.base_file; + std::copy( + hdr.rp_write_buf.begin(), hdr.rp_write_buf.end(), _rp_offsets.begin()); + std::copy( + hdr.kaf_write_buf.begin(), hdr.kaf_write_buf.end(), _kaf_offsets.begin()); + std::copy( + hdr.file_write_buf.begin(), + hdr.file_write_buf.end(), + _file_offsets.begin()); + _rp_index = encoder_t( + _initial_rp, num_rows, hdr.last_rp, std::move(hdr.rp_index)); + _kaf_index = encoder_t( + _initial_kaf, num_rows, hdr.last_kaf, std::move(hdr.kaf_index)); + _file_index = foffset_encoder_t( + _initial_file_pos, + num_rows, + hdr.last_file, + std::move(hdr.file_index), + delta_delta_t(_min_file_pos_step)); + _min_file_pos_step = hdr.min_file_pos_step; + + _initial_time = model::timestamp(hdr.base_time); + _time_index = encoder_t( + _initial_time.value(), + num_rows, + hdr.last_time, + std::move(hdr.time_index)); + std::copy( + hdr.time_write_buf.begin(), + hdr.time_write_buf.end(), + _time_offsets.begin()); +} + +std::optional +offset_index::_find_under(deltafor_decoder decoder, int64_t offset) { + size_t ix = 0; + std::array rp_buf{}; + std::optional candidate; + while (decoder.read(rp_buf)) { + for (auto o : rp_buf) { + if (o >= offset) { + return candidate; + } + candidate = {.ix = size_t(ix), .value = o}; + ix++; + } + rp_buf = {}; + } + return candidate; +} + +} // namespace cloud_storage diff --git a/src/v/cloud_storage/offset_index.h b/src/v/cloud_storage/offset_index.h new file mode 100644 index 0000000000000..e9e5d70201dff --- /dev/null +++ b/src/v/cloud_storage/offset_index.h @@ -0,0 +1,178 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + */ + +#pragma once + +#include "absl/container/btree_map.h" +#include "bytes/iobuf.h" +#include "model/fundamental.h" +#include "model/timestamp.h" +#include "utils/delta_for.h" + +#include +#include +#include +#include + +namespace cloud_storage { + +/// Offset index for remote_segment +/// +/// The object indexes tuples that contain three elements: +/// - redpanda offset +/// - kafka offset +/// - file offset +/// +/// The search is linear. The underlying data structure is a +/// fragmented buffer (iobuf). It is possible to search by redpanda +/// and kafka offsets, but not by file offset. +/// +/// The invariant of the offset_index is that all three encoders +/// have the same number of elements. All three buffers should also +/// have the same number of elements. +class offset_index { + static constexpr uint32_t buffer_depth = details::FOR_buffer_depth; + static constexpr uint32_t index_mask = buffer_depth - 1; + static_assert( + (buffer_depth & index_mask) == 0, + "buffer_depth have to be a power of two"); + +public: + offset_index( + model::offset initial_rp, + kafka::offset initial_kaf, + int64_t initial_file_pos, + int64_t file_pos_step, + model::timestamp initial_time); + + /// Add new tuple to the index. + void add( + model::offset rp_offset, + kafka::offset kaf_offset, + int64_t file_offset, + model::timestamp); + + struct find_result { + model::offset rp_offset; + kafka::offset kaf_offset; + int64_t file_pos; + }; + + /// Estimate memory usage by the index + size_t estimate_memory_use() const; + + /// Find index entry which is strictly lower than the redpanda offset + /// + /// The returned value has rp_offset less than upper_bound. + /// If all elements are larger than 'upper_bound' nullopt is returned. + /// If all elements are smaller than 'upper_bound' the last value is + /// returned. + std::optional find_rp_offset(model::offset upper_bound); + + /// Find index entry which is strictly lower than the kafka offset + /// + /// The returned value has kaf_offset less than upper_bound. + /// If all elements are larger than 'upper_bound' nullopt is returned. + /// If all elements are smaller than 'upper_bound' the last value is + /// returned. + std::optional find_kaf_offset(kafka::offset upper_bound); + + /// Find index entry which is strictly lower than the timestamp + /// + /// The returned value has timestamp less than upper_bound. + /// If all elements are larger than 'upper_bound' nullopt is returned. + /// If all elements are smaller than 'upper_bound' the last value is + /// returned. + std::optional find_timestamp(model::timestamp upper_bound); + + /// Builds a coarse index mapping kafka offsets to file positions. The step + /// size is the resolution of the index. So given a step size of 16MiB, the + /// result contains mappings of kafka offset to file position from the index + /// where entries are _roughly_ 16MiB apart in terms of file position. + using coarse_index_t = absl::btree_map; + coarse_index_t + build_coarse_index(uint64_t step_size, std::string_view index_path) const; + + /// Serialize offset_index + iobuf to_iobuf() const; + + /// Deserialize offset_index + void from_iobuf(iobuf in); + +private: + struct index_value { + size_t ix; + int64_t value; + }; + + /// Find index entry which is strictly lower than the provided value + /// + /// The encoder and the write buffer have to be provided via parameters. + /// The returned value is a variant which contains a monostate if no + /// value can be found; index_value if the value is found in the + /// encoder; find_result if the value is found in the write buffer (in + /// this case no further search is needed). + std::variant maybe_find_offset( + int64_t upper_bound, + deltafor_encoder& encoder, + const std::array& write_buffer); + + /// Find element inside the offset range stored in the decoder which is + /// less than offset. Return last element if no such element can be + /// found. Return nullopt if all emlements are larger or equal than + /// offset. + static std::optional + _find_under(deltafor_decoder decoder, int64_t offset); + + /// Return element by index. + template + static std::optional + _fetch_ix(DecoderT decoder, size_t target_ix) { + size_t ix = 0; + std::array buffer{}; + while (decoder.read(buffer)) { + for (auto o : buffer) { + if (ix == target_ix) { + return o; + } + ix++; + } + buffer = {}; + } + return std::nullopt; + } + +private: + std::array _rp_offsets; + std::array _kaf_offsets; + std::array _file_offsets; + std::array _time_offsets; + uint64_t _pos; + model::offset _initial_rp; + kafka::offset _initial_kaf; + int64_t _initial_file_pos; + model::timestamp _initial_time; + + using encoder_t = deltafor_encoder; + using decoder_t = deltafor_decoder; + using delta_delta_t = details::delta_delta; + using foffset_encoder_t = deltafor_encoder; + using foffset_decoder_t = deltafor_decoder; + + encoder_t _rp_index; + encoder_t _kaf_index; + foffset_encoder_t _file_index; + encoder_t _time_index; + int64_t _min_file_pos_step; + + friend class offset_index_accessor; +}; + +} // namespace cloud_storage diff --git a/src/v/cloud_storage/remote_segment_index.cc b/src/v/cloud_storage/remote_segment_index.cc index b715c55ba8660..1348add173247 100644 --- a/src/v/cloud_storage/remote_segment_index.cc +++ b/src/v/cloud_storage/remote_segment_index.cc @@ -10,425 +10,10 @@ #include "cloud_storage/remote_segment_index.h" -#include "cloud_storage/logger.h" #include "raft/consensus.h" -#include "serde/rw/envelope.h" -#include "serde/rw/iobuf.h" -#include "serde/rw/vector.h" namespace cloud_storage { -offset_index::offset_index( - model::offset initial_rp, - kafka::offset initial_kaf, - int64_t initial_file_pos, - int64_t file_pos_step, - model::timestamp initial_time) - : _rp_offsets{} - , _kaf_offsets{} - , _file_offsets{} - , _time_offsets{} - , _pos{} - , _initial_rp(initial_rp) - , _initial_kaf(initial_kaf) - , _initial_file_pos(initial_file_pos) - , _initial_time(initial_time) - , _rp_index(initial_rp) - , _kaf_index(initial_kaf) - , _file_index(initial_file_pos, delta_delta_t(file_pos_step)) - , _time_index(initial_time.value()) - , _min_file_pos_step(file_pos_step) {} - -size_t offset_index::estimate_memory_use() const { - return _file_index.mem_use() + _rp_index.mem_use() + _kaf_index.mem_use() - + _time_index.mem_use(); -} - -void offset_index::add( - model::offset rp_offset, - kafka::offset kaf_offset, - int64_t file_offset, - model::timestamp max_timestamp) { - auto ix = index_mask & _pos++; - _rp_offsets.at(ix) = rp_offset(); - _kaf_offsets.at(ix) = kaf_offset(); - _file_offsets.at(ix) = file_offset; - _time_offsets.at(ix) = max_timestamp.value(); - try { - if ((_pos & index_mask) == 0) { - _rp_index.add(_rp_offsets); - _kaf_index.add(_kaf_offsets); - _file_index.add(_file_offsets); - _time_index.add(_time_offsets); - } - } catch (...) { - // Get rid of the corrupted state in the encoders. - // If the exception is thrown out of 'add' method - // the invariant of the index is broken. - _pos = 0; - _rp_offsets = {}; - _kaf_offsets = {}; - _file_offsets = {}; - _rp_index = encoder_t(_initial_rp); - _kaf_index = encoder_t(_initial_kaf); - _file_index = foffset_encoder_t( - _initial_file_pos, delta_delta_t(_min_file_pos_step)); - _time_index = encoder_t(_initial_time.value()); - throw; - } -} - -std:: - variant - offset_index::maybe_find_offset( - int64_t upper_bound, - deltafor_encoder& encoder, - const std::array& write_buffer) { - deltafor_decoder decoder( - encoder.get_initial_value(), encoder.get_row_count(), encoder.share()); - auto max_index = encoder.get_row_count() * details::FOR_buffer_depth - 1; - auto maybe_ix = _find_under(std::move(decoder), upper_bound); - if (!maybe_ix || maybe_ix->ix == max_index) { - auto ixend = _pos & index_mask; - std::optional candidate; - for (size_t i = 0; i < ixend; i++) { - if (write_buffer.at(i) < upper_bound) { - candidate = find_result{ - .rp_offset = model::offset(_rp_offsets.at(i)), - .kaf_offset = kafka::offset(_kaf_offsets.at(i)), - .file_pos = _file_offsets.at(i), - }; - } else { - break; - } - } - // maybe_ix can point to the last element of the compressed - // chunk if all elements inside it are less than the offset that - // we're looking for. In this case we should use it even if - // we can't find anything inside the buffer. - // maybe_ix will be null if the compressed chunk is empty. - if (candidate) { - return *candidate; - } - if (!maybe_ix) { - return std::monostate(); - } - } - // Invariant: maybe_ix here can't be nullopt - return *maybe_ix; -} - -std::optional -offset_index::find_rp_offset(model::offset upper_bound) { - auto search_result = maybe_find_offset(upper_bound, _rp_index, _rp_offsets); - - return ss::visit( - search_result, - [](std::monostate) -> std::optional { return std::nullopt; }, - [](find_result result) -> std::optional { return result; }, - [this](index_value index_result) -> std::optional { - find_result res{}; - - size_t ix = index_result.ix; - res.rp_offset = model::offset(index_result.value); - - decoder_t kaf_dec( - _kaf_index.get_initial_value(), - _kaf_index.get_row_count(), - _kaf_index.copy()); - auto kaf_offset = _fetch_ix(std::move(kaf_dec), ix); - vassert(kaf_offset.has_value(), "Inconsistent index state"); - res.kaf_offset = kafka::offset(*kaf_offset); - foffset_decoder_t file_dec( - _file_index.get_initial_value(), - _file_index.get_row_count(), - _file_index.copy(), - delta_delta_t(_min_file_pos_step)); - auto file_pos = _fetch_ix(std::move(file_dec), ix); - res.file_pos = *file_pos; - return res; - }); -} - -std::optional -offset_index::find_kaf_offset(kafka::offset upper_bound) { - auto search_result = maybe_find_offset( - upper_bound, _kaf_index, _kaf_offsets); - return ss::visit( - search_result, - [](std::monostate) -> std::optional { return std::nullopt; }, - [](find_result result) -> std::optional { return result; }, - [this](index_value index_result) -> std::optional { - find_result res{}; - - size_t ix = index_result.ix; - res.kaf_offset = kafka::offset(index_result.value); - - decoder_t rp_dec( - _rp_index.get_initial_value(), - _rp_index.get_row_count(), - _rp_index.copy()); - auto rp_offset = _fetch_ix(std::move(rp_dec), ix); - vassert(rp_offset.has_value(), "Inconsistent index state"); - res.rp_offset = model::offset(*rp_offset); - foffset_decoder_t file_dec( - _file_index.get_initial_value(), - _file_index.get_row_count(), - _file_index.copy(), - delta_delta_t(_min_file_pos_step)); - auto file_pos = _fetch_ix(std::move(file_dec), ix); - res.file_pos = *file_pos; - return res; - }); -} - -std::optional -offset_index::find_timestamp(model::timestamp upper_bound) { - if (_initial_time == model::timestamp::missing()) { - // Bail out early if this is a version 1 index - // that does not index timestamps. - return std::nullopt; - } - - auto search_result = maybe_find_offset( - upper_bound.value(), _time_index, _time_offsets); - - return ss::visit( - search_result, - [](std::monostate) -> std::optional { return std::nullopt; }, - [](find_result result) -> std::optional { return result; }, - [this](index_value index_result) -> std::optional { - size_t ix = index_result.ix; - - // Decode all offset indices to build up the result. - decoder_t rp_dec( - _rp_index.get_initial_value(), - _rp_index.get_row_count(), - _rp_index.copy()); - auto rp_offset = _fetch_ix(std::move(rp_dec), ix); - vassert(rp_offset.has_value(), "Inconsistent index state"); - - decoder_t kaf_dec( - _kaf_index.get_initial_value(), - _kaf_index.get_row_count(), - _kaf_index.copy()); - auto kaf_offset = _fetch_ix(std::move(kaf_dec), ix); - vassert(kaf_offset.has_value(), "Inconsistent index state"); - - foffset_decoder_t file_dec( - _file_index.get_initial_value(), - _file_index.get_row_count(), - _file_index.copy(), - delta_delta_t(_min_file_pos_step)); - auto file_pos = _fetch_ix(std::move(file_dec), ix); - vassert(file_pos.has_value(), "Inconsistent index state"); - - return offset_index::find_result{ - .rp_offset = model::offset(*rp_offset), - .kaf_offset = kafka::offset(*kaf_offset), - .file_pos = *file_pos}; - }); -} - -offset_index::coarse_index_t offset_index::build_coarse_index( - uint64_t step_size, std::string_view index_path) const { - vlog( - cst_log.trace, - "{}: building coarse index from file offset index with {} rows", - index_path, - _file_index.get_row_count()); - vassert( - step_size > static_cast(_min_file_pos_step), - "{}: step size {} cannot be less than or equal to index step size {}", - index_path, - step_size, - _min_file_pos_step); - - foffset_decoder_t file_dec( - _file_index.get_initial_value(), - _file_index.get_row_count(), - _file_index.copy(), - delta_delta_t(_min_file_pos_step)); - std::array file_row{}; - - decoder_t kaf_dec( - _kaf_index.get_initial_value(), - _kaf_index.get_row_count(), - _kaf_index.copy()); - std::array kafka_row{}; - - coarse_index_t index; - auto populate_index = [step_size, &index, index_path]( - const auto& file_offsets, - const auto& kafka_offsets, - auto& span_start, - auto& span_end) { - for (auto it = file_offsets.cbegin(), kit = kafka_offsets.cbegin(); - it != file_offsets.cend() && kit != kafka_offsets.cend(); - ++it, ++kit) { - span_end = *it; - auto delta = span_end - span_start + 1; - if (span_end > span_start && delta >= step_size) { - vlog( - cst_log.trace, - "{}: adding entry to coarse index, current file pos: {}, " - "step size: {}, span size: {}", - index_path, - span_end, - step_size, - delta); - index[kafka::offset{*kit}] = span_end; - span_start = span_end + 1; - } - } - }; - - size_t start{0}; - size_t end{0}; - while (file_dec.read(file_row) && kaf_dec.read(kafka_row)) { - populate_index(file_row, kafka_row, start, end); - file_row = {}; - kafka_row = {}; - } - - populate_index(_file_offsets, _kaf_offsets, start, end); - return index; -} - -struct offset_index_header - : serde::envelope< - offset_index_header, - serde::version<2>, - serde::compat_version<1>> { - int64_t min_file_pos_step; - uint64_t num_elements; - int64_t base_rp; - int64_t last_rp; - int64_t base_kaf; - int64_t last_kaf; - int64_t base_file; - int64_t last_file; - std::vector rp_write_buf; - std::vector kaf_write_buf; - std::vector file_write_buf; - iobuf rp_index; - iobuf kaf_index; - iobuf file_index; - - // Version 2 fields - int64_t base_time{model::timestamp::missing().value()}; - int64_t last_time{model::timestamp::missing().value()}; - std::vector time_write_buf; - iobuf time_index; - - auto serde_fields() { - return std::tie( - min_file_pos_step, - num_elements, - base_rp, - last_rp, - base_kaf, - last_kaf, - base_file, - last_file, - rp_write_buf, - kaf_write_buf, - file_write_buf, - rp_index, - kaf_index, - file_index, - base_time, - last_time, - time_write_buf, - time_index); - } -}; - -iobuf offset_index::to_iobuf() const { - offset_index_header hdr{ - .min_file_pos_step = _min_file_pos_step, - .num_elements = _pos, - .base_rp = _initial_rp, - .last_rp = _rp_index.get_last_value(), - .base_kaf = _initial_kaf, - .last_kaf = _kaf_index.get_last_value(), - .base_file = _initial_file_pos, - .last_file = _file_index.get_last_value(), - .rp_write_buf = std::vector( - _rp_offsets.begin(), _rp_offsets.end()), - .kaf_write_buf = std::vector( - _kaf_offsets.begin(), _kaf_offsets.end()), - .file_write_buf = std::vector( - _file_offsets.begin(), _file_offsets.end()), - .rp_index = _rp_index.copy(), - .kaf_index = _kaf_index.copy(), - .file_index = _file_index.copy(), - .base_time = _initial_time.value(), - .last_time = _time_index.get_last_value(), - .time_write_buf = std::vector( - _time_offsets.begin(), _time_offsets.end()), - .time_index = _time_index.copy()}; - return serde::to_iobuf(std::move(hdr)); -} - -void offset_index::from_iobuf(iobuf b) { - iobuf_parser parser(std::move(b)); - auto hdr = serde::read(parser); - auto num_rows = hdr.num_elements / buffer_depth; - _pos = hdr.num_elements; - _initial_rp = model::offset(hdr.base_rp); - _initial_kaf = kafka::offset(hdr.base_kaf); - _initial_file_pos = hdr.base_file; - std::copy( - hdr.rp_write_buf.begin(), hdr.rp_write_buf.end(), _rp_offsets.begin()); - std::copy( - hdr.kaf_write_buf.begin(), hdr.kaf_write_buf.end(), _kaf_offsets.begin()); - std::copy( - hdr.file_write_buf.begin(), - hdr.file_write_buf.end(), - _file_offsets.begin()); - _rp_index = encoder_t( - _initial_rp, num_rows, hdr.last_rp, std::move(hdr.rp_index)); - _kaf_index = encoder_t( - _initial_kaf, num_rows, hdr.last_kaf, std::move(hdr.kaf_index)); - _file_index = foffset_encoder_t( - _initial_file_pos, - num_rows, - hdr.last_file, - std::move(hdr.file_index), - delta_delta_t(_min_file_pos_step)); - _min_file_pos_step = hdr.min_file_pos_step; - - _initial_time = model::timestamp(hdr.base_time); - _time_index = encoder_t( - _initial_time.value(), - num_rows, - hdr.last_time, - std::move(hdr.time_index)); - std::copy( - hdr.time_write_buf.begin(), - hdr.time_write_buf.end(), - _time_offsets.begin()); -} - -std::optional -offset_index::_find_under(deltafor_decoder decoder, int64_t offset) { - size_t ix = 0; - std::array rp_buf{}; - std::optional candidate; - while (decoder.read(rp_buf)) { - for (auto o : rp_buf) { - if (o >= offset) { - return candidate; - } - candidate = {.ix = size_t(ix), .value = o}; - ix++; - } - rp_buf = {}; - } - return candidate; -} - remote_segment_index_builder::remote_segment_index_builder( const model::ntp& ntp, offset_index& ix, diff --git a/src/v/cloud_storage/remote_segment_index.h b/src/v/cloud_storage/remote_segment_index.h index 7fd2cdab40274..d6e8d43f042d3 100644 --- a/src/v/cloud_storage/remote_segment_index.h +++ b/src/v/cloud_storage/remote_segment_index.h @@ -10,174 +10,17 @@ #pragma once -#include "absl/container/btree_map.h" #include "base/seastarx.h" -#include "base/units.h" #include "bytes/iobuf.h" -#include "bytes/iobuf_parser.h" +#include "cloud_storage/offset_index.h" #include "model/fundamental.h" #include "model/record_batch_types.h" #include "storage/parser.h" -#include "utils/delta_for.h" #include -#include - namespace cloud_storage { -/// Offset index for remote_segment -/// -/// The object indexes tuples that contain three elements: -/// - redpanda offset -/// - kafka offset -/// - file offset -/// -/// The search is linear. The underlying data structure is a -/// fragmented buffer (iobuf). It is possible to search by redpanda -/// and kafka offsets, but not by file offset. -/// -/// The invariant of the offset_index is that all three encoders -/// have the same number of elements. All three buffers should also -/// have the same number of elements. -class offset_index { - static constexpr uint32_t buffer_depth = details::FOR_buffer_depth; - static constexpr uint32_t index_mask = buffer_depth - 1; - static_assert( - (buffer_depth & index_mask) == 0, - "buffer_depth have to be a power of two"); - -public: - offset_index( - model::offset initial_rp, - kafka::offset initial_kaf, - int64_t initial_file_pos, - int64_t file_pos_step, - model::timestamp initial_time); - - /// Add new tuple to the index. - void add( - model::offset rp_offset, - kafka::offset kaf_offset, - int64_t file_offset, - model::timestamp); - - struct find_result { - model::offset rp_offset; - kafka::offset kaf_offset; - int64_t file_pos; - }; - - /// Estimate memory usage by the index - size_t estimate_memory_use() const; - - /// Find index entry which is strictly lower than the redpanda offset - /// - /// The returned value has rp_offset less than upper_bound. - /// If all elements are larger than 'upper_bound' nullopt is returned. - /// If all elements are smaller than 'upper_bound' the last value is - /// returned. - std::optional find_rp_offset(model::offset upper_bound); - - /// Find index entry which is strictly lower than the kafka offset - /// - /// The returned value has kaf_offset less than upper_bound. - /// If all elements are larger than 'upper_bound' nullopt is returned. - /// If all elements are smaller than 'upper_bound' the last value is - /// returned. - std::optional find_kaf_offset(kafka::offset upper_bound); - - /// Find index entry which is strictly lower than the timestamp - /// - /// The returned value has timestamp less than upper_bound. - /// If all elements are larger than 'upper_bound' nullopt is returned. - /// If all elements are smaller than 'upper_bound' the last value is - /// returned. - std::optional find_timestamp(model::timestamp upper_bound); - - /// Builds a coarse index mapping kafka offsets to file positions. The step - /// size is the resolution of the index. So given a step size of 16MiB, the - /// result contains mappings of kafka offset to file position from the index - /// where entries are _roughly_ 16MiB apart in terms of file position. - using coarse_index_t = absl::btree_map; - coarse_index_t - build_coarse_index(uint64_t step_size, std::string_view index_path) const; - - /// Serialize offset_index - iobuf to_iobuf() const; - - /// Deserialize offset_index - void from_iobuf(iobuf in); - -private: - struct index_value { - size_t ix; - int64_t value; - }; - - /// Find index entry which is strictly lower than the provided value - /// - /// The encoder and the write buffer have to be provided via parameters. - /// The returned value is a variant which contains a monostate if no - /// value can be found; index_value if the value is found in the - /// encoder; find_result if the value is found in the write buffer (in - /// this case no further search is needed). - std::variant maybe_find_offset( - int64_t upper_bound, - deltafor_encoder& encoder, - const std::array& write_buffer); - - /// Find element inside the offset range stored in the decoder which is - /// less than offset. Return last element if no such element can be - /// found. Return nullopt if all emlements are larger or equal than - /// offset. - static std::optional - _find_under(deltafor_decoder decoder, int64_t offset); - - /// Return element by index. - template - static std::optional - _fetch_ix(DecoderT decoder, size_t target_ix) { - size_t ix = 0; - std::array buffer{}; - while (decoder.read(buffer)) { - for (auto o : buffer) { - if (ix == target_ix) { - return o; - } - ix++; - } - buffer = {}; - } - return std::nullopt; - } - -private: - std::array _rp_offsets; - std::array _kaf_offsets; - std::array _file_offsets; - std::array _time_offsets; - uint64_t _pos; - model::offset _initial_rp; - kafka::offset _initial_kaf; - int64_t _initial_file_pos; - model::timestamp _initial_time; - - using encoder_t = deltafor_encoder; - using decoder_t = deltafor_decoder; - using delta_delta_t = details::delta_delta; - using foffset_encoder_t = deltafor_encoder; - using foffset_decoder_t = deltafor_decoder; - - encoder_t _rp_index; - encoder_t _kaf_index; - foffset_encoder_t _file_index; - encoder_t _time_index; - int64_t _min_file_pos_step; - - friend class offset_index_accessor; -}; - struct segment_record_stats { // Offset of the first record in the segment model::offset base_rp_offset; diff --git a/src/v/cloud_topics/level_one/common/BUILD b/src/v/cloud_topics/level_one/common/BUILD index c1241d3773856..bd6e2a40ae839 100644 --- a/src/v/cloud_topics/level_one/common/BUILD +++ b/src/v/cloud_topics/level_one/common/BUILD @@ -20,9 +20,13 @@ redpanda_cc_library( ], deps = [ "//src/v/base", + "//src/v/model", + "//src/v/serde", + "//src/v/serde:enum", "//src/v/utils:named_type", "//src/v/utils:uuid", "@fmt", + "@seastar", ], ) @@ -80,6 +84,7 @@ redpanda_cc_library( "//src/v/cloud_io:admission_control_types", "//src/v/cloud_storage_clients", "//src/v/container:chunked_vector", + "//src/v/model", "//src/v/utils:named_type", "@seastar", ], @@ -97,6 +102,87 @@ redpanda_cc_library( ], ) +redpanda_cc_library( + name = "object_handle", + srcs = ["object_handle.cc"], + hdrs = ["object_handle.h"], + deps = [ + ":abstract_io", + ":object", + ":object_id", + "//src/v/model", + "@seastar", + ], +) + +redpanda_cc_library( + name = "open_object", + srcs = ["open_object.cc"], + hdrs = ["open_object.h"], + implementation_deps = [ + ":object", + ":ts_object", + "//src/v/cloud_storage", + "//src/v/cloud_storage:offset_index", + "//src/v/cloud_topics:logger", + "//src/v/config", + "//src/v/model", + ], + deps = [ + ":abstract_io", + ":object_handle", + ":object_id", + "@seastar", + ], +) + +redpanda_cc_library( + name = "chunk_data_source", + hdrs = ["chunk_data_source.h"], + deps = [ + "//src/v/base", + "@fmt", + "@seastar", + ], +) + +redpanda_cc_library( + name = "ts_reader", + srcs = ["ts_reader.cc"], + hdrs = ["ts_reader.h"], + visibility = ["//visibility:private"], + implementation_deps = [ + "//src/v/bytes:iostream", + "//src/v/storage:record_batch_utils", + "@fmt", + ], + deps = [ + ":object", + "//src/v/model", + "@abseil-cpp//absl/container:btree", + "@seastar", + ], +) + +redpanda_cc_library( + name = "ts_object", + srcs = ["ts_object.cc"], + hdrs = ["ts_object.h"], + visibility = ["//visibility:private"], + implementation_deps = [ + ":ts_reader", + ], + deps = [ + ":abstract_io", + ":chunk_data_source", + ":object_handle", + "//src/v/cloud_storage:offset_index", + "//src/v/model", + "@abseil-cpp//absl/container:btree", + "@seastar", + ], +) + redpanda_cc_library( name = "file_io", srcs = [ @@ -109,6 +195,8 @@ redpanda_cc_library( ], implementation_deps = [ "//src/v/base", + "//src/v/bytes:iostream", + "//src/v/cloud_storage", ], visibility = [ "//src/v/cloud_topics:__pkg__", @@ -137,13 +225,17 @@ redpanda_cc_library( name = "fake_io", srcs = ["fake_io.cc"], hdrs = ["fake_io.h"], + implementation_deps = [ + "//src/v/base", + ], deps = [ ":abstract_io", + ":object", ":object_id", - ":object_utils", "//src/v/bytes:iobuf", "//src/v/bytes:iostream", "//src/v/cloud_storage_clients", + "//src/v/model", "@abseil-cpp//absl/container:btree", "@seastar", ], diff --git a/src/v/cloud_topics/level_one/common/abstract_io.cc b/src/v/cloud_topics/level_one/common/abstract_io.cc index 9863c343de8a5..e154a312ea589 100644 --- a/src/v/cloud_topics/level_one/common/abstract_io.cc +++ b/src/v/cloud_topics/level_one/common/abstract_io.cc @@ -18,7 +18,7 @@ ss::future> io::read_file(staging_file* file) { return file->input_stream(); } -ss::future> io::read_object_as_iobuf( +ss::future> io::fetch_native_footer( object_extent extent, ss::abort_source* as, cloud_io::group_id gid, diff --git a/src/v/cloud_topics/level_one/common/abstract_io.h b/src/v/cloud_topics/level_one/common/abstract_io.h index 9b63fec0f6283..065bebe1026e8 100644 --- a/src/v/cloud_topics/level_one/common/abstract_io.h +++ b/src/v/cloud_topics/level_one/common/abstract_io.h @@ -15,6 +15,7 @@ #include "cloud_storage_clients/multipart_upload.h" #include "cloud_topics/level_one/common/object_id.h" #include "container/chunked_vector.h" +#include "model/record.h" #include #include @@ -25,6 +26,11 @@ namespace cloud_topics::l1 { +// Forward declaration — full definition in object_handle.h. +// Do not #include object_handle.h here: object_handle.h already includes +// abstract_io.h for io::errc, so including it here creates a cycle. +class object_handle; + // An abstraction for a local file that is used for staging uploads to object // storage. class staging_file { @@ -95,17 +101,40 @@ class io { cloud_io::group_id g, bool skip_cache = false) = 0; - // The same as `read_object` except that instead of returning an input - // stream, the data is fully buffered into an `iobuf`. - virtual ss::future> read_object_as_iobuf( + // Read a native L1 object's footer region (the passed extent) fully + // buffered into an `iobuf`, so open_object can parse the footer index. + // Like read_object, may be served from or populate the cache unless + // `skip_cache` is set. + virtual ss::future> fetch_native_footer( object_extent, ss::abort_source*, cloud_io::group_id g, bool skip_cache = false); - // Delete the specified objects from object storage. + // Fetch the raw serialized offset index (.index sidecar) of an imported + // tiered-storage segment. `extent.imported` identifies the segment. Returns + // cloud_missing_object when the segment has no .index (the caller falls + // back to a full-segment scan). The bytes are returned unparsed so this + // thin io seam stays free of the cloud_storage offset-index format; + // open_object deserializes them. Only meaningful for imported extents. + virtual ss::future> + fetch_ts_index(object_extent, ss::abort_source*) = 0; + + // Fetch the aborted-transaction ranges of an imported tiered-storage + // segment (parsed from its .tx manifest sidecar), in raw log-offset space. + // `extent.imported` identifies the segment. Returns cloud_missing_object + // when the .tx object is absent. Ranges (rather than the manifest bytes) + // are returned so the cloud_storage manifest format stays behind the io + // seam; open_object assembles them into the aborted set. Only meaningful + // for imported extents. + virtual ss::future, errc>> + fetch_ts_tx(object_extent, ss::abort_source*) = 0; + + // Delete the specified objects from object storage. An entry with a ts_path + // is addressed by that tiered-storage segment path; otherwise the native L1 + // object path is used. Both live in the one configured object bucket. virtual ss::future> - delete_objects(chunked_vector, ss::abort_source*) = 0; + delete_objects(chunked_vector, ss::abort_source*) = 0; // Create a multipart upload for streaming data directly to object storage. virtual ss::future< diff --git a/src/v/cloud_topics/level_one/common/chunk_data_source.h b/src/v/cloud_topics/level_one/common/chunk_data_source.h new file mode 100644 index 0000000000000..cc075a53efa4b --- /dev/null +++ b/src/v/cloud_topics/level_one/common/chunk_data_source.h @@ -0,0 +1,138 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + */ +#pragma once + +#include "base/vassert.h" + +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include + +namespace cloud_topics::l1 { + +/// A data source that fetches the byte range [start_pos, start_pos+total_len) +/// one fixed-size chunk at a time, lazily, and presents the result as a single +/// contiguous stream. +/// +/// Only the chunks actually consumed are fetched: a reader that stops early +/// simply stops pulling, so the un-read tail is never fetched. Chunks are fixed +/// byte ranges and need not be aligned to any structure in the data -- the +/// source concatenates one chunk's tail with the next chunk's head, so a record +/// spanning a boundary is read seamlessly. EOF is signalled only at the true +/// end (start_pos+total_len), so a consumer never mistakes a chunk boundary for +/// the end of the data. +/// +/// `Fetch` is any callable +/// (size_t file_position, size_t length, ss::abort_source*) +/// -> ss::future, E>> +/// for any formattable error `E`. A chunk-fetch failure (an unexpected result) +/// surfaces as an exception from get(); a record-batch reader propagates it +/// (its read API has no error variant), matching how such readers already +/// report stream errors. +template +class chunk_data_source final : public ss::data_source_impl { +public: + chunk_data_source( + Fetch fetch, + size_t start_pos, + size_t total_len, + size_t chunk_size, + ss::abort_source* as) + : _fetch(std::move(fetch)) + , _end(start_pos + total_len) + , _chunk_size(chunk_size) + , _next_chunk_start(start_pos - (start_pos % chunk_size)) + , _skip_head(start_pos % chunk_size) + , _as(as) { + vassert( + chunk_size > 0, "chunk_data_source requires a non-zero chunk size"); + } + + ss::future> get() override { + while (true) { + if (!_cur.has_value()) { + if (_next_chunk_start >= _end) { + // Past the requested range: true EOF. + co_return ss::temporary_buffer{}; + } + auto chunk_end = std::min( + _next_chunk_start + _chunk_size, _end); + auto len = chunk_end - _next_chunk_start; + auto stream = co_await _fetch(_next_chunk_start, len, _as); + if (!stream.has_value()) { + throw std::runtime_error( + fmt::format( + "failed to fetch chunk at byte {} (len {}): {}", + _next_chunk_start, + len, + stream.error())); + } + _cur = std::move(*stream); + if (_skip_head > 0) { + co_await _cur->skip(_skip_head); + _skip_head = 0; + } + _next_chunk_start += _chunk_size; + } + auto buf = co_await _cur->read(); + if (buf.empty()) { + // Current chunk exhausted; advance to the next. + co_await _cur->close(); + _cur.reset(); + continue; + } + co_return buf; + } + } + + ss::future<> close() override { + if (_cur.has_value()) { + co_await _cur->close(); + _cur.reset(); + } + } + +private: + Fetch _fetch; + size_t _end; + size_t _chunk_size; + // Start of the next chunk to fetch, aligned down to a chunk boundary. + size_t _next_chunk_start; + // Bytes to discard from the first chunk so the stream starts at start_pos + // (start_pos may sit mid-chunk). Zeroed after the first chunk. + size_t _skip_head; + std::optional> _cur; + ss::abort_source* _as; +}; + +/// Deduce `Fetch` and wrap a chunk_data_source in an ss::data_source. +template +ss::data_source make_chunk_data_source( + Fetch fetch, + size_t start_pos, + size_t total_len, + size_t chunk_size, + ss::abort_source* as) { + return ss::data_source{std::make_unique>( + std::move(fetch), start_pos, total_len, chunk_size, as)}; +} + +} // namespace cloud_topics::l1 diff --git a/src/v/cloud_topics/level_one/common/fake_io.cc b/src/v/cloud_topics/level_one/common/fake_io.cc index 9ca1892f999d2..4b3b94592d3c5 100644 --- a/src/v/cloud_topics/level_one/common/fake_io.cc +++ b/src/v/cloud_topics/level_one/common/fake_io.cc @@ -10,8 +10,10 @@ #include "cloud_topics/level_one/common/fake_io.h" +#include "base/vassert.h" #include "bytes/iostream.h" #include "cloud_storage_clients/multipart_upload.h" +#include "cloud_topics/level_one/common/object.h" #include "cloud_topics/level_one/common/object_id.h" namespace cloud_topics::l1 { @@ -118,6 +120,17 @@ fake_io::read_object( // fake_io keeps everything in memory and never caches, so it is always // effectively a streaming read regardless of this flag. [[maybe_unused]] bool skip_cache) { + // An imported extent reads its bytes from the injected TS segment at + // ts_path; a native extent reads the object stored under its id. + // read_object is uniform over both, matching file_io. + if (extent.imported.has_value()) { + auto it = _ts_storage.find(extent.imported->ts_path); + if (it == _ts_storage.end()) { + co_return std::unexpected(io::errc::cloud_missing_object); + } + co_return make_iobuf_input_stream( + it->second.bytes.share(extent.position, extent.size)); + } co_return get_object(extent.id) .transform( [&extent]( @@ -128,14 +141,69 @@ fake_io::read_object( .value_or(std::unexpected(io::errc::cloud_missing_object)); } -ss::future> -fake_io::delete_objects(chunked_vector oids, ss::abort_source*) { - for (const auto& oid : oids) { - remove_object(oid); +void fake_io::put_ts_segment( + ts_segment_path ts_path, + iobuf segment_bytes, + aborted_transactions aborted, + std::optional index_bytes) { + _ts_storage.insert_or_assign( + std::move(ts_path), + ts_segment_fixture{ + .bytes = std::move(segment_bytes), + .aborted = std::move(aborted), + .index_bytes = std::move(index_bytes), + }); +} + +ss::future> +fake_io::fetch_ts_index(object_extent extent, ss::abort_source*) { + vassert( + extent.imported.has_value(), + "fetch_ts_index requires an imported extent"); + auto it = _ts_storage.find(extent.imported->ts_path); + if (it == _ts_storage.end() || !it->second.index_bytes.has_value()) { + // No injected .index: mirror file_io's notfound so open_object falls + // back to a full-segment scan with an empty index. + co_return std::unexpected(io::errc::cloud_missing_object); + } + co_return it->second.index_bytes->copy(); +} + +ss::future, io::errc>> +fake_io::fetch_ts_tx(object_extent extent, ss::abort_source*) { + vassert( + extent.imported.has_value(), "fetch_ts_tx requires an imported extent"); + auto it = _ts_storage.find(extent.imported->ts_path); + if (it == _ts_storage.end()) { + co_return std::unexpected(io::errc::cloud_missing_object); + } + // Return the injected aborted ranges as file_io would after parsing the .tx + // manifest; open_object collects them into the aborted set. + chunked_vector ranges; + for (const auto& r : it->second.aborted) { + ranges.push_back(r); + } + co_return std::move(ranges); +} + +ss::future> fake_io::delete_objects( + chunked_vector objects, ss::abort_source*) { + for (const auto& obj : objects) { + if (obj.ts_path.has_value()) { + // Imported object: its backing segment is addressed by ts_path + // (mirrors file_io's path routing). + _ts_storage.erase(*obj.ts_path); + } else { + remove_object(obj.id); + } } co_return std::expected{}; } +bool fake_io::has_ts_segment(const ts_segment_path& ts_path) const { + return _ts_storage.contains(ts_path); +} + std::optional fake_io::get_object(object_id id) { auto it = _storage.find(id); if (it == _storage.end()) { diff --git a/src/v/cloud_topics/level_one/common/fake_io.h b/src/v/cloud_topics/level_one/common/fake_io.h index c0a4514508437..93995842aed52 100644 --- a/src/v/cloud_topics/level_one/common/fake_io.h +++ b/src/v/cloud_topics/level_one/common/fake_io.h @@ -13,6 +13,9 @@ #include "absl/container/btree_map.h" #include "bytes/iobuf.h" #include "cloud_topics/level_one/common/abstract_io.h" +#include "cloud_topics/level_one/common/object.h" + +#include namespace cloud_topics::l1 { @@ -32,8 +35,14 @@ class fake_io : public io { cloud_io::group_id g, bool skip_cache) override; + ss::future> + fetch_ts_index(object_extent, ss::abort_source*) override; + + ss::future, errc>> + fetch_ts_tx(object_extent, ss::abort_source*) override; + ss::future> - delete_objects(chunked_vector, ss::abort_source*) override; + delete_objects(chunked_vector, ss::abort_source*) override; ss::future> create_multipart_upload( @@ -51,8 +60,34 @@ class fake_io : public io { // Return a list of the object IDs that haven't been removed. chunked_vector list_objects() const; + // Whether an injected TS segment (see put_ts_segment) is still present. + // For tests that exercise imported-object deletion. + bool has_ts_segment(const ts_segment_path& ts_path) const; + + /// Inject a raw TS-format segment for use with open_object on imported + /// extents whose ts_path matches. open_object always seeks through the real + /// ts_segment_index: with index_bytes (a serialized offset_index, as + /// file_io downloads) it is deserialized; without one the index is empty, + /// so seeks fall back to a full-segment scan from 0 (file_io's + /// missing-.index path). The segment's Kafka offset bounds come from the + /// read request (extent.imported), mirroring file_io, not from here. + void put_ts_segment( + ts_segment_path ts_path, + iobuf segment_bytes, + aborted_transactions aborted = {}, + std::optional index_bytes = std::nullopt); + private: + struct ts_segment_fixture { + iobuf bytes; + aborted_transactions aborted; + // Serialized offset_index (.index) for an index-backed seek; nullopt + // means the full-segment-scan fallback. + std::optional index_bytes; + }; + absl::btree_map _storage; + absl::btree_map _ts_storage; }; } // namespace cloud_topics::l1 diff --git a/src/v/cloud_topics/level_one/common/file_io.cc b/src/v/cloud_topics/level_one/common/file_io.cc index f0240ca27ce28..eb73128ad0670 100644 --- a/src/v/cloud_topics/level_one/common/file_io.cc +++ b/src/v/cloud_topics/level_one/common/file_io.cc @@ -10,8 +10,12 @@ #include "cloud_topics/level_one/common/file_io.h" +#include "base/vassert.h" +#include "bytes/iostream.h" #include "cloud_io/io_result.h" #include "cloud_io/remote.h" +#include "cloud_storage/remote_segment.h" +#include "cloud_storage/tx_range_manifest.h" #include "cloud_storage_clients/client.h" #include "cloud_topics/level_one/common/abstract_io.h" #include "cloud_topics/level_one/common/object_id.h" @@ -229,6 +233,46 @@ class streaming_download_source final : public ss::data_source_impl { ss::chunked_fifo> _current; }; +// Download a whole object by key into an iobuf. Used for small sidecar objects +// (the imported segment's index and tx-range manifest) that are fetched in one +// shot rather than streamed through the cache. +ss::future> download_raw_iobuf( + cloud_io::remote* remote, + const cloud_storage_clients::bucket_name& bucket, + const ss::sstring& key, + ss::abort_source* as) { + static constexpr auto timeout = 10s; + static constexpr auto backoff = 100ms; + retry_chain_node root(*as, ss::lowres_clock::now() + timeout, backoff); + iobuf result; + auto res_fut = co_await ss::coroutine::as_future( + remote->download_object({ + .transfer_details = { + .bucket = bucket, + .key = cloud_storage_clients::object_key{key}, + .parent_rtc = root, + }, + .display_str = "ts_raw_download", + .payload = result, + })); + if (res_fut.failed()) { + auto ex = res_fut.get_exception(); + vlog(cd_log.warn, "Error downloading raw object {}: {}", key, ex); + co_return std::unexpected(io::errc::cloud_op_error); + } + switch (res_fut.get()) { + case cloud_io::download_result::success: + co_return std::move(result); + case cloud_io::download_result::notfound: + co_return std::unexpected(io::errc::cloud_missing_object); + case cloud_io::download_result::timedout: + co_return std::unexpected(io::errc::cloud_op_timeout); + case cloud_io::download_result::failed: + co_return std::unexpected(io::errc::cloud_op_error); + } + std::unreachable(); +} + } // namespace file_io::file_io( @@ -319,21 +363,24 @@ ss::future file_io::save_to_cache( } ss::future> file_io::do_download_to_cache( - const object_extent& extent, + const cloud_storage_clients::object_key& key, + cloud_storage_clients::http_byte_range range, const std::filesystem::path& cache_key, + std::string_view download_label, retry_chain_node& root, ss::abort_source& as, cloud_io::group_id gid) { + const auto range_size = range.second - range.first + 1; // TODO(cloud_topics): reserving space should also take an abort_source auto reservation_fut = co_await ss::coroutine::as_future( - _cache->reserve_space(extent.size, 1)); + _cache->reserve_space(range_size, 1)); if (reservation_fut.failed()) { auto ex = reservation_fut.get_exception(); vlog( cd_log.warn, "Error reserving cache space for download of {}: {}", - extent, + key, ex); co_return std::unexpected(io::errc::file_io_error); } @@ -348,19 +395,18 @@ ss::future> file_io::do_download_to_cache( _remote->download_stream( cloud_io::transfer_details{ .bucket = _bucket, - .key = object_path_factory::level_one_path(extent.id), + .key = key, .parent_rtc = root, }, consumer, - "l1_file_download", + download_label, /*acquire_hydration_units=*/true, - cloud_storage_clients::http_byte_range{ - extent.position, extent.position + extent.size - 1}, + range, {}, gid)); if (result_fut.failed()) { auto ex = result_fut.get_exception(); - vlog(cd_log.warn, "Error downloading object {}: {}", extent, ex); + vlog(cd_log.warn, "Error downloading object {}: {}", key, ex); // Map abort to cloud_op_timeout so a leader-abort and a // merger-abort produce the same errc for the same event. co_return std::unexpected( @@ -401,7 +447,30 @@ file_io::read_object( // them). // TODO(cloud_topics): If reading just a footer, we should skip the cache. // Maybe we need another method for that which is iobuf based? - auto cache_key = file_io::cache_key(extent); + // + // Resolve the storage key, cache key, and download label uniformly for + // native and imported extents; the range is then read the same way for + // both. A native object lives at its L1 object path; an imported + // tiered-storage segment lives at its own ts_path (keyed per (pos,size) so + // the cache dedups across reads touching the same chunk). + cloud_storage_clients::object_key key; + std::filesystem::path cache_key; + std::string_view download_label; + if (extent.imported.has_value()) { + key = cloud_storage_clients::object_key{extent.imported->ts_path()}; + cache_key = fmt::format( + "ts_{}_position_{}_size_{}.partial", + extent.imported->ts_path(), + extent.position, + extent.size); + download_label = "ts_segment_download"; + } else { + key = object_path_factory::level_one_path(extent.id); + cache_key = file_io::cache_key(extent); + download_label = "l1_file_download"; + } + const cloud_storage_clients::http_byte_range range{ + extent.position, extent.position + extent.size - 1}; while (true) { auto stream_fut = co_await ss::coroutine::as_future< std::optional>(_cache->get_stream( @@ -430,9 +499,8 @@ file_io::read_object( std::make_unique( _remote, _bucket, - object_path_factory::level_one_path(extent.id), - cloud_storage_clients::http_byte_range{ - extent.position, extent.position + extent.size - 1}, + key, + range, *as, gid, config::shard_local_cfg() @@ -443,8 +511,9 @@ file_io::read_object( auto r = co_await _single_flight.run( cache_key, *as, - [this, &extent, &cache_key, &root, as, gid]() { - return do_download_to_cache(extent, cache_key, root, *as, gid); + [this, &key, range, &cache_key, download_label, &root, as, gid]() { + return do_download_to_cache( + key, range, cache_key, download_label, root, *as, gid); }, &cd_log); @@ -463,19 +532,61 @@ file_io::read_object( } } -ss::future> -file_io::delete_objects(chunked_vector ids, ss::abort_source* as) { - static constexpr auto timeout = 10s; - static constexpr auto backoff = 100ms; - retry_chain_node root(*as, ss::lowres_clock::now() + timeout, backoff); - chunked_vector keys; - for (const auto& id : ids) { - keys.push_back(object_path_factory::level_one_path(id)); +ss::future> file_io::fetch_native_footer( + object_extent extent, + ss::abort_source* as, + cloud_io::group_id gid, + bool skip_cache) { + if (_probe != nullptr) { + _probe->register_footer_read(extent.size); + } + return io::fetch_native_footer(extent, as, gid, skip_cache); +} + +ss::future> +file_io::fetch_ts_index(object_extent extent, ss::abort_source* as) { + vassert( + extent.imported.has_value(), + "fetch_ts_index requires an imported extent"); + auto index_path = cloud_storage::generate_index_path( + cloud_storage::remote_segment_path{ + std::filesystem::path{extent.imported->ts_path()}}); + auto index_iobuf = co_await download_raw_iobuf( + _remote, _bucket, index_path.native(), as); + if (index_iobuf.has_value() && _probe != nullptr) { + _probe->register_ts_index_read(index_iobuf->size_bytes()); + } + co_return index_iobuf; +} + +ss::future, io::errc>> +file_io::fetch_ts_tx(object_extent extent, ss::abort_source* as) { + vassert( + extent.imported.has_value(), "fetch_ts_tx requires an imported extent"); + cloud_storage::remote_segment_path seg_path{ + std::filesystem::path{extent.imported->ts_path()}}; + auto tx_path = cloud_storage::generate_remote_tx_path(seg_path); + auto tx_iobuf = co_await download_raw_iobuf( + _remote, _bucket, tx_path().native(), as); + if (!tx_iobuf.has_value()) { + co_return std::unexpected(tx_iobuf.error()); } + if (_probe != nullptr) { + _probe->register_ts_tx_read(tx_iobuf->size_bytes()); + } + cloud_storage::tx_range_manifest manifest(seg_path); + co_await manifest.update(make_iobuf_input_stream(std::move(*tx_iobuf))); + co_return std::move(manifest).get_tx_range(); +} + +ss::future> file_io::delete_keys( + const cloud_storage_clients::bucket_name& bucket, + chunked_vector keys, + retry_chain_node& root) { auto result_fut = co_await ss::coroutine::as_future( _remote->delete_objects( - _bucket, std::move(keys), root, [](size_t retry_count) { + bucket, std::move(keys), root, [](size_t retry_count) { std::ignore = retry_count; })); if (result_fut.failed()) { @@ -495,6 +606,64 @@ file_io::delete_objects(chunked_vector ids, ss::abort_source* as) { std::unreachable(); } +ss::future> file_io::delete_objects( + chunked_vector objects, ss::abort_source* as) { + static constexpr auto timeout = 10s; + static constexpr auto backoff = 100ms; + retry_chain_node root(*as, ss::lowres_clock::now() + timeout, backoff); + + chunked_vector native_keys; + chunked_vector ts_keys; + size_t ts_count = 0; + for (const auto& obj : objects) { + if (obj.ts_path.has_value()) { + ++ts_count; + // An imported segment owns three cloud objects (the segment, its + // .tx range manifest, and its .index), the same set the archiver + // would have deleted. Remove all three so nothing is orphaned. + cloud_storage::remote_segment_path seg_path{ + std::filesystem::path{(*obj.ts_path)()}}; + ts_keys.push_back( + cloud_storage_clients::object_key{(*obj.ts_path)()}); + ts_keys.push_back( + cloud_storage_clients::object_key{ + cloud_storage::generate_remote_tx_path(seg_path)().native()}); + ts_keys.push_back( + cloud_storage_clients::object_key{ + cloud_storage::generate_index_path(seg_path).native()}); + } else { + native_keys.push_back(object_path_factory::level_one_path(obj.id)); + } + } + + if (!native_keys.empty()) { + auto l1_object_count = native_keys.size(); + auto res = co_await delete_keys(_bucket, std::move(native_keys), root); + if (!res.has_value()) { + vlog( + cd_log.warn, + "Failed to delete {} L1 objects: {}", + l1_object_count, + res.error()); + co_return res; + } + } + if (!ts_keys.empty()) { + auto ts_key_count = ts_keys.size(); + auto res = co_await delete_keys(_bucket, std::move(ts_keys), root); + if (!res.has_value()) { + vlog( + cd_log.warn, + "Failed to delete {} keys for {} imported TS segments: {}", + ts_key_count, + ts_count, + res.error()); + co_return res; + } + } + co_return std::expected{}; +} + ss::future> file_io::create_multipart_upload( object_id oid, size_t part_size, ss::abort_source* as) { diff --git a/src/v/cloud_topics/level_one/common/file_io.h b/src/v/cloud_topics/level_one/common/file_io.h index 66c7f418584c3..ee5e490e7dc68 100644 --- a/src/v/cloud_topics/level_one/common/file_io.h +++ b/src/v/cloud_topics/level_one/common/file_io.h @@ -60,8 +60,20 @@ class file_io : public io { cloud_io::group_id g, bool skip_cache) override; + ss::future> fetch_native_footer( + object_extent, + ss::abort_source*, + cloud_io::group_id g, + bool skip_cache) override; + + ss::future> + fetch_ts_index(object_extent, ss::abort_source*) override; + + ss::future, errc>> + fetch_ts_tx(object_extent, ss::abort_source*) override; + ss::future> - delete_objects(chunked_vector, ss::abort_source*) override; + delete_objects(chunked_vector, ss::abort_source*) override; ss::future> create_multipart_upload( @@ -74,17 +86,28 @@ class file_io : public io { std::filesystem::path, uint64_t content_length); - /// Reserve cache space, run the S3 GET, and stream the bytes into - /// the cloud cache under `cache_key`. Succeeds, or fails with the - /// mapped errc on reservation / download failure. + /// Reserve cache space, run the S3 GET against `key` for byte `range`, and + /// stream the bytes into the cloud cache under `cache_key` (labelling the + /// transfer `download_label`). Succeeds, or fails with the mapped errc on + /// reservation / download failure. ss::future> do_download_to_cache( - const object_extent& extent, + const cloud_storage_clients::object_key& key, + cloud_storage_clients::http_byte_range range, const std::filesystem::path& cache_key, + std::string_view download_label, retry_chain_node& root, ss::abort_source& as, cloud_io::group_id gid); + // Delete a batch of keys from the given bucket. + ss::future> delete_keys( + const cloud_storage_clients::bucket_name& bucket, + chunked_vector keys, + retry_chain_node& parent); + cloud_io::remote* _remote; + // Holds both native L1 objects and imported tiered-storage segments: a + // single cluster always stores both in the one configured object bucket. cloud_storage_clients::bucket_name _bucket; std::filesystem::path _staging_dir; cloud_io::cache* _cache; diff --git a/src/v/cloud_topics/level_one/common/file_io_probe.cc b/src/v/cloud_topics/level_one/common/file_io_probe.cc index 1aa6e19f9f287..5746c5e0815fa 100644 --- a/src/v/cloud_topics/level_one/common/file_io_probe.cc +++ b/src/v/cloud_topics/level_one/common/file_io_probe.cc @@ -42,6 +42,22 @@ void file_io_probe::setup_metrics() { sm::description( "Cache misses that joined an in-flight download for the " "same extent.")), + sm::make_counter( + "footer_read_bytes", + [this] { return _footer_bytes_read; }, + sm::description("Native L1 object footer bytes read by L1 readers.")), + sm::make_counter( + "ts_index_read_bytes", + [this] { return _ts_index_bytes_read; }, + sm::description( + "Number of imported tiered-storage segment index " + "bytes read by L1 readers.")), + sm::make_counter( + "ts_tx_read_bytes", + [this] { return _ts_tx_bytes_read; }, + sm::description( + "Number of imported tiered-storage segment " + "tx-manifest bytes read by L1 readers.")), }); } diff --git a/src/v/cloud_topics/level_one/common/file_io_probe.h b/src/v/cloud_topics/level_one/common/file_io_probe.h index d952561343c7c..d95e2617f87f8 100644 --- a/src/v/cloud_topics/level_one/common/file_io_probe.h +++ b/src/v/cloud_topics/level_one/common/file_io_probe.h @@ -11,6 +11,7 @@ #include "metrics/metrics.h" +#include #include namespace cloud_topics::l1 { @@ -24,6 +25,10 @@ class file_io_probe { void register_cache_miss() { ++_cache_misses; } void register_concurrent_read_merge() { ++_concurrent_read_merges; } + void register_footer_read(size_t bytes) { _footer_bytes_read += bytes; } + void register_ts_index_read(size_t bytes) { _ts_index_bytes_read += bytes; } + void register_ts_tx_read(size_t bytes) { _ts_tx_bytes_read += bytes; } + private: void setup_metrics(); @@ -31,6 +36,10 @@ class file_io_probe { uint64_t _cache_misses{0}; uint64_t _concurrent_read_merges{0}; + uint64_t _footer_bytes_read{0}; + uint64_t _ts_index_bytes_read{0}; + uint64_t _ts_tx_bytes_read{0}; + metrics::internal_metric_groups _metrics; }; diff --git a/src/v/cloud_topics/level_one/common/object.h b/src/v/cloud_topics/level_one/common/object.h index 41197a46ec33d..72a6048c6159b 100644 --- a/src/v/cloud_topics/level_one/common/object.h +++ b/src/v/cloud_topics/level_one/common/object.h @@ -11,6 +11,7 @@ #pragma once #include "absl/container/btree_map.h" +#include "absl/container/btree_set.h" #include "base/format_to.h" #include "base/seastarx.h" #include "base/units.h" @@ -26,10 +27,16 @@ #include +#include #include namespace cloud_topics::l1 { +// The aborted transaction ranges carried alongside a tiered-storage segment, +// ordered descending so a forward scan can drop already-consumed ranges off +// the front as it advances. +using aborted_transactions = absl::btree_set>; + // clang-format off // L1 Object File Format: // ===================== diff --git a/src/v/cloud_topics/level_one/common/object_handle.cc b/src/v/cloud_topics/level_one/common/object_handle.cc new file mode 100644 index 0000000000000..d802a1b9ac49b --- /dev/null +++ b/src/v/cloud_topics/level_one/common/object_handle.cc @@ -0,0 +1,53 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + */ +#include "cloud_topics/level_one/common/object_handle.h" + +#include "cloud_topics/level_one/common/abstract_io.h" +#include "cloud_topics/level_one/common/object.h" +#include "cloud_topics/level_one/common/object_id.h" + +namespace cloud_topics::l1 { + +l1_footer_index::l1_footer_index(footer f) + : _footer(std::move(f)) {} + +std::optional l1_footer_index::seek_to_offset( + model::topic_id_partition tidp, kafka::offset offset) const { + auto r = _footer.file_position_before_kafka_offset(tidp, offset); + if (r == footer::npos) { + return std::nullopt; + } + return seek_result{.file_position = r.file_position, .length = r.length}; +} + +std::optional l1_footer_index::seek_to_timestamp( + model::topic_id_partition tidp, model::timestamp ts) const { + auto r = _footer.file_position_before_max_timestamp(tidp, ts); + if (r == footer::npos) { + return std::nullopt; + } + return seek_result{.file_position = r.file_position, .length = r.length}; +} + +l1_native_object_handle::l1_native_object_handle(footer f, fetch_range_fn fetch) + : _index(std::move(f)) + , _fetch(std::move(fetch)) {} + +ss::future, io::errc>> +l1_native_object_handle::open_reader( + const seek_result& seek, ss::abort_source* as) { + auto stream_result = co_await _fetch(seek.file_position, seek.length, as); + if (!stream_result.has_value()) { + co_return std::unexpected(stream_result.error()); + } + co_return object_reader::create(std::move(stream_result).value()); +} + +} // namespace cloud_topics::l1 diff --git a/src/v/cloud_topics/level_one/common/object_handle.h b/src/v/cloud_topics/level_one/common/object_handle.h new file mode 100644 index 0000000000000..fa1249701540b --- /dev/null +++ b/src/v/cloud_topics/level_one/common/object_handle.h @@ -0,0 +1,118 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + */ + +#pragma once + +#include "cloud_topics/level_one/common/abstract_io.h" +#include "cloud_topics/level_one/common/object.h" +#include "cloud_topics/level_one/common/object_id.h" +#include "model/fundamental.h" + +#include + +#include +#include +#include +#include + +namespace cloud_topics::l1 { + +/// Result of an index seek. `file_position` is always a byte offset within +/// the object. `delta`, when set, is the log-to-Kafka offset-translation delta +/// at `file_position`, letting the reader translate log offsets to Kafka +/// offsets directly rather than inferring the delta from the first batch; it is +/// `nullopt` when the index already reports positions in Kafka-offset space. +struct seek_result { + size_t file_position{0}; + /// Bytes from file_position to the end of the readable range. + /// Always non-zero for a valid seek. + size_t length{0}; + std::optional delta; +}; + +/// An index over a single L1 object. +/// +/// Abstracts seeking to a byte position by Kafka offset or timestamp, +/// independent of the object's on-disk index format. +class object_index { +public: + virtual ~object_index() = default; + + /// Return the seek position to start reading at or before the given + /// Kafka offset. Returns nullopt when the object has no data at or after + /// the offset. + virtual std::optional + seek_to_offset(model::topic_id_partition, kafka::offset) const = 0; + + /// Return the seek position for data with max_timestamp >= ts. + /// Returns nullopt when no matching data exists. + virtual std::optional + seek_to_timestamp(model::topic_id_partition, model::timestamp) const = 0; +}; + +/// An open reference to a single L1 object. +/// +/// Obtained from l1::open_object. Holds the object's index and can open readers +/// positioned at any seek result. +class object_handle { +public: + virtual ~object_handle() = default; + + /// Return a reference to the index for this object. + virtual const object_index& index() const = 0; + + /// Open a reader starting at the seek result returned by index(). + /// `seek.file_position` is always a byte offset; `seek.delta`, when set, + /// carries the offset delta at the seek point. + virtual ss::future, io::errc>> + open_reader(const seek_result& seek, ss::abort_source*) = 0; +}; + +/// Fetches a stream over an object's raw bytes [file_position, +/// file_position+length). +/// +/// Used by object_handle implementations to read from the underlying +/// extent. Copyable: the reader (and its data source's copy of this +/// fetch) outlives the handle, so it must not capture the handle. +using fetch_range_fn + = std::function, io::errc>>( + size_t file_position, size_t length, ss::abort_source*)>; + +/// object_index over a native L1 object's footer. +class l1_footer_index final : public object_index { +public: + explicit l1_footer_index(footer f); + + std::optional + seek_to_offset(model::topic_id_partition, kafka::offset) const override; + std::optional seek_to_timestamp( + model::topic_id_partition, model::timestamp) const override; + +private: + footer _footer; +}; + +/// object_handle for a native L1 object. Byte ranges are read through the +/// `fetch` callback supplied by open_object. +class l1_native_object_handle final : public object_handle { +public: + l1_native_object_handle(footer f, fetch_range_fn fetch); + + const object_index& index() const override { return _index; } + + ss::future, io::errc>> + open_reader(const seek_result& seek, ss::abort_source* as) override; + +private: + l1_footer_index _index; + fetch_range_fn _fetch; +}; + +} // namespace cloud_topics::l1 diff --git a/src/v/cloud_topics/level_one/common/object_id.h b/src/v/cloud_topics/level_one/common/object_id.h index 27cfcb50e2444..e5cd8f058750c 100644 --- a/src/v/cloud_topics/level_one/common/object_id.h +++ b/src/v/cloud_topics/level_one/common/object_id.h @@ -12,9 +12,17 @@ #pragma once #include "base/format_to.h" +#include "model/fundamental.h" +#include "serde/envelope.h" +#include "serde/rw/enum.h" #include "utils/named_type.h" #include "utils/uuid.h" +#include + +#include +#include + namespace cloud_topics::l1 { // An object ID is a unique identifier for a cloud topic L1 object. @@ -22,13 +30,102 @@ using object_id = named_type; inline object_id create_object_id() { return object_id{uuid_t::create()}; } -// An extent of a remote object, which is a pair of offset and size. +/// The cloud-storage object path of a tiered-storage segment imported into L1 +/// by reference (an opaque sname_format path). +using ts_segment_path = named_type; + +/// Whether an imported tiered-storage segment has an aborted-transaction (.tx) +/// manifest. Resolved once at import time from the source segment_meta so the +/// read path need not probe object storage to find out -- preserving a +/// capability native tiered storage already has (see remote_segment.cc): a v3 +/// segment records its .tx manifest size in metadata_size_hint (0 => none), and +/// a compacted segment has no aborted batches by construction. Only v1/v2 +/// non-compacted segments are genuinely unknowable without looking. +enum class tx_manifest_state : uint8_t { + /// Presence not encoded in the source metadata (v1/v2, non-compacted): the + /// read path probes object storage and tolerates a missing .tx. The + /// default, so a value that was never resolved degrades to the safe + /// (probe-and-tolerate) behavior rather than silently skipping a real .tx. + unknown = 0, + /// Known empty -- compacted (aborted batches removed by compaction) or v3 + /// with metadata_size_hint == 0. The read path skips the .tx download and + /// treats the segment as having no aborted transactions. + absent = 1, + /// Known present -- v3 with metadata_size_hint > 0. The read path downloads + /// the .tx. If it is unexpectedly missing (a lost/partial upload), the read + /// path warns but tolerates it (reads committed-only), matching tiered + /// storage, which treats a notfound .tx as empty. + present = 2, +}; + +/// The full descriptor of a tiered-storage segment imported into L1 by +/// reference, as the metastore *interface* and the IO read path see it: where +/// the segment lives plus how to interpret its data. +/// +/// The metastore *storage* layer decomposes this by ownership -- the path is an +/// object property and the delta/term are extent properties (see +/// `imported_ts_object_location` / `imported_ts_segment_info` in state.h) -- +/// but the API and the reader treat it as one unit. The extent's Kafka offset +/// bounds (carried separately by the API/IO types) give the segment's offset +/// range. +struct imported_ts_info + : public serde:: + envelope, serde::compat_version<0>> { + friend bool + operator==(const imported_ts_info&, const imported_ts_info&) = default; + auto serde_fields() { + return std::tie(ts_path, segment_term, delta_base, tx_state); + } + + /// Opaque tiered-storage segment object path (sname_format path). + ts_segment_path ts_path; + /// Raft term the segment was written in. Stamped onto every imported batch + /// as its partition leader epoch (a tiered-storage segment is single-term), + /// so a Kafka fetch of the imported region reports the original leader + /// epoch rather than -1. + model::term_id segment_term{}; + /// Offset-translation delta at the segment's base (the source + /// segment_meta's delta_offset: base log offset minus base Kafka offset). + /// Seeds the reader's running delta when a seek scans from the segment + /// start, so translation is correct even when compaction has removed the + /// leading records -- the first surviving batch may sit past the declared + /// base, and inferring the delta from it would renumber survivors down into + /// the hole. + model::offset_delta delta_base{}; + /// Whether the segment's aborted-transaction (.tx) manifest exists, + /// resolved at import time so the read path can skip the probe where the + /// answer is known. See tx_manifest_state. + tx_manifest_state tx_state{tx_manifest_state::unknown}; +}; + +/// An L1 object or imported TS segment, as seen by the IO read path. For an +/// imported extent `imported` is the segment descriptor (path + term + delta). +/// +/// `position` and `size` carry format-specific semantics: +/// native (imported == nullopt): +/// position = footer byte offset within the object +/// size = footer length in bytes (= object_size - footer_pos) +/// imported (imported.has_value()): +/// position = 0 (unused) +/// size = total segment size in bytes struct object_extent { object_id id; size_t position = 0; size_t size = 0; + /// Set for an imported extent (nullopt for native L1); carries the + /// segment's path, term, and offset-translation delta. + std::optional imported; fmt::iterator format_to(fmt::iterator it) const; }; +/// One object to delete via io::delete_objects. A native L1 object (ts_path +/// nullopt) is deleted by its id-derived path; an imported tiered-storage +/// segment (ts_path set) is deleted at that path (the segment plus its .tx and +/// .index). Both reside in the one configured object bucket. +struct object_location { + object_id id; + std::optional ts_path; +}; + } // namespace cloud_topics::l1 diff --git a/src/v/cloud_topics/level_one/common/open_object.cc b/src/v/cloud_topics/level_one/common/open_object.cc new file mode 100644 index 0000000000000..d5aa3aa946f8c --- /dev/null +++ b/src/v/cloud_topics/level_one/common/open_object.cc @@ -0,0 +1,176 @@ +/* + * Copyright 2026 Redpanda Data, Inc. + * + * Licensed as a Redpanda Enterprise file under the Redpanda Community + * License (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * https://github.com/redpanda-data/redpanda/blob/master/licenses/rcl.md + */ +#include "cloud_topics/level_one/common/open_object.h" + +#include "cloud_storage/offset_index.h" +#include "cloud_storage/remote_segment.h" +#include "cloud_topics/level_one/common/object.h" +#include "cloud_topics/level_one/common/ts_object.h" +#include "cloud_topics/logger.h" +#include "config/configuration.h" + +namespace cloud_topics::l1 { + +namespace { + +// Open an imported tiered-storage segment: build its index from the .index +// sidecar (io::fetch_ts_index), its aborted-transaction set from the .tx +// sidecar (io::fetch_ts_tx, gated by the import-time tx_state), and return a +// ts_object_handle whose byte transport reads chunks back through +// io::read_object (routed to the segment's ts_path). +ss::future, io::errc>> +open_imported_object( + io& io_impl, + const object_extent& extent, + ss::abort_source* as, + cloud_io::group_id g, + bool skip_cache) { + const auto& imported = *extent.imported; + + // Seek through the segment's real offset index: a present .index is + // deserialized; a missing one (notfound) leaves the index empty, so seeks + // fall back to a full-segment scan from position 0 with the segment base + // delta -- matching native tiered storage. Any other error propagates + // rather than being masked as an empty-index full scan. + cloud_storage::offset_index oi( + model::offset{0}, + kafka::offset{0}, + 0, + cloud_storage::remote_segment_sampling_step_bytes, + model::timestamp::missing()); + auto index_bytes = co_await io_impl.fetch_ts_index(extent, as); + if (index_bytes.has_value()) { + oi.from_iobuf(std::move(*index_bytes)); + } else if (index_bytes.error() != io::errc::cloud_missing_object) { + vlog( + cd_log.warn, + "Failed to fetch index for imported segment {}: {}", + imported.ts_path, + index_bytes.error()); + co_return std::unexpected(index_bytes.error()); + } + auto idx = std::make_unique( + std::move(oi), imported.delta_base, extent.size); + + // Aborted-transaction ranges for this segment, so the reader can strip + // aborted data and make the imported region committed-only (like native CT + // L1). Ranges are in raw log-offset space, as the reader needs. Whether to + // consult the .tx sidecar is decided by the import-time tx_state, which + // mirrors what native tiered storage knows without a probe: only v1/v2 + // non-compacted segments require one. + aborted_transactions aborted; + if (imported.tx_state != tx_manifest_state::absent) { + // present or unknown: consult the .tx sidecar. + auto tx = co_await io_impl.fetch_ts_tx(extent, as); + if (tx.has_value()) { + for (auto& r : *tx) { + aborted.insert(r); + } + } else if (tx.error() == io::errc::cloud_missing_object) { + // A missing .tx means no aborted transactions. Tolerate it and read + // the segment as committed-only, matching native tiered storage, + // which treats a notfound .tx as empty. When tx_state == present + // the source metadata recorded a .tx, so its absence is unexpected + // (a lost or partial upload) and we warn -- but as a compatibility + // layer we degrade gracefully rather than turning behavior tiered + // storage silently tolerated into a hard read failure. + if (imported.tx_state == tx_manifest_state::present) { + vlog( + cd_log.warn, + "Imported segment {} was expected to have a .tx manifest " + "(its " + "metadata recorded one) but it is missing; reading the " + "segment " + "as committed-only", + imported.ts_path); + } + } else { + // A transient/other error (not a definitive notfound): propagate so + // the read is retried, rather than silently dropping + // aborted-transaction filtering. + vlog( + cd_log.warn, + "Failed to fetch tx ranges for imported segment {}: {}", + imported.ts_path, + tx.error()); + co_return std::unexpected(tx.error()); + } + } + // tx_state == absent: known to have no aborted transactions (compacted, or + // v3 with an empty .tx manifest), so skip the sidecar entirely -- `aborted` + // stays empty. + + // Serve the segment as lazily-fetched fixed-size chunks so a read only + // downloads the chunks it touches (a read bounded by max_offset/max_bytes + // stops pulling and the tail is never fetched). chunk_size == 0 + // (cloud_storage_disable_chunk_reads) falls back to a single whole-suffix + // download. The fetch is invoked per chunk with a chunk-aligned (pos,len); + // the (pos,len)-keyed cache entry then dedups across reads that touch the + // same chunk. + const auto chunk_size + = config::shard_local_cfg().cloud_storage_disable_chunk_reads() + ? size_t{0} + : config::shard_local_cfg().cloud_storage_cache_chunk_size(); + co_return std::make_unique( + std::move(idx), + imported.segment_term, + std::move(aborted), + [io = &io_impl, imported, g, skip_cache]( + size_t pos, size_t len, ss::abort_source* fetch_as) + -> ss::future, io::errc>> { + // Read the chunk through the uniform read_object path: an imported + // extent routes it to the segment's ts_path (with cache/skip + // handling), same as any other L1 read. + return io->read_object( + object_extent{.position = pos, .size = len, .imported = imported}, + fetch_as, + g, + skip_cache); + }, + chunk_size); +} + +} // namespace + +ss::future, io::errc>> open_object( + io& io_impl, + object_extent extent, + ss::abort_source* as, + cloud_io::group_id g, + bool skip_cache) { + if (extent.imported.has_value()) { + co_return co_await open_imported_object( + io_impl, extent, as, g, skip_cache); + } + // Native L1 object: read and parse the footer, then hand the handle a fetch + // that reads data ranges back through io::read_object. + auto footer_buf = co_await io_impl.fetch_native_footer( + extent, as, g, skip_cache); + if (!footer_buf.has_value()) { + co_return std::unexpected(footer_buf.error()); + } + auto footer_result = co_await footer::read(std::move(footer_buf).value()); + if (!std::holds_alternative