fix(slatedb): redesign ListOffsets(Timestamp) to match Kafka's time-index semantics

[solace-aross]

What changes for a user or operator

• SlateDB: ListOffsets with a timestamp answers the offset and timestamp of the first record at or after that timestamp, instead of the base offset of its batch. A lookup with no such record answers no match (offset -1), as on every other engine.
• Under READ_COMMITTED, a match at or past the last stable offset answers no match, as in Apache Kafka.
• The Earliest timestamp comes from the first record at or after the log start offset, read from the batch that holds the log start or from the next batch. The Latest timestamp comes from the newest batch's maximum timestamp, and is available on each partition after the partition's first write on this version; before that write it is missing.
• The engine keeps a new t/ time-index keyspace, and builds it for a partition on the first write to that partition after the upgrade. Each build logs an info line with the batch count and elapsed time.
• Compaction and retention no longer stop the maintenance tick when a produce commits to the partition they are working on. The partition is logged at warn and retried on the next tick; the partitions after it, and lake maintenance, run in the same tick.
• Downgrade: an older binary still reads each partition's low and high, and ignores the t/ keyspace, but it answers a timestamp lookup and the Earliest and Latest timestamps only from the watermark's timestamps map, which this version always writes as None. After a downgrade, a timestamp lookup therefore answers no match on every partition that this version wrote to, until the older binary appends a batch to it; from then on a lookup for any earlier target answers that batch's offset, so a consumer that resets by time skips the data before it. A topic that the older binary deletes leaves its t/ entries behind. On the next upgrade, the first write to each partition that the older binary wrote to rebuilds that partition's index from its batches.

Problem

SlateDB's ListOffsets(Timestamp) did not answer Kafka's question. It kept, inside each partition's watermark, a base_timestamp -> base_offset map and answered with the first entry at or after T. That differs from Kafka in five ways:

1. Two batches with the same base_timestamp share a map key, so the later batch overwrites the earlier one and the lookup skips the earlier batch.
2. The map is ordered by time, not offset, so a later batch with an earlier time is found ahead of earlier batches that also reach T, a later offset than Kafka's answer.
3. The key is a batch's base time, not its max time, so a batch whose base time is before T but whose max time reaches T is missed entirely.
4. It answers a batch's base offset and base time, not the first record at or after T and that record's own time.
5. Every produce rewrote the whole watermark value, map included. Only delete, compaction and DeleteRecords pruned it, so a produce's write size grew with the partition's live batch count.

Design

• Per-batch t/ keys, keyed by max_timestamp (big-endian, so offset-ordered alongside it), instead of an in-watermark map. A batch is indexed only when its max_timestamp raises the partition's running maximum (append_time_index, Kafka's own TimeIndex.maybeAppend), so entries rise in both time and offset together. This fixes bugs 1 to 3 and bounds index growth to new maxima rather than one entry per produce. A timestamp of -1 or below (NO_TIMESTAMP) is never indexed.
• Lookup (list_offset_for_timestamp): an O(1) no-match answer when T is beyond everything ever indexed; otherwise find the index's ceiling entry and scan b/ batches forward from its offset, inspecting each batch's own records for the first one at or after the log start whose timestamp is >= T (fixes bug 4). A batch header can overstate but never understate the true max after compaction, so a qualifying header is a reason to look inside the batch, not a match by itself. When the index cannot rule out a gap below its first surviving entry (pruning deletes t/ entries without uncovering new ones, which Kafka's per-segment index never has to handle), the scan starts at the batch that holds the log start, and records below the log start are skipped, so a batch that straddles the log start is handled.
• One snapshot per request: list_offsets reads the watermark, the index and the batches of every partition in the request through one SlateDB snapshot. The index is read twice (its first entry decides whether the ceiling entry can be trusted), and a prune that committed between two reads of the live database removed entries from the second read only, so the lookup skipped the batches between the new log start and the first surviving entry. Fetch takes a snapshot for its batch probe and scan too, since it shares the probe helpers.
• Isolation level: under READ_COMMITTED, a match at or past the last stable offset answers no match, as in Kafka.
• Pruning and compaction: delete_records and policy_delete prune t/ entries below the new log start through the write transaction (a prefix delete that stops at the first survivor, since entries are both timestamp- and offset-ordered). policy_compact deletes and replays the index from the surviving batches in offset order, because a partially compacted batch keeps its original, now possibly stale, header, and pruning individual entries can leave the index naming an offset that no longer carries the batch that justified it. The compaction reads the batches, the watermark and the index through its transaction, so a produce that commits during the compaction makes the compaction fail to commit instead of losing its index entry. When a prune empties the partition, the watermark's time-index state is reset, so ListOffsets(Latest) does not answer a deleted batch's timestamp and a refill with older timestamps is indexed again.
• Maintenance conflicts: the per-partition body of each policy (compact_partition, delete_expired_prefix) takes the transaction, and commit_maintenance commits it. A commit that fails with a transaction conflict is logged at warn with the topic and partition and skipped until the next tick, instead of returned, because a returned error ended the tick before the partitions after that one and before lake maintenance, and the broker logs it only at debug; topics iterate in name order, so a partition under steady writes held back the same partitions on every tick. Any other error is still returned.
• Earliest and Latest: the Earliest timestamp is the first readable record at or after the log start. It is the lookup scan with a target that every timestamp reaches, started at the batch that holds the log start, so it steps over the batch before the log start when compaction removed the batch at it (compaction leaves the log start in place) and over a corrupt batch. The Latest timestamp is the newest batch's max_timestamp, kept on the watermark so the answer is O(1).
• Corrupt batches: the lookup scan, and so the Earliest timestamp, and the backfill skip a batch that fails to decode or inflate, and log the decode error at warn with topic, partition and offset. Kafka returns an error for a corrupt batch here; this is a deliberate difference, noted at the skip site. A failure in the backfill would abort the write that triggered the migration, and every later write to the partition would fail the same way.

Migration

A legacy 3-field watermark is distinguished from the new 5-field shape by its decoding: a value is treated as legacy only when it fails to decode as the current layout and decodes as exactly the legacy layout with no bytes left over. On the first write that touches the partition, it is migrated in place: the time index is rebuilt by replaying the monotonic rule over every batch in the partition's b/ keyspace (backfill_time_index), reading through the write's transaction, rather than trusting the old, lossy, wrongly keyed map. The backfill deletes any t/ entries already present first, so entries left behind by a downgrade and re-upgrade cannot survive out of order. Each migration logs an info line with the batch count and elapsed time. Read-only paths (list_offsets, offset_stage) tolerate the legacy shape without migrating, and skip the O(1) shortcut in that case, since an index that is not built yet understates reality.

A binary without the time index still reads low and high from the 5-field value, because postcard ignores trailing bytes. What its lookups answer after a downgrade is described above.

Conventions

None is the no-match answer on both the O(1) shortcut and the scan miss, as on every engine since #838 merged. The broker maps it to -1.

Testing

cargo nextest run -p nisshi-storage-slatedb --all-features: 63 passed, including:

• The worked example as a real test (5 batches of 10 records, out-of-order and overlapping timestamps, before and after delete_records).
• Ceiling lookups between entries, at an exact boundary, and at an exact middle entry (the middle-entry case fails when the ceiling seek is mutated to target + 1).
• A batch that straddles the log start: records below it are skipped by a timestamp lookup, and the Earliest timestamp comes from inside it (fails when the batch before the log start is dropped, or when the next batch is tried first).
• The Earliest timestamp when the batch that straddles the log start, or the batch that starts at it, cannot be inflated: it comes from the next batch (both fail when the scan stops at a corrupt batch).
• A compaction scenario that discriminates full rebuild from survivors from a naive per-entry prune, and a stale-header scenario that must not match on the header alone.
• A compaction and a retention pass that lose a conflict to a produce committed while they ran: the commit answers a conflict, the produce's batch and t/ entry survive, nothing of the pass applies, and the next tick applies it (both fail when the conflict is returned as an error).
• A legacy-watermark scenario that discriminates the backfill from the lookup-side scan fallback, a migration triggered by delete_records that must keep a rebuilt entry, and a migration that must delete stale t/ entries.
• READ_COMMITTED stops at the last stable offset; a future timestamp and a scan miss both answer no match.
• delete_records that empties a partition clears the Latest timestamp, a refill with older timestamps is indexed again, and a partial prune keeps the running maximum.
• Earliest after compaction removes the batch at the log start answers the first surviving record's timestamp.
• delete_topic leaves no t/ keys behind.
• A lookup through a snapshot taken before delete_records answers from the state before it (fails when the index or batch reads go to the live database).

cargo nextest run -p nisshi-broker -p nisshi-storage -p nisshi-storage-slatedb --all-features --no-fail-fast -E '!test(/::pg::/)': 527 run, 527 passed.

cargo fmt --all --check, cargo clippy --workspace --all-features --all-targets -- -D warnings, and RUSTDOCFLAGS="-D warnings" cargo doc --workspace --all-features --no-deps --document-private-items are clean.

Related

• fix(dynostore): ListOffsets by timestamp now answers from record time, not object write time #840 redesigns the same lookup for dynostore. Both answer the first record at or after the target, decode records rather than trust headers, apply the READ_COMMITTED bound, answer None for no match, and never index NO_TIMESTAMP. They differ on a batch that cannot be read: fix(dynostore): ListOffsets by timestamp now answers from record time, not object write time #840 answers KAFKA_STORAGE_ERROR in that partition's error_code while the other partitions of the request answer normally; this PR skips the batch with a warn in the lookup scan, the backfill and the Earliest timestamp. A follow-up ticket has been raised to align them.

🤖 Generated with Claude Code

https://claude.ai/code/session_01D8pQUJmdicavN4zsumUdaD

[sgamelin]

Thanks for this. The lookup is right: every batch before the ceiling entry has a max timestamp below that entry's, so starting the forward scan there skips nothing, and inspecting records rather than trusting the header matches Kafka's FileRecords.searchForTimestamp. Moving the index out of the watermark value is the right call, too.

I'm requesting changes for one thing: the compaction rebuild can drop a time index entry written by a concurrent produce (inline on storage.rs:804). After that, ListOffsets(Timestamp) answers "no match" or a later offset for data that exists, until a later compaction removes a record. The other inline comments are non-blocking.

Not inline:

• CHANGELOG. [Unreleased] has no entry. Clients see three changes: a timestamp lookup now answers the matching record's offset and timestamp instead of the batch's base, the Latest and Earliest timestamps change source, and SlateDB gets a new t/ keyspace with a migration on first write. The migration and downgrade behaviour (see the storage.rs:122 comment) are worth a line for operators.
• Isolation level. Kafka drops a timestamp match at or above the last fetchable offset (the LSO for READ_COMMITTED, the high watermark otherwise; Partition.scala#L1589-L1593, #L1622-L1623). The Timestamp arm here still ignores isolation_level, so a READ_COMMITTED lookup can answer an offset inside an open transaction. That was true before this PR, so it doesn't need to be fixed here, but the PR claims Kafka semantics. Could you either filter here or open an issue and note the difference at the code site?
• Corrupt batches. sequential_timestamp_scan skips a batch that fails to decode or inflate and returns a later offset with ErrorCode::None, so a consumer resetting by time seeks past the corrupt data without being told. Kafka surfaces an error there. If skipping is deliberate, please say so in a comment at the skip site, and include the decode error in the warn! so the log shows why it failed.
• #838. Both no-match returns in list_offset_for_timestamp (the shortcut and the scan miss) need #838's None whichever PR merges second. That's mechanical.

Kafka behaviour checked against 3.9.1 by reading code; I didn't build or run this branch.

[sgamelin]

A few things about the migration:

1. It runs inside whichever write touches the partition first (produce, EndTxn, DeleteRecords, retention, compaction). It reads every batch of the partition and buffers about one t/ put per batch in that transaction, and txn_end can do this for several partitions at once. Right after an upgrade, every busy partition pays this on its first produce, concurrent produces repeat it and then lose the watermark conflict, and nothing logs that a migration is running. Could it run in maintain (or at startup), committing in chunks? At the least, an info! per migrated partition with batch count and elapsed time would help an operator explain slow first produces.
2. Read-only paths never migrate, so a partition that is never written again (cold, or retention.ms=-1 with no compaction) stays legacy forever, and every timestamp lookup on it scans from low. A migration in maintain would fix this as well.
3. The backfill doesn't call delete_time_index first, although its doc says the caller should. After a downgrade and re-upgrade, t/ entries from before the downgrade survive. They can be out of order with the backfilled ones, and prune_time_index_below stops at the first survivor, so it never removes them.
4. Treating any decode error as "legacy" works for exactly one format change. When a sixth field is added, a 5-field value fails to decode, then decodes as WatermarkLegacy (postcard ignores trailing bytes), and the index state is silently dropped and backfilled again. A version marker (an enum wrapper, or a new key) would make the next change explicit.

[solace-aross]

Partly done in 4d4bf7a:

1. Added an info! per migrated partition with the batch count and elapsed_ms. Moving the migration into maintain and chunking it is left for a follow-up.
2. Same follow-up.
3. The backfill now calls delete_time_index first, and reads batches through tx. Added a test with stale t/ entries.
4. I didn't add a version marker. Postcard ignores trailing bytes, so a binary built from main can still read the new 5-field value today, which keeps downgrade working. A leading version byte would make that binary fail on every migrated watermark. Instead, a value is treated as legacy only when it decodes as exactly the legacy layout with no bytes left over; anything else is an error, so a future 6-field value can't silently fall back. Happy to revisit if you'd rather trade downgrade for an explicit marker.

[solace-aross]

No change on items 1 and 2 this round. Running the migration from maintain, in chunks so cold partitions migrate too, stays a follow-up. Item 3 is in. On item 4, a 3-field value can never decode as the current layout, and postcard ignores trailing bytes, so a future 6-field value decodes as the current layout with the extra field dropped, never as legacy. The exact-length check only guards a value that is neither. The downgrade path stays. Happy to switch to a version byte if you would rather trade that away.

One more fix from this round: prune_time_index_below now reads through the transaction. A migration triggered by DeleteRecords or retention rebuilds the index in the same transaction, and the prune was reading the committed index, so an old entry below the cutoff that shared its key with a rebuilt entry deleted the rebuilt one (legacy_migration_by_delete_records_keeps_a_rebuilt_entry).

[solace-aross]

No change on 1 and 2 this round; running the migration from maintain, in chunks so cold partitions migrate too, stays a follow-up. 3 is in. On 4, the exact-length check stands as above, and I am happy to switch to a version byte if you would rather trade the downgrade path for it. One related change: lookups now read the watermark, the index and the batches through one snapshot, so a prune that commits mid-lookup cannot move the index out from under the start point.

[sgamelin]

Question: Kafka 3.9.1 returns NO_TIMESTAMP for both EARLIEST and LATEST (UnifiedLog.scala#L1325-L1356: "For the earliest and latest, we do not need to return the timestamp"). This PR keeps Nisshi's existing behaviour of returning one, and pays for it: Earliest now does up to three scans (more when low isn't a batch base) plus a full batch inflate on every call, and Latest needs the new persisted last_batch_max_timestamp with its own invalidation in prune, compaction and migration. Lag exporters poll Earliest and Latest on every partition. pg returns record timestamps here too, so this is a cross-backend decision rather than this PR's, but would you consider returning None for both and dropping the field, or doing that in a follow-up?

[solace-aross]

I'd do that as a follow-up, since pg returns record timestamps here too and it should change on every backend at once. This PR keeps the current behaviour.

[solace-aross]

Still a follow-up. pg returns record timestamps here too, so it should change on every backend at once.

[solace-aross]

Still a follow-up for every backend at once; pg returns record timestamps here too. One fix this round in the same code: Earliest reads the next batch when compaction removed the batch at the log start, so the timestamp matches the changelog wording.

[sgamelin]

The ceiling path is only exercised for speed, not for correctness. Here T=150 finds the last entry, and if the ceiling seek were off by one (scan_from(target_millis + 1)), the fallback to low would still answer 2. As far as I can tell, forcing start_offset = low everywhere also passes every test. A case where T equals a middle entry's timestamp (for example {100→0, 150→2, 200→5} at T=150) would catch both.

[solace-aross]

Added the middle-entry case in 4d4bf7a: {100->0, 150->2, 200->5} at T=150 must answer 2. With the seek mutated to target_millis + 1, only this test fails, so it covers the off-by-one. Forcing start_offset = low can't be caught through the answer, though: every batch before the ceiling entry has a max timestamp below T, so a scan from low returns the same offset, only more slowly. That one would need a read counter.

[solace-aross]

Nothing new; the middle-entry case is in. A read counter for the start_offset = low case would need a counting Db wrapper, which I would rather not add for one test.

[solace-aross]

The middle-entry case is in. A read counter for the start_offset = low case needs a counting Db wrapper, which I would rather not add for one test. Happy to add it if you feel strongly.

[solace-aross]

@sgamelin pushed 92b86b31, rebased on main. Summary of what changed since your review:

• Isolation level: now filtered rather than noted. Under READ_COMMITTED, a match at or past the last stable offset answers no match (read_committed_lookup_stops_at_the_last_stable_offset), the same rule as the dynostore PR.
• fix: ListOffsets by timestamp should answer offset -1, not 0, when no record matches #838: both no-match returns answer None now that it is on main (future_timestamp_returns_no_match, scan_miss_after_delete_records_answers_no_match).
• Rebased over fix: answer OFFSET_OUT_OF_RANGE for a fetch offset outside the partition #826: compaction no longer moves the log start; the rebuild from survivors stays.
• prune_time_index_below reads through the transaction, so a migration in the same write is not undone by the prune.

[solace-aross]

Rebased on main and ready for another look. Your seven commits are unchanged. Two commits are new:

• Earliest after compaction: ListOffsets(Earliest) answered no timestamp when compaction removed the batch at the log start, because the lookup only tried the batch before it. It now tries the batch at or before the offset, then the next one. The two warn! lines on that path carry the decode error, as you asked for the scan. earliest_timestamp_after_compaction_removes_the_first_batch fails without the fix.
• One snapshot per request: the watermark, the index and the batches are now read through one snapshot. The index was read twice, and a prune that committed between the two reads could make the lookup answer a later offset than the right one. timestamp_lookup_reads_through_one_snapshot covers the mechanism.

I also checked whether the backfill can race a produce here, as it could on #840. It cannot: produce assigns the offset, writes the batch and the watermark, and commits in one transaction, and the backfill runs inside the write's own transaction.

One difference between the engines to settle in a single change: on a corrupt batch this PR skips and logs, as you suggested, while #840 fails the request. A storage error fails every partition in the ListOffsets frame, which is harsher than Kafka's per-partition error code, so the right answer for both may be a per-partition error.

[sgamelin]

Thanks for the thorough round. The compaction race I blocked on is fixed: policy_compact now begins its transaction before it scans, and reads the batches, the index and the watermark through tx (7936bbd). The isolation filter matches Kafka 3.9.1, which keeps a timestamp match only below the last stable offset under READ_COMMITTED (Partition.scala#L1589-L1593, #L1622-L1623). The straddling-batch fix, the backfill's delete_time_index call, the empty-partition reset, the middle-entry test and the None answers all look right. Reading each ListOffsets request through one snapshot also closes a race I had missed.

Two things I'd like before merge, both inline: a compaction that loses its commit to a produce now ends the whole maintenance pass, and the CHANGELOG's downgrade sentence doesn't say what timestamp lookups do on the older binary. The rest is optional.

Not inline:

• #847. sequential_timestamp_scan skips a batch whose header max_timestamp is below the target, and the index advances on the same header. Until #847 lands, a CreateTime batch's header is whatever the producer wrote, so a header that understates its records hides them from the lookup. Kafka's LogValidator prevents this by recomputing the header, and #847 does the same in the shared produce path. It would be good to land #847 before this PR or with it.
• Follow-ups. Three items are deferred: running the migration from maintain (a partition that is never written again stays legacy, and every timestamp lookup on it scans from the log start), returning NO_TIMESTAMP for Earliest and Latest on every backend, and a per-partition error for a corrupt batch on both engines (your note from the last round). Could you open an issue for each, so they don't live only in this thread?
• Optional. Four of the five new commits fix the same mistake: a read through self.db where the operation commits through a transaction or answers from a snapshot. In slatedb 0.14, Db, DbSnapshot and DbTransaction all implement DbReadOps, so the read helpers could take R: DbReadOps. Then policy_delete, delete_records and delete_topic, which still scan self.db inside a transaction, could move to tx.scan_prefix. Those three are correct today only because their watermark read conflicts. Fine as a follow-up.

Earlier points:

• Compaction race (storage.rs:804): fixed.
• Straddling batch (:402): fixed, with a test.
• Migration (:122): 3 is fixed. On 4, I agree with keeping the downgrade path, and the exact-length check covers the case I was worried about. 1 is partly done with the info! line; the rest of 1, and 2, are the follow-up above.
• latest_indexed_timestamp reset (:277): fixed.
• Earliest and Latest timestamps (:1799): fine as a follow-up.
• Comment wording (:197): fixed.
• Middle-entry test (tests.rs:1513): added. I agree a read counter isn't worth a Db wrapper.
• CHANGELOG, isolation level, corrupt batches and #838: done, with one CHANGELOG fix inline.

I compared against Kafka 3.9.1 by reading code; I didn't build or run this branch.