krabka_remote_storage/index/
time.rs

1//! The Kafka `TimeIndex` on-disk layout and the relative-offset floor a
2//! timestamp lookup scans from.
3//!
4//! An entry maps a timestamp to a relative offset. The index is sparse and
5//! preallocated, so the floor lookup deliberately stops short of the first
6//! entry at or above the target and treats non-increasing relative offsets as
7//! trailing padding.
8
9use {
10    super::{RelativeOffset, TimestampMs, corrupt_index},
11    crate::error::RemoteStorageError,
12    krabka_verified::log_index::TimeEntry,
13    std::mem,
14    zerocopy::{
15        BigEndian, FromBytes, Immutable, KnownLayout, Unaligned,
16        byteorder::{I64, U32},
17    },
18};
19
20/// 12 bytes per entry: ts i64 BE, then rel u32 BE. It mirrors
21/// `krabka_log::index::TimeEntryRaw`.
22#[derive(Debug, Clone, Copy, FromBytes, KnownLayout, Immutable, Unaligned)]
23#[repr(C)]
24pub struct TimeIndexEntry {
25    /// Largest record timestamp at or before the indexed offset.
26    pub timestamp: I64<BigEndian>,
27    /// Offset that carries the timestamp, relative to the segment's base
28    /// offset.
29    pub relative_offset: U32<BigEndian>,
30}
31
32/// Byte length of one serialized time-index entry.
33const TIME_INDEX_ENTRY_LEN: usize = mem::size_of::<TimeIndexEntry>();
34
35const _: [(); 12] = [(); TIME_INDEX_ENTRY_LEN];
36
37/// Borrows Kafka's `TimeIndex` on-disk format as a zero-copy
38/// `&[TimeIndexEntry]`.
39///
40/// Each entry is 12 bytes: ts i64 BE, then rel u32 BE. It ignores trailing bytes that do not complete a 12-byte entry. The result
41/// borrows from `bytes`.
42///
43/// # Errors
44///
45/// Returns [`RemoteStorageError::Io`] when the object store returned bytes that
46/// do not form an entry array.
47pub fn parse_time_index(bytes: &[u8]) -> Result<&[TimeIndexEntry], RemoteStorageError> {
48    let truncated_len = (bytes.len() / TIME_INDEX_ENTRY_LEN) * TIME_INDEX_ENTRY_LEN;
49    bytes
50        .get(..truncated_len)
51        .and_then(|whole| <[TimeIndexEntry]>::ref_from_bytes(whole).ok())
52        .ok_or_else(|| corrupt_index("time"))
53}
54
55/// Returns a safe relative-offset floor for an exact timestamp scan.
56///
57/// The last entry strictly below `target_ts` is used rather than an entry at
58/// or above it: a sparse index entry is only a seek hint and may follow the
59/// earliest qualifying record. Non-increasing relative offsets mark trailing
60/// preallocation padding and end the usable index.
61#[must_use]
62pub fn relative_offset_floor_for_timestamp(
63    entries: &[TimeIndexEntry],
64    target_ts: TimestampMs,
65) -> RelativeOffset {
66    let decoded: Vec<TimeEntry> = entries
67        .iter()
68        .map(|entry| TimeEntry::new(entry.timestamp.get(), entry.relative_offset.get()))
69        .collect();
70    // The kernel ends the usable index at the padding itself.
71    let candidate_count = krabka_verified::remote_time_index_candidate_count(&decoded, target_ts);
72    candidate_count
73        .checked_sub(1)
74        .and_then(|index| decoded.get(index))
75        .map_or(0, |entry| entry.coordinate)
76}
77
78#[cfg(test)]
79mod tests {
80    use {super::*, assert2::assert, std::fs};
81
82    /// A time index entry's decoded timestamp and relative offset.
83    #[derive(Debug, PartialEq, Eq)]
84    struct TimeSpec {
85        timestamp: i64,
86        relative_offset: u32,
87    }
88
89    const fn time_at(timestamp: i64, relative_offset: u32) -> TimeSpec {
90        TimeSpec {
91            timestamp,
92            relative_offset,
93        }
94    }
95
96    fn time_entries(specs: &[TimeSpec]) -> Vec<TimeIndexEntry> {
97        specs
98            .iter()
99            .map(|spec| TimeIndexEntry {
100                timestamp: I64::new(spec.timestamp),
101                relative_offset: U32::new(spec.relative_offset),
102            })
103            .collect()
104    }
105
106    #[yare::parameterized(
107        one_entry = { vec![(1_000_i64, 0_u32)] },
108        two_entries = { vec![(1_000_i64, 0_u32), (2_000, 10)] },
109        all_known_entries = { vec![(1_000_i64, 0_u32), (2_000, 10), (3_000, 20)] },
110    )]
111    fn parse_time_index_round_trips_known_entries(specs: Vec<(i64, u32)>) {
112        let mut buf = Vec::new();
113        for (ts, rel) in &specs {
114            buf.extend_from_slice(&ts.to_be_bytes());
115            buf.extend_from_slice(&rel.to_be_bytes());
116        }
117        let entries = parse_time_index(&buf).expect("valid time index");
118        let decoded: Vec<TimeSpec> = entries
119            .iter()
120            .map(|e| time_at(e.timestamp.get(), e.relative_offset.get()))
121            .collect();
122        assert2::assert!(
123            decoded
124                == specs
125                    .iter()
126                    .map(|(ts, rel)| time_at(*ts, *rel))
127                    .collect::<Vec<_>>()
128        );
129    }
130
131    #[yare::parameterized(
132        before_first = { 500, 0 },
133        exact_first = { 1_000, 0 },
134        between_entries = { 1_500, 0 },
135        duplicate_exact_matches = { 2_000, 0 },
136        between_duplicate_and_last = { 2_500, 20 },
137        after_last = { 4_000, 30 },
138    )]
139    fn timestamp_floor_covers_strict_predecessor_boundaries(ts: i64, want: u32) {
140        let entries = time_entries(&[
141            time_at(1_000, 0),
142            time_at(2_000, 10),
143            time_at(2_000, 20),
144            time_at(3_000, 30),
145        ]);
146        assert2::assert!(relative_offset_floor_for_timestamp(&entries, ts) == want);
147    }
148
149    #[test]
150    fn timestamp_floor_for_empty_index_is_zero() {
151        assert!(relative_offset_floor_for_timestamp(&[], 1_000) == 0);
152    }
153
154    #[yare::parameterized(
155        zero_interval = { 0 },
156        sparse_interval = { 65_536 },
157    )]
158    fn archived_segment_timestamp_scan_matches_local_and_record_oracle(interval_bytes: u32) {
159        use {
160            crate::{
161                first_record_at_or_after_timestamp, parse_offset_index,
162                position_for_relative_offset,
163            },
164            krabka_ids::Offset,
165            krabka_log::Segment,
166            krabka_protocol::records::{Record, RecordBatch},
167            krabka_units::prelude::bytes,
168        };
169
170        let timestamps = [[100, 100], [500, 100], [100, 300]];
171        let interval = bytes(interval_bytes);
172        let dir = tempfile::tempdir().unwrap();
173        let mut segment = Segment::create(dir.path(), Offset(100)).unwrap();
174        let mut records = Vec::new();
175        for (batch_index, timestamps) in timestamps.iter().enumerate() {
176            let base = 100 + i64::try_from(batch_index).unwrap() * 3;
177            let mut batch = RecordBatch {
178                base_offset: base,
179                last_offset_delta: 1,
180                base_timestamp: timestamps[0],
181                max_timestamp: *timestamps.iter().max().unwrap(),
182                ..Default::default()
183            };
184            for (delta, timestamp) in timestamps.iter().enumerate() {
185                let delta = i32::try_from(delta).unwrap();
186                batch.records.push(Record {
187                    offset_delta: delta,
188                    timestamp_delta: timestamp - timestamps[0],
189                    ..Default::default()
190                });
191                records.push((base + i64::from(delta), *timestamp));
192            }
193            segment.append(&batch, interval).unwrap();
194        }
195        segment.flush().unwrap();
196        let time_bytes = fs::read(dir.path().join("00000000000000000100.timeindex")).unwrap();
197        let offset_bytes = fs::read(dir.path().join("00000000000000000100.index")).unwrap();
198        let log_bytes = fs::read(dir.path().join("00000000000000000100.log")).unwrap();
199        let mut padded = time_bytes.clone();
200        padded.extend_from_slice(&[0; 24]);
201        for time_bytes in [&time_bytes[..], &padded[..]] {
202            for target in [i64::MIN, 100, 150, 300, 500, i64::MAX] {
203                let relative = relative_offset_floor_for_timestamp(
204                    parse_time_index(time_bytes).unwrap(),
205                    target,
206                );
207                let position = position_for_relative_offset(
208                    parse_offset_index(&offset_bytes).unwrap(),
209                    relative,
210                );
211                let remote = first_record_at_or_after_timestamp(
212                    &log_bytes[usize::try_from(position).unwrap()..],
213                    100 + i64::from(relative),
214                    target,
215                    None,
216                )
217                .unwrap();
218                let expected = records
219                    .iter()
220                    .copied()
221                    .find(|(_, timestamp)| *timestamp >= target);
222                let local = segment
223                    .offset_for_timestamp(target)
224                    .map(|(offset, timestamp)| (offset.0, timestamp));
225                assert2::assert!(remote == expected);
226                assert2::assert!(local == expected);
227            }
228        }
229    }
230
231    #[test]
232    fn timestamp_floor_ignores_trailing_index_padding() {
233        let entries = time_entries(&[
234            time_at(1_000, 0),
235            time_at(2_000, 10),
236            time_at(0, 0),
237            time_at(0, 0),
238        ]);
239        assert!(relative_offset_floor_for_timestamp(&entries, 3_000) == 10);
240    }
241}
krabka_remote_storage/index/time.rs

krabka_remote_storage/index/
time.rs
