lsm_tree/scan_since.rs
1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024-present, fjall-rs
3// Copyright (c) 2026-present, Dmitry Prudnikov
4
5//! Change-data-capture event stream for [`Tree::scan_since_seqno`].
6//!
7//! [`Tree::scan_since_seqno`]: crate::Tree::scan_since_seqno
8
9use crate::{SeqNo, Slice};
10
11/// A single change event emitted by [`Tree::scan_since_seqno`](crate::Tree::scan_since_seqno).
12///
13/// Each event carries the sequence number at which the change was committed.
14/// Events are emitted in increasing seqno order, so a downstream consumer
15/// (replica, Kafka connector, Debezium-style pipeline) can replay them in
16/// order to reconstruct the source's history. Superseded versions are **not**
17/// collapsed: a key updated three times after the target seqno yields three
18/// events, mirroring the source's full change history rather than just its
19/// latest visible state.
20///
21/// # Replay semantics
22///
23/// Applying events in seqno order reconstructs the state delta. An
24/// `Insert(K, V1, s=150)` followed by a `PointTombstone(K, s=200)` means "K was
25/// inserted with V1 at 150, then deleted at 200" — the net effect on a replica
26/// starting before 150 is "create K with V1, then delete K", matching the
27/// source.
28///
29/// # Merge operands
30///
31/// A store using a [`MergeOperator`](crate::MergeOperator) records partial
32/// updates as [`MergeOperand`](Self::MergeOperand) events rather than resolved
33/// values: the consumer applies the same merge operator to reproduce the
34/// source's state. Emitting a merge as an `Insert` would make a replica
35/// overwrite instead of merge, diverging from the source; resolving the merge
36/// chain here would require reading the full base+operand history and defeat
37/// the block-skip optimization, so the raw operand is surfaced instead.
38///
39/// # KV-separated (blob) values
40///
41/// When a value is stored out-of-line in a blob file, the blob is resolved and
42/// the real value is carried in the [`Insert`](Self::Insert) event, so the
43/// consumer never needs access to the source's blob files to replicate.
44#[derive(Clone, Debug, Eq, PartialEq)]
45pub enum ScanSinceEvent {
46 /// A record was written (or overwritten) at `seqno`.
47 ///
48 /// Covers both inline values and values resolved from a blob file.
49 Insert {
50 /// User key that was written.
51 key: Slice,
52 /// Value written at `seqno` (resolved from a blob file if the entry
53 /// was KV-separated).
54 value: Slice,
55 /// Sequence number at which the write was committed.
56 seqno: SeqNo,
57 },
58
59 /// A merge operand was written at `seqno`.
60 ///
61 /// The consumer must apply the source's [`MergeOperator`](crate::MergeOperator)
62 /// to combine this operand with the prior value / operands, exactly as the
63 /// source does.
64 MergeOperand {
65 /// User key the operand applies to.
66 key: Slice,
67 /// Raw merge operand bytes, to be combined via the merge operator.
68 operand: Slice,
69 /// Sequence number at which the operand was committed.
70 seqno: SeqNo,
71 },
72
73 /// A single key was deleted at `seqno` with a REGULAR tombstone.
74 PointTombstone {
75 /// User key that was deleted.
76 key: Slice,
77 /// Sequence number at which the deletion was committed.
78 seqno: SeqNo,
79 },
80
81 /// A single key was WEAK-deleted (single-delete) at `seqno`.
82 ///
83 /// Distinct from [`PointTombstone`](Self::PointTombstone) because the two
84 /// are observably different at the source: a weak tombstone annihilates
85 /// exactly its matching put during compaction and can then expose an
86 /// older value from another run, while a regular tombstone keeps hiding
87 /// it. A consumer replays this with its own weak delete (e.g.
88 /// [`AbstractTree::remove_weak`](crate::AbstractTree::remove_weak)) so
89 /// the replica reproduces the source's operation semantics.
90 WeakTombstone {
91 /// User key that was weak-deleted.
92 key: Slice,
93 /// Sequence number at which the deletion was committed.
94 seqno: SeqNo,
95 },
96
97 /// A half-open key range `[start_key, end_key)` was deleted at `seqno`.
98 RangeTombstone {
99 /// Inclusive lower bound of the deleted range.
100 start_key: Slice,
101 /// Exclusive upper bound of the deleted range.
102 end_key: Slice,
103 /// Sequence number at which the range deletion was committed.
104 seqno: SeqNo,
105 },
106}
107
108impl ScanSinceEvent {
109 /// Sequence number at which this change was committed.
110 ///
111 /// Events from [`Tree::scan_since_seqno`](crate::Tree::scan_since_seqno)
112 /// arrive in increasing order of this value.
113 #[must_use]
114 pub fn seqno(&self) -> SeqNo {
115 match self {
116 Self::Insert { seqno, .. }
117 | Self::MergeOperand { seqno, .. }
118 | Self::PointTombstone { seqno, .. }
119 | Self::WeakTombstone { seqno, .. }
120 | Self::RangeTombstone { seqno, .. } => *seqno,
121 }
122 }
123
124 /// The user key this change applies to — the START key for a range
125 /// deletion, which is where its own ordering is anchored.
126 #[must_use]
127 pub fn key(&self) -> &Slice {
128 match self {
129 Self::Insert { key, .. }
130 | Self::MergeOperand { key, .. }
131 | Self::PointTombstone { key, .. }
132 | Self::WeakTombstone { key, .. } => key,
133 Self::RangeTombstone { start_key, .. } => start_key,
134 }
135 }
136
137 /// Total order that brings byte-identical events ADJACENT so one pass can
138 /// count them: seqno, then kind, then the full payload. Identical copies
139 /// are a real post-repair state — a manifest-loss repair publishes every
140 /// surviving SST, including both the inputs and outputs of a compaction
141 /// that crashed before deleting its inputs — and each copy carries the same
142 /// key, value, and seqno.
143 ///
144 /// This is NOT the emitted order. Events sharing a seqno are emitted oldest
145 /// SOURCE first, so the value the tree serves is replayed last; deciding
146 /// that by payload bytes would hand precedence to byte order.
147 pub(crate) fn grouping_order(&self, other: &Self) -> core::cmp::Ordering {
148 // A RANGE DELETION sorts before everything at its own seqno, and that
149 // ordering survives into the emitted stream: suppression is strictly
150 // `entry.seqno < tombstone.seqno`, so the tree KEEPS an entry written at
151 // the tombstone's own seqno, and a replay that applied the deletion last
152 // would drop it. The remaining kinds only need a stable, deterministic
153 // order — they touch one key each, so their relative position at one
154 // seqno cannot change the state a consumer converges to.
155 fn rank(e: &ScanSinceEvent) -> u8 {
156 match e {
157 ScanSinceEvent::RangeTombstone { .. } => 0,
158 ScanSinceEvent::Insert { .. } => 1,
159 ScanSinceEvent::MergeOperand { .. } => 2,
160 ScanSinceEvent::PointTombstone { .. } => 3,
161 ScanSinceEvent::WeakTombstone { .. } => 4,
162 }
163 }
164 fn payload(e: &ScanSinceEvent) -> (&Slice, Option<&Slice>) {
165 match e {
166 ScanSinceEvent::Insert { key, value, .. } => (key, Some(value)),
167 ScanSinceEvent::MergeOperand { key, operand, .. } => (key, Some(operand)),
168 ScanSinceEvent::PointTombstone { key, .. }
169 | ScanSinceEvent::WeakTombstone { key, .. } => (key, None),
170 ScanSinceEvent::RangeTombstone {
171 start_key, end_key, ..
172 } => (start_key, Some(end_key)),
173 }
174 }
175 self.seqno()
176 .cmp(&other.seqno())
177 .then_with(|| rank(self).cmp(&rank(other)))
178 .then_with(|| payload(self).cmp(&payload(other)))
179 }
180}