Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/v/cloud_topics/level_zero/stm/placeholder.cc
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,8 @@ model::record_batch encode_placeholder_batch(
builder.add_raw_kv(std::nullopt, std::nullopt);
}

auto ph = std::move(builder).build();
auto ph = std::move(builder).build(
storage::record_batch_builder::reset_size_checksum::no);
// In order for timequeries to work correctly, we need to ensure we never
// look inside the batch to answer the query. If the time is append time we
// don't need to unpack the batch, but instead all records have the
Expand Down
11 changes: 9 additions & 2 deletions src/v/storage/record_batch_builder.cc
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
#include "storage/record_batch_builder.h"

#include "model/batch_compression.h"
#include "model/compression.h"
#include "model/record.h"
#include "model/record_utils.h"
#include "model/timeout_clock.h"
Expand Down Expand Up @@ -56,15 +57,21 @@ record_batch_builder& record_batch_builder::add_raw_kw(
return *this;
}

model::record_batch record_batch_builder::build() && {
model::record_batch record_batch_builder::build(reset_size_checksum reset) && {
vassert(
reset == reset_size_checksum::yes
|| _compression == model::compression::none,
"reset_size::checksum::no is only valid for compression::none");
if (!_timestamp) {
_timestamp = model::timestamp::now();
}
auto header = build_header();
auto batch = model::record_batch(
header, std::move(_records), model::record_batch::tag_ctor_ng{});
if (_compression == model::compression::none) {
batch.header().reset_size_checksum_metadata(batch.data());
if (reset == reset_size_checksum::yes) {
batch.header().reset_size_checksum_metadata(batch.data());
}
return batch;
}
return model::compress_batch_sync(_compression, std::move(batch));
Expand Down
11 changes: 10 additions & 1 deletion src/v/storage/record_batch_builder.h
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
#include "model/record.h"
#include "utils/vint.h"

#include <seastar/util/bool_class.hh>

namespace storage {
class record_batch_builder {
public:
Expand All @@ -30,7 +32,14 @@ class record_batch_builder {
std::optional<iobuf>&& key,
std::optional<iobuf>&& value,
chunked_vector<model::record_header> headers);
model::record_batch build() &&;
// Whether `build()` should finalize the batch by computing its size and
// checksums. Only valid if compression mode is compression::none. If set to
// `no`, it is the callers responsibility to invoke
// `batch.reset_checksum_metadata()`.
using reset_size_checksum = ss::bool_class<struct reset_size_checksum_tag>;

model::record_batch
build(reset_size_checksum reset = reset_size_checksum::yes) &&;
ss::future<model::record_batch> build_async() &&;
virtual ~record_batch_builder();

Expand Down