storage: fix compaction minting future record timestamps from strictly-past input

src/v/compaction/filter.cc

@@ -66,16 +66,20 @@ ss::future<std::optional<filter::filtered_batch>> 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<int64_t> 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),
       "offset_deltas must be ascending in record-iteration order");
     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<std::optional<filter::filtered_batch>> 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<std::optional<filter::filtered_batch>> 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;

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",

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<model::record_header>{});
+}
 } // 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<int>::max()))
+      .get();
+
+    chunked_circular_buffer<model::record_batch> 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<int>(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);
+}

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<int32_t>(vint::vint_size(delta))
+                       - static_cast<int32_t>(
+                         vint::vint_size(_timestamp_delta));
+        _timestamp_delta = delta;
+    }
+
     int32_t offset_delta() const { return _offset_delta; }
 
     int32_t key_size() const { return _key_size; }

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<model::compression> {};
 

src/v/storage/compaction_reducers.cc

@@ -230,16 +230,21 @@ 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<int64_t> 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),
       "offset_deltas must be ascending in record-iteration order");
     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;
 

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",