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_storage/tests/cloud_storage_e2e_test.cc b/src/v/cloud_storage/tests/cloud_storage_e2e_test.cc index 04dfd77ab7046..fb8f0f8ea0631 100644 --- a/src/v/cloud_storage/tests/cloud_storage_e2e_test.cc +++ b/src/v/cloud_storage/tests/cloud_storage_e2e_test.cc @@ -18,6 +18,7 @@ #include "cluster/controller_api.h" #include "cluster/health_monitor_frontend.h" #include "kafka/client/transport.h" +#include "kafka/data/partition_proxy.h" #include "kafka/data/replicated_partition.h" #include "kafka/protocol/fetch.h" #include "kafka/protocol/find_coordinator.h" @@ -198,6 +199,166 @@ TEST_F(ManualFixture, TestSizeEstimationWithCloud) { } } +// A partition mid tiered->cloud migration is served transparently as tiered +// storage. The migration is entered by the operator flipping the topic's +// storage mode to cloud (topic_mode == cloud), but the durable partition_mode +// lags at tiered until cutover, and the serving accessors key on partition_mode +// -- so cloud_topic_enabled()/is_remote_fetch_enabled() still report the tiered +// values and nothing about the read path changes. This exercises the full +// replicated_partition API surface plus make_partition_proxy routing and +// asserts byte-for-byte parity with the same partition before the flip: +// offsets, size estimates, and timequery are unchanged, prefix truncation still +// works, and the partition still routes to the replicated_partition path. +// Regression for the timequery corner -- it must consult the tiered/imported +// prefix, not the local-log start, throughout. +TEST_F(ManualFixture, MigratingPartitionServedAsTieredStorage) { + test_local_cfg.get("cloud_storage_disable_upload_loop_for_tests") + .set_value(true); + test_local_cfg.get("cloud_storage_spillover_manifest_max_segments") + .set_value(std::make_optional(5)); + test_local_cfg.get("cloud_storage_spillover_manifest_size") + .set_value(std::optional{}); + const model::topic topic_name("migrating-rp"); + model::ntp ntp(model::kafka_namespace, topic_name, 0); + + cluster::topic_properties props; + props.shadow_indexing = model::shadow_indexing_mode::full; + props.cleanup_policy_bitflags = model::cleanup_policy_bitflags::deletion; + add_topic({model::kafka_namespace, topic_name}, 1, props).get(); + wait_for_leader(ntp).get(); + + auto partition = app.partition_manager.local().get(ntp); + auto& archiver = partition->archiver().value().get(); + archiver.initialize_probe(); + tests::remote_segment_generator gen(make_kafka_client().get(), *partition); + auto deferred_g_close = ss::defer([&gen] { gen.stop().get(); }); + auto total_records = gen.num_segments(40) + .batches_per_segment(5) + .additional_local_segments(10) + .produce() + .get(); + ASSERT_GE(total_records, 200); + ASSERT_TRUE(archiver.sync_for_tests().get()); + archiver.apply_spillover().get(); + + // Keep some segments local and the prefix only in tiered storage, so the + // API spans both regions (cloud prefix + local suffix) and the parity check + // is meaningful. + auto log = partition->log(); + auto& manifest = partition->archival_meta_stm()->manifest(); + log->set_cloud_gc_offset(model::next_offset(manifest.get_last_offset())); + // Wait for GC to trim the local prefix so it lives only in tiered storage + // (the local log start advances above 0). Reads of the prefix must then + // come from the cloud, which is what makes the migrating-parity check below + // meaningful (and is the condition the timequery bug needed to manifest). + RPTEST_REQUIRE_EVENTUALLY( + 20s, [&] { return log->offsets().start_offset > model::offset(0); }); + + struct snapshot { + model::offset start; + model::offset hwm; + model::offset local_start; + model::offset lso; + kafka::leader_epoch epoch; + model::offset offset_lag; + size_t total_sz; + size_t cloud_sz; + std::optional tq_earliest; + }; + auto capture = [](kafka::replicated_partition& rp) { + auto lso = rp.last_stable_offset(); + EXPECT_FALSE(lso.has_error()); + auto last = kafka::prev_offset(model::offset_cast(lso.value())); + auto local_start = model::offset_cast(rp.local_start_offset()); + // timequery for the epoch resolves to the earliest available offset; it + // must consult the tiered/imported prefix, not just the local log. + storage::timequery_config tq_cfg{ + rp.start_offset(), + model::timestamp(0), + rp.high_watermark(), + {model::record_batch_type::raft_data}, + std::nullopt}; + auto tq = rp.timequery(tq_cfg).get(); + return snapshot{ + .start = rp.start_offset(), + .hwm = rp.high_watermark(), + .local_start = rp.local_start_offset(), + .lso = lso.value(), + .epoch = rp.leader_epoch(), + .offset_lag = rp.offset_lag(), + .total_sz = rp.estimate_size_between(kafka::offset(0), last), + .cloud_sz = rp.estimate_size_between( + kafka::offset(0), kafka::prev_offset(local_start)), + .tq_earliest = tq.has_value() + ? std::optional(tq->offset) + : std::nullopt, + }; + }; + + kafka::replicated_partition rp_tiered(partition); + auto tiered = capture(rp_tiered); + // The prefix really is cloud-only (local log starts above 0), so timequery + // for the epoch must reach into tiered storage to answer 0. + ASSERT_GT(tiered.local_start, model::offset(0)); + ASSERT_EQ( + tiered.tq_earliest, std::optional(model::offset(0))); + + // Enter the migrating state: the operator flips the topic's storage mode to + // cloud (topic_mode), while the durable partition_mode lags at tiered until + // cutover. Serving accessors key on partition_mode, so + // cloud_topic_enabled() stays false and the partition keeps being served as + // tiered storage. + ASSERT_TRUE(log->config().has_overrides()); + auto overrides = log->config().get_overrides(); + overrides.storage_mode = model::redpanda_storage_mode::cloud; + log->set_overrides(overrides); + log->set_partition_mode(model::redpanda_storage_mode::tiered); + ASSERT_EQ(log->config().topic_mode(), model::redpanda_storage_mode::cloud); + ASSERT_EQ( + log->config().partition_mode(), model::redpanda_storage_mode::tiered); + ASSERT_FALSE(log->config().cloud_topic_enabled()); + + kafka::replicated_partition rp_mig(partition); + auto migrating = capture(rp_mig); + + // Transparent migration: every read/size API returns exactly what it did as + // plain tiered storage. + EXPECT_EQ(tiered.start, migrating.start); + EXPECT_EQ(tiered.hwm, migrating.hwm); + EXPECT_EQ(tiered.local_start, migrating.local_start); + EXPECT_EQ(tiered.lso, migrating.lso); + EXPECT_EQ(tiered.epoch, migrating.epoch); + EXPECT_EQ(tiered.offset_lag, migrating.offset_lag); + // Entering the migrating state writes nothing to the log (it only adjusts + // the ntp_config overrides), so every size estimate is exactly unchanged. + EXPECT_EQ(tiered.total_sz, migrating.total_sz); + EXPECT_EQ(tiered.cloud_sz, migrating.cloud_sz); + // Regression: timequery still resolves to offset 0 via the imported extents + // (not the local-log start) while migrating. + EXPECT_EQ(tiered.tq_earliest, migrating.tq_earliest); + EXPECT_EQ( + migrating.tq_earliest, std::optional(model::offset(0))); + + // make_partition_proxy routes the migrating partition to the tiered-storage + // (replicated_partition) path via the standard cloud_topic_enabled()==false + // branch -- partition_mode lags at tiered, so no manifest-specific routing + // gate is needed. It must behave identically to a directly-constructed + // replicated_partition, not the (empty) cloud-topic path. + kafka::partition_proxy proxy = kafka::make_partition_proxy(partition); + EXPECT_EQ(proxy.start_offset(), rp_mig.start_offset()); + EXPECT_EQ(proxy.high_watermark(), rp_mig.high_watermark()); + EXPECT_GT(proxy.high_watermark(), model::offset(0)); + + // DeleteRecords / prefix truncation works while migrating and advances the + // start offset (eviction STM + archival manifest head-prune path). + auto trunc = model::offset(total_records / 2); + auto tec + = rp_mig.prefix_truncate(trunc, ss::lowres_clock::now() + 30s).get(); + ASSERT_EQ(tec, kafka::error_code::none); + RPTEST_REQUIRE_EVENTUALLY( + 10s, [&] { return rp_mig.start_offset() >= trunc; }); +} + class EndToEndFixture : public s3_imposter_fixture , public manual_metadata_upload_mixin diff --git a/src/v/cloud_topics/BUILD b/src/v/cloud_topics/BUILD index 164dfb57a69e2..e0afa66bd51d2 100644 --- a/src/v/cloud_topics/BUILD +++ b/src/v/cloud_topics/BUILD @@ -141,6 +141,7 @@ redpanda_cc_library( "//src/v/cloud_topics/level_one/metastore:replicated_metastore", "//src/v/cloud_topics/level_zero/cluster_services_impl", "//src/v/cloud_topics/level_zero/common:extent_meta", + "//src/v/cloud_topics/migration:metastore_sink", "//src/v/cloud_topics/reconciler", "//src/v/cluster", "//src/v/cluster:cluster_link_table", diff --git a/src/v/cloud_topics/app.cc b/src/v/cloud_topics/app.cc index ecfb5414ce2e4..6463e1a509830 100644 --- a/src/v/cloud_topics/app.cc +++ b/src/v/cloud_topics/app.cc @@ -25,7 +25,9 @@ #include "cloud_topics/read_replica/snapshot_manager.h" #include "cloud_topics/reconciler/reconciler.h" #include "cloud_topics/topic_manifest_upload_manager.h" +#include "cluster/archival/archival_metadata_stm.h" #include "cluster/controller.h" +#include "cluster/partition.h" #include "cluster/utils/partition_change_notifier_impl.h" #include "config/configuration.h" #include "config/node_config.h" @@ -117,6 +119,19 @@ ss::future<> app::construct( ss::sharded_parameter([&remote] { return std::ref(remote->local()); }), bucket); + // The migration mirror runs in the archiver (cluster) but writes to the L1 + // metastore (here); inject the sink so the archiver can reach it without + // cluster depending on cloud_topics. + co_await construct_service( + migration_sink, + ss::sharded_parameter([this] { return &replicated_metastore.local(); }), + ss::sharded_parameter( + [&metadata_cache] { return &metadata_cache->local(); })); + co_await controller->get_partition_manager().invoke_on_all( + [this](cluster::partition_manager& pm) { + pm.set_migration_metastore(&migration_sink.local()); + }); + co_await construct_service( rr_snapshot_manager_, config::node().l1_staging_path(), @@ -306,6 +321,18 @@ ss::future<> app::wire_up_notifications() { const model::topic_id_partition& tidp, auto partition) noexcept { if (partition) { + // Attach every leader partition whose topic config is a cloud + // topic, including ones mid tiered->cloud migration (config + // flips to cloud at the migration trigger, before cutover). + // While the partition is served as tiered storage + // (partition_mode lags at tiered) the reconciler skips it + // each round (!source::is_cloud_topic()), so the archiver's + // mirror is the sole L1 writer; once cutover advances + // partition_mode to cloud the reconciler picks it up on the + // next round without a re-attach. Leadership notifications do + // not fire on cutover, so a per-round skip -- not an + // attach-time gate -- is what makes the transition take + // effect. r.attach_partition( ntp, tidp, data_plane.get(), std::move(*partition)); } else { @@ -406,6 +433,12 @@ ss::future<> app::cleanup_tmp_files() { } ss::future<> app::stop() { + // shutdown() blocks (ss::future::get) and so must run in the ss::thread + // context the caller invokes stop() from -- i.e. before the first co_await. + // The migration sink need not be deregistered from the partition manager + // here: the partition manager (and the archivers that hold the sink) is + // torn down before cloud_topics::app, so the sink always outlives its + // users. ssx::sharded_service_container::shutdown(); co_await data_plane->stop(); } diff --git a/src/v/cloud_topics/app.h b/src/v/cloud_topics/app.h index d25aef6b67f74..c97f6db05c0c1 100644 --- a/src/v/cloud_topics/app.h +++ b/src/v/cloud_topics/app.h @@ -18,6 +18,7 @@ #include "cloud_topics/level_one/metastore/leader_router.h" #include "cloud_topics/level_one/metastore/replicated_metastore.h" #include "cloud_topics/level_zero/cluster_services_impl/cluster_services.h" +#include "cloud_topics/migration/metastore_sink.h" #include "cloud_topics/reconciler/reconciler.h" #include "cloud_topics/state_accessors.h" #include "ssx/sharded_service_container.h" @@ -110,6 +111,7 @@ class app : public ssx::sharded_service_container { ss::sharded state; ss::sharded l1_io; ss::sharded replicated_metastore; + ss::sharded migration_sink; ss::sharded> reconciler; ss::sharded domain_supervisor; ss::sharded l1_metastore_router; diff --git a/src/v/cloud_topics/housekeeper/housekeeper.cc b/src/v/cloud_topics/housekeeper/housekeeper.cc index eeb04989f659a..fdbd2e888b78f 100644 --- a/src/v/cloud_topics/housekeeper/housekeeper.cc +++ b/src/v/cloud_topics/housekeeper/housekeeper.cc @@ -116,6 +116,21 @@ ss::future<> housekeeper::do_loop() { simple_time_jitter jitter(_loop_interval()); co_await ss::sleep_abortable(jitter.next_duration(), _as); try { + // CT housekeeping runs only on cloud topics. A partition served as + // tiered storage -- plain tiered, or still migrating tiered->cloud + // (partition_mode lags at tiered until cutover) -- has an idle ctp_stm; + // running housekeeping would force epoch/placeholder maintenance that + // seeds a meaningless reconciled offset and pins local-log GC, freezing + // retention. Skip until it becomes a cloud topic (cutover), which is + // also where the reconciler picks it up. + if (!_l0_metastore->is_cloud_topic(_tidp)) { + vlog( + cd_log.trace, + "{}: not a cloud topic (served as tiered storage), skipping " + "housekeeping", + _tidp); + co_return; + } co_await do_housekeeping(); co_await do_bump_epoch(); } catch (...) { diff --git a/src/v/cloud_topics/housekeeper/housekeeper.h b/src/v/cloud_topics/housekeeper/housekeeper.h index 51f8776ee1de0..bc8d776952621 100644 --- a/src/v/cloud_topics/housekeeper/housekeeper.h +++ b/src/v/cloud_topics/housekeeper/housekeeper.h @@ -80,6 +80,17 @@ class housekeeper { virtual ss::future<> sync_to_next_placeholder( const model::topic_id_partition& tidp, ss::abort_source*) noexcept = 0; + + // Whether the partition is a cloud topic that CT housekeeping should + // run against. False for a partition served as tiered storage: a plain + // tiered partition (which carries a pre-installed idle ctp_stm) or one + // still migrating tiered->cloud (partition_mode lags at tiered, so + // cloud_topic_enabled() is false until cutover). Its ctp_stm is idle + // and housekeeping would seed a meaningless reconciled offset and pin + // GC. + virtual bool is_cloud_topic(const model::topic_id_partition&) { + return true; + } }; // A wrapper around a source of configuration for a give topic id + @@ -123,8 +134,13 @@ class housekeeper { ss::future<> do_bump_epoch(); -private: + // Run a single iteration of the housekeeping loop (sleep + the migration + // guard + do_housekeeping + do_bump_epoch). + // + // Public for testing. ss::future<> do_loop(); + +private: ss::future do_bytes_retention(size_t size); ss::future do_time_retention(std::chrono::milliseconds); // Syncs the start offset from L0 metadata storage to L1 metastore. diff --git a/src/v/cloud_topics/housekeeper/manager.cc b/src/v/cloud_topics/housekeeper/manager.cc index 53216f41f0a91..0672a2cc66f63 100644 --- a/src/v/cloud_topics/housekeeper/manager.cc +++ b/src/v/cloud_topics/housekeeper/manager.cc @@ -125,6 +125,11 @@ class l0_metastore_impl : public housekeeper::l0_metadata_storage { co_return; } + bool is_cloud_topic(const model::topic_id_partition& tidp) override { + auto& state = _state->at(tidp); + return state.partition->get_ntp_config().cloud_topic_enabled(); + } + private: ctp_stm_api get_api(const model::topic_id_partition& tidp) { auto& state = _state->at(tidp); diff --git a/src/v/cloud_topics/housekeeper/tests/housekeeper_test.cc b/src/v/cloud_topics/housekeeper/tests/housekeeper_test.cc index 065f642a9d44c..21012c491d050 100644 --- a/src/v/cloud_topics/housekeeper/tests/housekeeper_test.cc +++ b/src/v/cloud_topics/housekeeper/tests/housekeeper_test.cc @@ -95,6 +95,12 @@ class fake_l0_metastore co_return; } + bool is_cloud_topic(const model::topic_id_partition&) override { + return _is_cloud_topic; + } + + void set_is_cloud_topic(bool v) { _is_cloud_topic = v; } + void set_max_allowed_start_offset(kafka::offset offset) { _max_allowed_start_offset = offset; } @@ -130,6 +136,8 @@ class fake_l0_metastore kafka::offset _start_offset; kafka::offset _max_allowed_start_offset; + bool _is_cloud_topic{true}; + // Epoch-related state std::optional _estimated_inactive_epoch = cloud_topics::cluster_epoch::min(); @@ -299,6 +307,8 @@ class HousekeeperTest : public testing::Test { void reset_epoch_call_tracking() { _l0_metastore.reset_call_tracking(); } + void set_is_cloud_topic(bool v) { _l0_metastore.set_is_cloud_topic(v); } + private: model::topic_id_partition _tidp{ model::create_topic_id(), model::partition_id{0}}; @@ -788,6 +798,37 @@ TEST_F(HousekeeperTest, BumpEpochIdleTriggersAdvance) { EXPECT_EQ(sync_to_next_placeholder_calls(), 1); } +TEST_F(HousekeeperTest, MigratingPartitionSkipsHousekeeping) { + // A partition still migrating from tiered storage is not yet a cloud topic + // (partition_mode lags at tiered), is served from TS, and its ctp_stm is + // idle; CT housekeeping must not run against it -- do_bump_epoch would seed + // a meaningless reconciled offset and pin local-log GC. do_loop gates on + // is_cloud_topic(). + auto housekeeper = make_housekeeper({}); + + // The idle scenario that WOULD force an epoch advance + sync on a cloud + // topic (cf. BumpEpochIdleTriggersAdvance): a stable inactive epoch. + set_estimated_inactive_epoch(cloud_topics::cluster_epoch{1}); + set_current_cluster_epoch(cloud_topics::cluster_epoch{5}); + set_is_cloud_topic(false); + + // Even the second iteration (which forces advance + sync on a cloud topic) + // must be skipped entirely while served as tiered storage. + housekeeper.do_loop().get(); + housekeeper.do_loop().get(); + EXPECT_TRUE(advance_epoch_calls().empty()); + EXPECT_EQ(sync_to_next_placeholder_calls(), 0); + + // Control: once it cuts over to a cloud topic, the same idle scenario + // advances. + set_is_cloud_topic(true); + reset_epoch_call_tracking(); + housekeeper.do_loop().get(); // re-initializes _last_epoch + housekeeper.do_loop().get(); // idle -> forces advance + sync + ASSERT_EQ(advance_epoch_calls().size(), 1); + EXPECT_EQ(sync_to_next_placeholder_calls(), 1); +} + TEST_F(HousekeeperTest, BumpEpochGetCurrentEpochReturnsNullopt) { // When get_current_cluster_epoch returns nullopt, we should not call // advance_epoch even if the partition is idle. diff --git a/src/v/cloud_topics/level_one/common/BUILD b/src/v/cloud_topics/level_one/common/BUILD index c1241d3773856..192693daf9953 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 = [ + ":chunk_data_source", + ":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", + ":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..facc3aebb40ea --- /dev/null +++ b/src/v/cloud_topics/level_one/common/chunk_data_source.h @@ -0,0 +1,164 @@ +/* + * 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 +#include + +namespace cloud_topics::l1 { + +/// Controls where the first chunk starts. When yes, chunk boundaries are +/// chunk_size aligned, so overlapping reads request identical +/// (position, length) chunks that a downstream cache can dedup. When no, the +/// first chunk starts exactly at start_pos -- for a caller with no cache to +/// dedup against, which would otherwise fetch and discard the bytes between the +/// chunk boundary and start_pos. +using use_chunk_aligned_reads + = ss::bool_class; + +/// 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: + /// See use_chunk_aligned_reads. When yes the first fetch is aligned + /// at or before start_pos and the leading `start_pos % chunk_size` bytes + /// are skipped; when no the first chunk starts exactly at start_pos with no + /// skipped head. + chunk_data_source( + Fetch fetch, + size_t start_pos, + size_t total_len, + size_t chunk_size, + ss::abort_source* as, + use_chunk_aligned_reads aligned) + : _fetch(std::move(fetch)) + , _end(start_pos + total_len) + , _chunk_size(chunk_size) + , _next_chunk_start( + aligned ? start_pos - (start_pos % chunk_size) : start_pos) + , _skip_head(aligned ? start_pos % chunk_size : 0) + , _as(as) { + vassert( + chunk_size > 0, "chunk_data_source requires a non-zero chunk size"); + } + + ss::future> get() override { + while (true) { + // Surface an abort as an exception on every read -- including while + // draining a buffered chunk or between chunks -- so a cancelled + // read fails loudly rather than terminating as a (truncated) clean + // end-of-stream. + if (_as != nullptr) { + _as->check(); + } + 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. When chunk-aligned this is the chunk + // boundary at or before start_pos; otherwise it is start_pos itself. + size_t _next_chunk_start; + // Bytes to discard from the first chunk so the stream starts at start_pos + // (non-zero only when chunk-aligned and start_pos sits 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, + use_chunk_aligned_reads aligned) { + return ss::data_source{std::make_unique>( + std::move(fetch), start_pos, total_len, chunk_size, as, aligned)}; +} + +} // 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..833e2f6d51b30 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 { @@ -116,8 +118,27 @@ fake_io::read_object( ss::abort_source*, [[maybe_unused]] cloud_io::group_id gid, // 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) { + // effectively a streaming read regardless of this flag. It is still recorded + // so tests can assert how open_object chunks a read and whether it bypasses + // the cache. + bool skip_cache) { + _read_object_calls.push_back( + read_object_call{ + .position = extent.position, + .size = extent.size, + .skip_cache = skip_cache, + .imported = extent.imported.has_value()}); + // 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 +149,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..135c38abf3f8f 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,10 @@ #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 +#include namespace cloud_topics::l1 { @@ -32,8 +36,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 +61,47 @@ 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); + + /// A recorded read_object call, so tests can assert how open_object chunks + /// a read (chunk size / chunk alignment) and whether it bypasses the cache. + struct read_object_call { + size_t position; + size_t size; + bool skip_cache; + bool imported; + }; + const std::vector& read_object_calls() const { + return _read_object_calls; + } + 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; + std::vector _read_object_calls; }; } // 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..30907d3366c06 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" @@ -20,7 +24,6 @@ #include "config/configuration.h" #include -#include #include #include #include @@ -93,120 +96,45 @@ struct one_time_stream_provider : public stream_provider { std::optional> _st; }; -// A streaming download source for L1 objects that chunks downloads per -// `chunk_size`. This may result in many GET requests per object download. -// -// The object's byte range is read one bounded chunk at a time. `get()` fully -// buffers the next chunk via `download_stream` (a separate ranged GET), which -// returns the lease to the pool as soon as the chunk's body has been read off -// the connection. The buffered chunk is then served from memory, lease-free, -// while subsequent `get()`s drain it; only once it is exhausted is the next -// chunk fetched. Peak in-memory bytes and the connection-hold duration per read -// are both bounded by `chunk_size`. -class streaming_download_source final : public ss::data_source_impl { -public: - // chunk_size bounds both the peak in-memory bytes held per read and the - // span of object bytes downloaded under a single held lease. _next_pos and - // _last_pos are initialized to inclusive byte ranges. - streaming_download_source( - cloud_io::remote* remote, - cloud_storage_clients::bucket_name bucket, - cloud_storage_clients::object_key key, - cloud_storage_clients::http_byte_range range, - ss::abort_source& as, - cloud_io::group_id gid, - size_t chunk_size) - : _remote(remote) - , _bucket(std::move(bucket)) - , _key(std::move(key)) - , _next_pos(range.first) - , _last_pos(range.second) - , _as(as) - , _gid(gid) - , _chunk_size(chunk_size) {} - - streaming_download_source(const streaming_download_source&) = delete; - streaming_download_source& - operator=(const streaming_download_source&) = delete; - streaming_download_source(streaming_download_source&&) = delete; - streaming_download_source& operator=(streaming_download_source&&) = delete; - ~streaming_download_source() override = default; - - ss::future> get() override { - while (true) { - _as.check(); - if (!_current.empty()) { - auto buf = std::move(_current.front()); - _current.pop_front(); - co_return buf; - } - if (_next_pos > _last_pos) { - // An empty buffer signals end-of-stream. - co_return ss::temporary_buffer(); - } - co_await download_next_chunk(); - } - } - - ss::future<> close() override { return ss::now(); } - -private: - // Fetches the next chunk of the object's byte range into `_current`. The - // lease is released when `download_stream` returns, before the buffered - // chunk is served. - ss::future<> download_next_chunk() { - static constexpr auto timeout = 10s; - static constexpr auto backoff = 100ms; - const auto chunk_first = _next_pos; - const auto chunk_last = std::min( - chunk_first + _chunk_size - 1, _last_pos); - _next_pos = chunk_last + 1; - retry_chain_node root(_as, ss::lowres_clock::now() + timeout, backoff); - - cloud_io::try_consume_stream consumer = - [this](uint64_t /*content_length*/, ss::input_stream stream) { - return drain_chunk(std::move(stream)); - }; - - auto result_fut = co_await ss::coroutine::as_future( - _remote->download_stream( - cloud_io::transfer_details{ - .bucket = _bucket, - .key = _key, - .parent_rtc = root, - }, - consumer, - "l1_stream_download", - /*acquire_hydration_units=*/true, - cloud_storage_clients::http_byte_range{chunk_first, chunk_last}, - {}, - _gid)); - if (result_fut.failed()) { - std::rethrow_exception(result_fut.get_exception()); - } - auto result = result_fut.get(); - if (result != cloud_io::download_result::success) { - throw std::runtime_error( - fmt::format("L1 streaming download failed: {}", result)); - } - } +// Cache-bypassing ranged download of the object byte range [pos, pos+len) at +// `key`, fully buffered into an input_stream. `download_stream` (a single +// ranged GET) returns the lease to the pool as soon as the range body has been +// read off the connection, before the buffered bytes are served -- so both the +// peak in-memory bytes and the connection-hold duration are bounded by the +// range length. Backs read_object's skip_cache path without populating the +// cloud cache; the range is bounded because open_object invokes read_object +// once per chunk. A download failure throws (rethrowing the original +// exception, or a runtime_error for a non-success result), surfaced from the +// caller's read of the returned stream. +ss::future, io::errc>> +download_range_bypassing_cache( + cloud_io::remote* remote, + const cloud_storage_clients::bucket_name& bucket, + const cloud_storage_clients::object_key& key, + cloud_io::group_id gid, + size_t pos, + size_t len, + 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); - // Reads a chunk's whole body into `_current`. Cleared up-front so that a - // retried download re-buffers from scratch rather than appending to a - // partially-read result. - ss::future drain_chunk(ss::input_stream stream) { - _current.clear(); + iobuf buf; + cloud_io::try_consume_stream consumer = + [&buf, as]( + uint64_t /*content_length*/, + ss::input_stream stream) -> ss::future { uint64_t total = 0; std::exception_ptr ex; try { while (true) { - _as.check(); - auto buf = co_await stream.read(); - if (buf.empty()) { + as->check(); + auto b = co_await stream.read(); + if (b.empty()) { break; } - total += buf.size(); - _current.push_back(std::move(buf)); + total += b.size(); + buf.append(std::move(b)); } } catch (...) { ex = std::current_exception(); @@ -216,18 +144,70 @@ class streaming_download_source final : public ss::data_source_impl { std::rethrow_exception(ex); } co_return total; + }; + + auto result_fut = co_await ss::coroutine::as_future(remote->download_stream( + cloud_io::transfer_details{ + .bucket = bucket, + .key = key, + .parent_rtc = root, + }, + consumer, + "l1_stream_download", + /*acquire_hydration_units=*/true, + cloud_storage_clients::http_byte_range{pos, pos + len - 1}, + {}, + gid)); + if (result_fut.failed()) { + std::rethrow_exception(result_fut.get_exception()); + } + auto result = result_fut.get(); + if (result != cloud_io::download_result::success) { + throw std::runtime_error( + fmt::format("L1 streaming download failed: {}", result)); } + co_return make_iobuf_input_stream(std::move(buf)); +} - cloud_io::remote* _remote; - cloud_storage_clients::bucket_name _bucket; - cloud_storage_clients::object_key _key; - size_t _next_pos; - size_t _last_pos; - ss::abort_source& _as; - cloud_io::group_id _gid; - size_t _chunk_size; - 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 @@ -319,21 +299,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 +331,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 +383,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( @@ -424,27 +429,21 @@ file_io::read_object( } if (skip_cache) { - // Not in the cache: stream directly from object storage in bounded - // chunks without populating the cache. - co_return ss::input_stream(ss::data_source( - 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}, - *as, - gid, - config::shard_local_cfg() - .cloud_topics_l1_streaming_read_chunk_size()))); + // Not in the cache: serve the requested range directly from object + // storage as a single ranged GET, without populating the cache. Any + // chunking that bounds peak memory is applied a layer up in + // open_object, which invokes read_object once per chunk. + co_return co_await download_range_bypassing_cache( + _remote, _bucket, key, gid, extent.position, extent.size, as); } // single_flight dedups concurrent downloads for this extent. 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 +462,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 +536,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..42ce81a5c41c5 --- /dev/null +++ b/src/v/cloud_topics/level_one/common/object_handle.h @@ -0,0 +1,121 @@ +/* + * 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. +/// The handle just streams whatever this returns; any chunking (bounding peak +/// memory / cache-chunk granularity) is baked into the fetch by open_object +/// when it builds the handle, not applied by the handle itself. 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..4f1458c575fde --- /dev/null +++ b/src/v/cloud_topics/level_one/common/open_object.cc @@ -0,0 +1,225 @@ +/* + * 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/chunk_data_source.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 { + +// Chunking for a read, given the object format and whether the cache is +// bypassed. Applied by wrapping the per-range read_object fetch in a +// chunk_data_source; chunk_size == 0 means "don't chunk" -- pass the whole +// requested range straight through. +struct read_chunking { + size_t chunk_size; + use_chunk_aligned_reads aligned; +}; + +read_chunking chunking_for(bool imported, bool skip_cache) { + const auto& cfg = config::shard_local_cfg(); + if (skip_cache) { + // Cache-bypassing bulk/one-shot read (leveling, compaction): bound peak + // memory with the streaming chunk size. Nothing is cached, so there is + // no (pos,size) key to align to -- chunks start at the read's position. + return { + cfg.cloud_topics_l1_streaming_read_chunk_size(), + use_chunk_aligned_reads::no}; + } + if (imported) { + // Cached imported read: chunk at the cloud-storage cache-chunk + // granularity, chunk-aligned so overlapping reads request identical + // (pos,size) chunks that the cache dedups. Disabling chunk reads falls + // back to a single whole-suffix cache entry. + if (cfg.cloud_storage_disable_chunk_reads()) { + return {0, use_chunk_aligned_reads::yes}; + } + return { + cfg.cloud_storage_cache_chunk_size(), use_chunk_aligned_reads::yes}; + } + // Cached native read: read_object stores the whole requested extent in the + // cache and serves it from a file, so the cache file bounds memory and no + // stream-level chunking is needed. + return {0, use_chunk_aligned_reads::yes}; +} + +// Wrap a per-range fetch so a request for [pos, pos+len) is served as a +// chunk_data_source that lazily pulls fixed-size chunks (only the chunks a read +// touches are fetched). chunk_size == 0 passes the whole range straight through +// (the cached-native / chunk-reads-disabled case). +fetch_range_fn chunked_fetch(fetch_range_fn inner, read_chunking c) { + if (c.chunk_size == 0) { + return inner; + } + return [inner = std::move(inner), + c](size_t pos, size_t len, ss::abort_source* as) + -> ss::future, io::errc>> { + co_return ss::input_stream{ + make_chunk_data_source(inner, pos, len, c.chunk_size, as, c.aligned)}; + }; +} + +// 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. + + // The per-range transport: read one byte range 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). chunked_fetch then + // layers the read's chunking on top so a read only downloads the chunks it + // touches. + fetch_range_fn inner = [io = &io_impl, imported, g, skip_cache]( + size_t pos, size_t len, ss::abort_source* fetch_as) + -> ss::future, io::errc>> { + return io->read_object( + object_extent{.position = pos, .size = len, .imported = imported}, + fetch_as, + g, + skip_cache); + }; + co_return std::make_unique( + std::move(idx), + imported.segment_term, + std::move(aborted), + chunked_fetch( + std::move(inner), chunking_for(/*imported=*/true, skip_cache))); +} + +} // 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, with the read's + // chunking layered on by chunked_fetch. + 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