diff --git a/src/v/compaction/filter.cc b/src/v/compaction/filter.cc index 6d6db19e5c59b..3951cb7fefaaf 100644 --- a/src/v/compaction/filter.cc +++ b/src/v/compaction/filter.cc @@ -66,8 +66,12 @@ ss::future> filter::do_filter_batch( // filter iobuf ret; int32_t rec_count = 0; + // Re-base surviving records onto the first surviving record's timestamp, as + // Kafka does on compaction (MemoryRecordsBuilder.appendWithOffset): the + // first kept record sets the new first_timestamp and every kept delta is + // re-based relative to it. std::optional first_timestamp_delta; - int64_t last_timestamp_delta; + int64_t max_timestamp_delta = 0; // We expect and enforce that offset_deltas is sorted. dassert( std::ranges::is_sorted(offset_deltas), @@ -75,7 +79,7 @@ ss::future> filter::do_filter_batch( size_t keep_idx = 0; co_await b.for_each_record_async([&rec_count, &first_timestamp_delta, - &last_timestamp_delta, + &max_timestamp_delta, &ret, &keep_idx, &offset_deltas](model::record record) { @@ -94,7 +98,12 @@ ss::future> filter::do_filter_batch( if (!first_timestamp_delta) { first_timestamp_delta = record.timestamp_delta(); } - last_timestamp_delta = record.timestamp_delta(); + const auto rebased = record.timestamp_delta() + - *first_timestamp_delta; + if (rebased > max_timestamp_delta) { + max_timestamp_delta = rebased; + } + record.set_timestamp_delta(rebased); model::append_record_to_buffer(ret, record); ++rec_count; } @@ -104,23 +113,16 @@ ss::future> filter::do_filter_batch( co_return std::nullopt; } - // There is no need to preserve the timestamp from the original - // batch after compaction. The FirstTimestamp field therefore always - // reflects the timestamp of the first record in the batch. If the batch - // is empty, the FirstTimestamp will be set to -1 (NO_TIMESTAMP). - // - // Similarly, the MaxTimestamp field reflects the maximum timestamp of - // the current records if the timestamp type is CREATE_TIME. For - // LOG_APPEND_TIME, on the other hand, the MaxTimestamp field reflects - // the timestamp set by the broker and is preserved after compaction. - // Additionally, the MaxTimestamp of an empty batch always retains the - // previous value prior to becoming empty. + // FirstTimestamp reflects the first surviving record; the per-record deltas + // were re-based onto it above, so absolute timestamps are unchanged. + // MaxTimestamp is the greatest surviving timestamp for CREATE_TIME; for + // LOG_APPEND_TIME it is the broker-set value and is preserved. auto& hdr = b.header(); const auto first_time = model::timestamp( hdr.first_timestamp() + first_timestamp_delta.value()); auto last_time = hdr.max_timestamp; if (hdr.attrs.timestamp_type() == model::timestamp_type::create_time) { - last_time = model::timestamp(first_time() + last_timestamp_delta); + last_time = model::timestamp(first_time() + max_timestamp_delta); } auto new_hdr = hdr; new_hdr.first_timestamp = first_time; diff --git a/src/v/compaction/tests/BUILD b/src/v/compaction/tests/BUILD index 0630121c50427..f62b27b42db43 100644 --- a/src/v/compaction/tests/BUILD +++ b/src/v/compaction/tests/BUILD @@ -69,7 +69,9 @@ redpanda_cc_gtest( "//src/v/compaction:reducer", "//src/v/compaction:utils", "//src/v/container:chunked_circular_buffer", + "//src/v/container:chunked_vector", "//src/v/model", + "//src/v/model:batch_builder", "//src/v/model:batch_compression", "//src/v/model/tests:random", "//src/v/reflection:adl", diff --git a/src/v/compaction/tests/reducer_test.cc b/src/v/compaction/tests/reducer_test.cc index 9b55721720490..59a8ef174c3c1 100644 --- a/src/v/compaction/tests/reducer_test.cc +++ b/src/v/compaction/tests/reducer_test.cc @@ -13,8 +13,13 @@ #include "compaction/reducer.h" #include "compaction/tests/simple_reducer.h" #include "container/chunked_circular_buffer.h" +#include "container/chunked_vector.h" +#include "model/batch_builder.h" #include "model/batch_compression.h" #include "model/fundamental.h" +#include "model/record.h" +#include "model/record_batch_types.h" +#include "model/timestamp.h" #include "reflection/adl.h" #include "storage/record_batch_builder.h" #include "storage/tests/batch_generators.h" @@ -87,6 +92,17 @@ model::record_batch make_int_key_batch(int num_records) { } return std::move(builder).build(); } + +model::record +ts_record(int32_t offset_delta, int64_t timestamp_delta, int key) { + return model::record( + model::record_attributes{}, + timestamp_delta, + offset_delta, + reflection::to_iobuf(key), + reflection::to_iobuf(key), + chunked_vector{}); +} } // namespace // The rebuild loop matches records against the kept-deltas vector with an @@ -164,3 +180,70 @@ TEST(CompactionReducerTest, FilterReusesCompressedBatchWhenUnchanged) { ASSERT_EQ(output.size(), 1); EXPECT_TRUE(output.front().compressed()); } + +// do_filter_batch re-encodes a batch when it drops records. Dropping the +// delta-0 base record of an out-of-order batch must not shift the surviving +// records' timestamps: like Kafka, first_timestamp advances to the first +// survivor and the per-record deltas are re-based onto it, so absolute +// timestamps are unchanged. Regression for compaction minting future timestamps +// from strictly-past input. +TEST(CompactionReducerTest, FilterPreservesTimestampsWhenBaseRecordDropped) { + const int64_t base = 1'700'000'000'000; // in the past + const int64_t delta_b = int64_t{30} * 24 * 3600 * 1000; // +30 days + const int64_t delta_c = 100; // +100 ms + + // Offsets [0,1,2], first_timestamp=base: key 0 delta 0 (base record, + // dropped below); key 1 delta +30d (first survivor, big delta); key 2 delta + // +100 (out of order vs key 1). + model::batch_builder bb; + bb.set_batch_type(model::record_batch_type::raft_data); + bb.set_term(model::term_id{0}); + bb.set_batch_timestamp( + model::timestamp_type::create_time, model::timestamp(base)); + bb.set_base_offset(model::offset{0}); + bb.add_record(ts_record(0, 0, 0)); + bb.add_record(ts_record(1, delta_b, 1)); + bb.add_record(ts_record(2, delta_c, 2)); + + // Supersede key 0 (index it above its offset) so the base record is dropped + // and the batch has to be re-encoded. + compaction::simple_key_offset_map map{}; + map + .put( + compaction::compaction_key{iobuf_to_bytes(reflection::to_iobuf(0))}, + model::offset(std::numeric_limits::max())) + .get(); + + chunked_circular_buffer output; + compaction::simple_sink sink(output); + compaction::simple_map_filter filter(sink, map, test_ntp); + filter(bb.build_sync()).get(); + + ASSERT_EQ(output.size(), 1); + auto& batch = output.front(); + ASSERT_FALSE(batch.compressed()); + EXPECT_EQ(batch.record_count(), 2); + const int64_t first_ts = batch.header().first_timestamp.value(); + // Kafka-style re-encode: first_timestamp advances to the first survivor + // (key 1); max_timestamp is the greatest surviving timestamp. + EXPECT_EQ(first_ts, base + delta_b); + EXPECT_EQ(batch.header().max_timestamp.value(), base + delta_b); + + bool saw_b = false; + bool saw_c = false; + batch.for_each_record([&](model::record r) { + const int64_t abs_ts = first_ts + r.timestamp_delta(); + const int key = reflection::from_iobuf(r.key().copy()); + if (key == 1) { + EXPECT_EQ(abs_ts, base + delta_b) << "key 1 shifted"; + saw_b = true; + } else if (key == 2) { + EXPECT_EQ(abs_ts, base + delta_c) << "key 2 shifted"; + saw_c = true; + } + EXPECT_LE(abs_ts, base + delta_b) + << "record timestamp shifted forward by compaction"; + }); + EXPECT_TRUE(saw_b); + EXPECT_TRUE(saw_c); +} diff --git a/src/v/model/record.h b/src/v/model/record.h index b2baac54dae28..fc8addec2bbd8 100644 --- a/src/v/model/record.h +++ b/src/v/model/record.h @@ -262,6 +262,15 @@ class record { int64_t timestamp_delta() const { return _timestamp_delta; } + // Replaces the timestamp delta, keeping _size_bytes in sync with the + // new delta's vint encoding width. + void set_timestamp_delta(int64_t delta) { + _size_bytes += static_cast(vint::vint_size(delta)) + - static_cast( + vint::vint_size(_timestamp_delta)); + _timestamp_delta = delta; + } + int32_t offset_delta() const { return _offset_delta; } int32_t key_size() const { return _key_size; } diff --git a/src/v/model/tests/record_batch_test.cc b/src/v/model/tests/record_batch_test.cc index 7a290bb7be2a7..31eb029340cc5 100644 --- a/src/v/model/tests/record_batch_test.cc +++ b/src/v/model/tests/record_batch_test.cc @@ -226,6 +226,33 @@ TEST_F(RecordBatchTest, RecordSizeBytesWithNullHeaderValues) { check_serialization_size(r); } +TEST_F(RecordBatchTest, SetTimestampDeltaKeepsSizeBytesAccurate) { + auto r = model::record( + model::record_attributes(0), + 0, + 0, + iobuf::from("key"), + iobuf::from("value"), + {}); + check_serialization_size(r); + + // Cross vint width boundaries in both directions (zigzag: -64..63 fits in + // one byte), including negative and multi-day deltas. + for (int64_t delta : + {int64_t{63}, + int64_t{64}, + int64_t{1209600000}, // 14 days in ms + int64_t{-1}, + int64_t{-64}, + int64_t{-65}, + int64_t{-1209600000}, + int64_t{0}}) { + r.set_timestamp_delta(delta); + EXPECT_EQ(r.timestamp_delta(), delta); + check_serialization_size(r); + } +} + class RecordBatchCompressionTest : public ::testing::TestWithParam {}; diff --git a/src/v/storage/compaction_reducers.cc b/src/v/storage/compaction_reducers.cc index 8b9e2f5b5dddc..f0da03e0d967c 100644 --- a/src/v/storage/compaction_reducers.cc +++ b/src/v/storage/compaction_reducers.cc @@ -230,8 +230,13 @@ copy_data_segment_reducer::filter(model::record_batch batch) { // 4. filter iobuf ret; int32_t rec_count = 0; + // Re-base surviving records onto the first surviving record's timestamp, as + // Kafka does on compaction (MemoryRecordsBuilder.appendWithOffset). Records + // are dropped in offset order, so the first kept record sets the new batch + // first_timestamp; every kept delta is re-based relative to it, and we + // track the greatest re-based delta for MaxTimestamp. std::optional first_timestamp_delta; - int64_t last_timestamp_delta; + int64_t max_timestamp_delta = 0; // We expect and enforce that offset_deltas is sorted. dassert( std::ranges::is_sorted(offset_deltas), @@ -239,7 +244,7 @@ copy_data_segment_reducer::filter(model::record_batch batch) { size_t keep_idx = 0; batch.for_each_record([&rec_count, &first_timestamp_delta, - &last_timestamp_delta, + &max_timestamp_delta, &ret, &keep_idx, &offset_deltas](model::record record) { @@ -258,7 +263,12 @@ copy_data_segment_reducer::filter(model::record_batch batch) { if (!first_timestamp_delta) { first_timestamp_delta = record.timestamp_delta(); } - last_timestamp_delta = record.timestamp_delta(); + const auto rebased = record.timestamp_delta() + - *first_timestamp_delta; + if (rebased > max_timestamp_delta) { + max_timestamp_delta = rebased; + } + record.set_timestamp_delta(rebased); model::append_record_to_buffer(ret, record); ++rec_count; } @@ -290,24 +300,16 @@ copy_data_segment_reducer::filter(model::record_batch batch) { co_return std::nullopt; } - // There is no similar need to preserve the timestamp from the original - // batch after compaction. The FirstTimestamp field therefore always - // reflects the timestamp of the first record in the batch. If the batch is - // empty, the FirstTimestamp will be set to -1 (NO_TIMESTAMP). - // - // Similarly, the MaxTimestamp field reflects the maximum timestamp of the - // current records if the timestamp type is CREATE_TIME. For - // LOG_APPEND_TIME, on the other hand, the MaxTimestamp field reflects the - // timestamp set by the broker and is preserved after compaction. - // Additionally, the MaxTimestamp of an empty batch always retains the - // previous value prior to becoming empty. - // + // Header timestamps for the re-encoded batch: first_timestamp = first + // surviving record (deltas were re-based onto it above); max_timestamp = + // greatest surviving timestamp for CREATE_TIME, or the preserved broker + // value for LOG_APPEND_TIME. auto& hdr = batch.header(); const auto first_time = model::timestamp( hdr.first_timestamp() + first_timestamp_delta.value()); auto last_time = hdr.max_timestamp; if (hdr.attrs.timestamp_type() == model::timestamp_type::create_time) { - last_time = model::timestamp(first_time() + last_timestamp_delta); + last_time = model::timestamp(first_time() + max_timestamp_delta); } auto new_hdr = hdr; diff --git a/src/v/storage/tests/BUILD b/src/v/storage/tests/BUILD index ea9907e0b8662..2fb9f0793b49c 100644 --- a/src/v/storage/tests/BUILD +++ b/src/v/storage/tests/BUILD @@ -945,6 +945,7 @@ redpanda_cc_gtest( ], cpu = 1, deps = [ + ":disk_log_builder", "//src/v/base", "//src/v/cluster", "//src/v/cluster:client_quota_backend", @@ -962,6 +963,7 @@ redpanda_cc_gtest( "//src/v/features", "//src/v/kafka/server/tests:kafka_test_utils", "//src/v/model", + "//src/v/model:batch_builder", "//src/v/model:batch_compression", "//src/v/random:generators", "//src/v/redpanda/tests:fixture", diff --git a/src/v/storage/tests/compaction_e2e_test.cc b/src/v/storage/tests/compaction_e2e_test.cc index 7c7a65eae0a03..c84e9bcd5c7f2 100644 --- a/src/v/storage/tests/compaction_e2e_test.cc +++ b/src/v/storage/tests/compaction_e2e_test.cc @@ -16,7 +16,9 @@ #include "features/feature_table.h" #include "gtest/gtest.h" #include "kafka/server/tests/produce_consume_utils.h" +#include "model/batch_builder.h" #include "model/namespace.h" +#include "model/record.h" #include "model/record_batch_reader.h" #include "model/record_batch_types.h" #include "model/timeout_clock.h" @@ -28,6 +30,7 @@ #include "storage/segment_set.h" #include "storage/segment_utils.h" #include "storage/tests/manual_mixin.h" +#include "storage/tests/utils/disk_log_builder.h" #include "storage/types.h" #include "test_utils/async.h" #include "test_utils/container_ostream.h" // IWYU pragma: keep @@ -38,9 +41,12 @@ #include +#include #include +#include #include #include +#include using namespace std::chrono_literals; @@ -2504,3 +2510,161 @@ TEST_F(CompactionFixtureTest, CommitTransactions) { ASSERT_EQ(num_data_records, 1); } } + +namespace { + +iobuf ts_test_iobuf(std::string_view s) { + iobuf b; + b.append(s.data(), s.size()); + return b; +} + +model::record ts_test_record( + int32_t offset_delta, int64_t timestamp_delta, std::string_view key) { + return model::record( + model::record_attributes{}, + timestamp_delta, + offset_delta, + ts_test_iobuf(key), + ts_test_iobuf("v"), + chunked_vector{}); +} + +// Appends batch1 and batch2 into separate sealed segments, runs sliding-window +// compaction (which drops batch1's superseded base record and re-encodes it +// through copy_data_segment_reducer::filter()) and reads the log back. +ss::future ts_compact_and_read( + model::record_batch batch1, + model::record_batch batch2, + model::offset max_off) { + storage::log_config cfg = storage::log_builder_config(); + storage::disk_log_builder b( + cfg, model::offset_translator_batch_types(), raft::group_id{0}); + model::ntp ntp( + model::kafka_namespace, + model::topic_partition( + model::topic("compaction-ts"), model::partition_id{0})); + co_await b.start(ntp); + ss::abort_source as; + co_await b.get_disk_log_impl().start(std::nullopt, as); + std::exception_ptr error = nullptr; + try { + co_await b.add_batch(std::move(batch1)); + co_await b.get_disk_log_impl().force_roll(); + co_await b.add_batch(std::move(batch2)); + co_await b.get_disk_log_impl().force_roll(); + auto compact_cfg = compaction::compaction_config( + max_off, max_off, max_off, std::nullopt, std::nullopt, as); + std::ignore = co_await b.apply_sliding_window_compaction(compact_cfg); + } catch (...) { + error = std::current_exception(); + } + model::record_batch_reader::data_t batches; + if (!error) { + auto reader = co_await b.get_disk_log_impl().make_reader( + storage::local_log_reader_config( + model::offset{0}, model::offset::max())); + batches = co_await model::consume_reader_to_memory( + std::move(reader), model::no_timeout); + } + co_await b.stop(); + if (error) { + std::rethrow_exception(error); + } + co_return batches; +} + +} // namespace + +// Regression test for compaction minting future record timestamps. +// copy_data_segment_reducer::filter() re-encodes a batch when compaction drops +// records. When the delta-0 base record is dropped (superseded/tombstoned) and +// the batch is out of order, advancing first_timestamp without re-basing the +// per-record deltas shifted every survivor forward -- minting FUTURE timestamps +// from strictly-past input. This drives that path end to end and asserts that +// absolute timestamps are preserved and encoded as Kafka does (first_timestamp +// = first surviving record, deltas re-based, max_timestamp = greatest +// survivor). +TEST(CompactionReducerTimestampTest, PreservesTimestampsWhenBaseRecordDropped) { + const int64_t base = 1'700'000'000'000; // 2023-11-14, past + const int64_t delta_b = int64_t{30} * 24 * 3600 * 1000; // +30 days + const int64_t delta_c = 100; // +100 ms + + // Batch 1 (first_timestamp=base), offsets [0,1,2]: A@0 delta 0 (base + // record, dropped below); B@1 delta +30d (first survivor, big delta); C@2 + // delta +100 (out of order vs B). + model::batch_builder bb1; + bb1.set_batch_type(model::record_batch_type::raft_data); + bb1.set_term(model::term_id{0}); + bb1.set_batch_timestamp( + model::timestamp_type::create_time, model::timestamp(base)); + bb1.set_base_offset(model::offset{0}); + bb1.add_record(ts_test_record(0, 0, "A")); + bb1.add_record(ts_test_record(1, delta_b, "B")); + bb1.add_record(ts_test_record(2, delta_c, "C")); + + // Batch 2: a newer A@3 supersedes A@0, so compaction drops batch 1's + // delta-0 base record and re-encodes the batch. + model::batch_builder bb2; + bb2.set_batch_type(model::record_batch_type::raft_data); + bb2.set_term(model::term_id{0}); + bb2.set_batch_timestamp( + model::timestamp_type::create_time, model::timestamp(base)); + bb2.set_base_offset(model::offset{3}); + bb2.add_record(ts_test_record(0, 0, "A")); + + auto batch1 = bb1.build_sync(); + auto batch2 = bb2.build_sync(); + const auto max_off = batch2.last_offset(); + auto batches = ts_compact_and_read( + std::move(batch1), std::move(batch2), max_off) + .get(); + + bool saw_b = false; + bool saw_c = false; + bool saw_rebuilt = false; + for (auto& batch : batches) { + if (batch.header().type != model::record_batch_type::raft_data) { + continue; + } + ASSERT_FALSE(batch.compressed()); + const int64_t first_ts = batch.header().first_timestamp.value(); + int64_t observed_max_delta = std::numeric_limits::min(); + for (const auto& r : batch.copy_records()) { + const int64_t abs_ts = first_ts + r.timestamp_delta(); + observed_max_delta = std::max( + observed_max_delta, r.timestamp_delta()); + if (r.key() == ts_test_iobuf("A")) { + EXPECT_EQ(abs_ts, base) << "A shifted"; + } else if (r.key() == ts_test_iobuf("B")) { + // The bug shifted this to base + 2*delta_b (a future + // timestamp). + EXPECT_EQ(abs_ts, base + delta_b) << "B shifted"; + saw_b = true; + } else if (r.key() == ts_test_iobuf("C")) { + EXPECT_EQ(abs_ts, base + delta_c) << "C shifted"; + saw_c = true; + } + EXPECT_LE(abs_ts, base + delta_b) + << "record timestamp shifted forward by compaction"; + } + // Batch-level invariant for every batch: max_timestamp is + // first_timestamp plus the greatest surviving delta. + EXPECT_EQ( + batch.header().max_timestamp.value(), first_ts + observed_max_delta); + // The re-encoded batch 1 keeps base offset 0 but drops its delta-0 base + // record; like Kafka, first_timestamp advances to the first survivor + // (B) and max_timestamp is the greatest surviving timestamp. + if (batch.base_offset() == model::offset{0}) { + saw_rebuilt = true; + EXPECT_EQ(batch.record_count(), 2); + EXPECT_EQ(first_ts, base + delta_b); + EXPECT_EQ(batch.header().max_timestamp.value(), base + delta_b); + } + } + // B and C must have survived (A@0 dropped, A@3 survives) and batch 1 must + // have been re-encoded. + EXPECT_TRUE(saw_b); + EXPECT_TRUE(saw_c); + EXPECT_TRUE(saw_rebuilt); +}