risingwave_hummock_sdk/
change_log.rs

1// Copyright 2024 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{HashMap, VecDeque};
16use std::ops::RangeBounds;
17
18use risingwave_common::catalog::TableId;
19use risingwave_pb::hummock::{PbEpochNewChangeLog, PbSstableInfo, PbTableChangeLog};
20use tracing::warn;
21
22use crate::HummockObjectId;
23use crate::sstable_info::SstableInfo;
24use crate::version::ObjectIdReader;
25
26#[derive(Debug, Clone, PartialEq)]
27pub struct TableChangeLogCommon<T>(
28    // older log at the front
29    VecDeque<EpochNewChangeLogCommon<T>>,
30);
31
32impl<T> TableChangeLogCommon<T> {
33    pub fn new(logs: impl IntoIterator<Item = EpochNewChangeLogCommon<T>>) -> Self {
34        let logs = logs.into_iter().collect::<VecDeque<_>>();
35        debug_assert!(logs.iter().flat_map(|log| log.epochs()).is_sorted());
36        Self(logs)
37    }
38
39    pub fn iter(&self) -> impl Iterator<Item = &EpochNewChangeLogCommon<T>> {
40        self.0.iter()
41    }
42
43    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut EpochNewChangeLogCommon<T>> {
44        self.0.iter_mut()
45    }
46
47    pub fn first(&self) -> Option<&EpochNewChangeLogCommon<T>> {
48        self.0.front()
49    }
50
51    pub fn last(&self) -> Option<&EpochNewChangeLogCommon<T>> {
52        self.0.back()
53    }
54
55    pub fn get(&self, index: usize) -> Option<&EpochNewChangeLogCommon<T>> {
56        self.0.get(index)
57    }
58
59    pub fn range(
60        &self,
61        range: impl RangeBounds<usize>,
62    ) -> impl Iterator<Item = &EpochNewChangeLogCommon<T>> + '_ {
63        self.0.range(range)
64    }
65
66    pub fn add_change_log(&mut self, new_change_log: EpochNewChangeLogCommon<T>) {
67        if let Some(prev_log) = self.0.back() {
68            assert!(prev_log.checkpoint_epoch < new_change_log.first_epoch());
69        }
70        self.0.push_back(new_change_log);
71    }
72
73    pub fn epochs(&self) -> impl Iterator<Item = u64> + '_ {
74        self.0
75            .iter()
76            .flat_map(|epoch_change_log| epoch_change_log.epochs())
77    }
78
79    pub fn is_empty(&self) -> bool {
80        self.0.is_empty()
81    }
82
83    pub fn binary_search_by_checkpoint_epoch(&self, epoch: u64) -> Result<usize, usize> {
84        self.0
85            .binary_search_by_key(&epoch, |log| log.checkpoint_epoch)
86    }
87}
88
89impl<T> IntoIterator for TableChangeLogCommon<T> {
90    type Item = EpochNewChangeLogCommon<T>;
91
92    type IntoIter = impl Iterator<Item = EpochNewChangeLogCommon<T>>;
93
94    fn into_iter(self) -> Self::IntoIter {
95        self.0.into_iter()
96    }
97}
98
99pub type TableChangeLog = TableChangeLogCommon<SstableInfo>;
100pub type TableChangeLogs = HashMap<TableId, TableChangeLog>;
101
102impl TableChangeLog {
103    pub fn get_object_ids(&self) -> impl Iterator<Item = HummockObjectId> + '_ {
104        self.0.iter().flat_map(|c| {
105            c.old_value
106                .iter()
107                .chain(c.new_value.iter())
108                .map(|t| HummockObjectId::Sstable(t.object_id()))
109        })
110    }
111}
112
113#[derive(Debug, Clone, PartialEq)]
114pub struct EpochNewChangeLogCommon<T> {
115    pub new_value: Vec<T>,
116    pub old_value: Vec<T>,
117    // epochs are sorted in ascending order
118    pub non_checkpoint_epochs: Vec<u64>,
119    pub checkpoint_epoch: u64,
120}
121
122impl EpochNewChangeLog {
123    pub fn change_log_ssts(&self) -> impl Iterator<Item = &SstableInfo> + '_ {
124        self.new_value.iter().chain(self.old_value.iter())
125    }
126}
127
128pub(crate) fn resolve_pb_log_epochs(epochs: &Vec<u64>) -> (Vec<u64>, u64) {
129    (
130        Vec::from(&epochs[0..(epochs.len() - 1)]),
131        *epochs.last().expect("non-empty"),
132    )
133}
134
135impl<T> EpochNewChangeLogCommon<T> {
136    pub fn epochs(&self) -> impl Iterator<Item = u64> + '_ {
137        self.non_checkpoint_epochs
138            .iter()
139            .cloned()
140            .chain([self.checkpoint_epoch])
141    }
142
143    pub fn first_epoch(&self) -> u64 {
144        self.non_checkpoint_epochs
145            .first()
146            .cloned()
147            .unwrap_or(self.checkpoint_epoch)
148    }
149}
150
151pub type EpochNewChangeLog = EpochNewChangeLogCommon<SstableInfo>;
152
153impl<T> From<&EpochNewChangeLogCommon<T>> for PbEpochNewChangeLog
154where
155    PbSstableInfo: for<'a> From<&'a T>,
156{
157    fn from(val: &EpochNewChangeLogCommon<T>) -> Self {
158        Self {
159            new_value: val.new_value.iter().map(|a| a.into()).collect(),
160            old_value: val.old_value.iter().map(|a| a.into()).collect(),
161            epochs: val.epochs().collect(),
162        }
163    }
164}
165
166impl<T> From<&PbEpochNewChangeLog> for EpochNewChangeLogCommon<T>
167where
168    T: for<'a> From<&'a PbSstableInfo>,
169{
170    fn from(value: &PbEpochNewChangeLog) -> Self {
171        let (non_checkpoint_epochs, checkpoint_epoch) = resolve_pb_log_epochs(&value.epochs);
172        Self {
173            new_value: value.new_value.iter().map(|a| a.into()).collect(),
174            old_value: value.old_value.iter().map(|a| a.into()).collect(),
175            non_checkpoint_epochs,
176            checkpoint_epoch,
177        }
178    }
179}
180
181impl<T> From<EpochNewChangeLogCommon<T>> for PbEpochNewChangeLog
182where
183    PbSstableInfo: From<T>,
184{
185    fn from(val: EpochNewChangeLogCommon<T>) -> Self {
186        Self {
187            epochs: val.epochs().collect(),
188            new_value: val.new_value.into_iter().map(|a| a.into()).collect(),
189            old_value: val.old_value.into_iter().map(|a| a.into()).collect(),
190        }
191    }
192}
193
194impl<T> From<PbEpochNewChangeLog> for EpochNewChangeLogCommon<T>
195where
196    T: From<PbSstableInfo>,
197{
198    fn from(value: PbEpochNewChangeLog) -> Self {
199        let (non_checkpoint_epochs, checkpoint_epoch) = resolve_pb_log_epochs(&value.epochs);
200        Self {
201            new_value: value.new_value.into_iter().map(|a| a.into()).collect(),
202            old_value: value.old_value.into_iter().map(|a| a.into()).collect(),
203            non_checkpoint_epochs,
204            checkpoint_epoch,
205        }
206    }
207}
208
209impl<T> TableChangeLogCommon<T> {
210    pub fn filter_epoch(
211        &self,
212        (min_epoch, max_epoch): (u64, u64),
213    ) -> impl Iterator<Item = &EpochNewChangeLogCommon<T>> + '_ {
214        assert!(
215            min_epoch <= max_epoch,
216            "invalid epoch range: {min_epoch}..={max_epoch}"
217        );
218        let start = self
219            .0
220            .partition_point(|epoch_change_log| epoch_change_log.checkpoint_epoch < min_epoch);
221        let end = self
222            .0
223            .partition_point(|epoch_change_log| epoch_change_log.first_epoch() <= max_epoch);
224        self.0.range(start..end)
225    }
226
227    /// Get the `next_epoch` of the given `epoch`
228    /// Return:
229    ///     - Ok(Some(`next_epoch`)): the `next_epoch` of `epoch`
230    ///     - Ok(None): `next_epoch` of `epoch` is not added to change log yet
231    ///     - Err(()): `epoch` is not an existing or to exist one
232    #[expect(clippy::result_unit_err)]
233    pub fn next_epoch(&self, epoch: u64) -> Result<Option<u64>, ()> {
234        let start = self
235            .0
236            .partition_point(|epoch_change_log| epoch_change_log.checkpoint_epoch < epoch);
237        debug_assert!(
238            self.0
239                .range(start..)
240                .flat_map(|epoch_change_log| epoch_change_log.epochs())
241                .is_sorted()
242        );
243        let mut later_epochs = self
244            .0
245            .range(start..)
246            .flat_map(|epoch_change_log| epoch_change_log.epochs())
247            .skip_while(|log_epoch| *log_epoch < epoch);
248        if let Some(first_epoch) = later_epochs.next() {
249            assert!(
250                first_epoch >= epoch,
251                "first_epoch {} < epoch {}",
252                first_epoch,
253                epoch
254            );
255            if first_epoch != epoch {
256                return Err(());
257            }
258            if let Some(next_epoch) = later_epochs.next() {
259                assert!(
260                    next_epoch > epoch,
261                    "next_epoch {} not exceed epoch {}",
262                    next_epoch,
263                    epoch
264                );
265                Ok(Some(next_epoch))
266            } else {
267                // `epoch` is latest
268                Ok(None)
269            }
270        } else {
271            // all epochs are less than `epoch`
272            Ok(None)
273        }
274    }
275
276    pub fn truncate(&mut self, truncate_epoch: u64) {
277        while let Some(change_log) = self.0.front()
278            && change_log.checkpoint_epoch < truncate_epoch
279        {
280            let _change_log = self.0.pop_front().expect("non-empty");
281        }
282        if let Some(first_log) = self.0.front_mut() {
283            first_log
284                .non_checkpoint_epochs
285                .retain(|epoch| *epoch >= truncate_epoch);
286        }
287    }
288}
289
290impl<T> TableChangeLogCommon<T>
291where
292    PbSstableInfo: for<'a> From<&'a T>,
293{
294    pub fn to_protobuf(&self) -> PbTableChangeLog {
295        PbTableChangeLog {
296            change_logs: self.0.iter().map(|a| a.into()).collect(),
297        }
298    }
299}
300
301impl<T> TableChangeLogCommon<T>
302where
303    T: for<'a> From<&'a PbSstableInfo>,
304{
305    pub fn from_protobuf(val: &PbTableChangeLog) -> Self {
306        Self(val.change_logs.iter().map(|a| a.into()).collect())
307    }
308}
309
310impl<T> TableChangeLogCommon<T>
311where
312    T: From<PbSstableInfo>,
313{
314    pub fn from_protobuf_owned(val: PbTableChangeLog) -> Self {
315        Self(val.change_logs.into_iter().map(|a| a.into()).collect())
316    }
317}
318
319pub fn build_table_change_log_delta<'a>(
320    old_value_ssts: impl Iterator<Item = SstableInfo>,
321    new_value_ssts: impl Iterator<Item = &'a SstableInfo>,
322    epochs: &Vec<u64>,
323    log_store_table_ids: impl Iterator<Item = (TableId, u64)>,
324) -> HashMap<TableId, ChangeLogDelta> {
325    let mut table_change_log: HashMap<_, _> = log_store_table_ids
326        .map(|(table_id, truncate_epoch)| {
327            let (non_checkpoint_epochs, checkpoint_epoch) = resolve_pb_log_epochs(epochs);
328            (
329                table_id,
330                ChangeLogDelta {
331                    truncate_epoch,
332                    new_log: EpochNewChangeLog {
333                        new_value: vec![],
334                        old_value: vec![],
335                        non_checkpoint_epochs,
336                        checkpoint_epoch,
337                    },
338                },
339            )
340        })
341        .collect();
342    for sst in old_value_ssts {
343        for table_id in &sst.table_ids {
344            match table_change_log.get_mut(table_id) {
345                Some(log) => {
346                    log.new_log.old_value.push(sst.clone());
347                }
348                None => {
349                    warn!(%table_id, ?sst, "old value sst contains non-log-store table");
350                }
351            }
352        }
353    }
354    for sst in new_value_ssts {
355        for table_id in &sst.table_ids {
356            if let Some(log) = table_change_log.get_mut(table_id) {
357                log.new_log.new_value.push(sst.clone());
358            }
359        }
360    }
361    table_change_log
362}
363
364#[derive(Debug, PartialEq, Clone)]
365pub struct ChangeLogDeltaCommon<T> {
366    pub truncate_epoch: u64,
367    pub new_log: EpochNewChangeLogCommon<T>,
368}
369
370pub type ChangeLogDelta = ChangeLogDeltaCommon<SstableInfo>;
371
372#[cfg(test)]
373mod tests {
374    use itertools::Itertools;
375
376    use crate::change_log::{EpochNewChangeLog, TableChangeLogCommon};
377    use crate::sstable_info::SstableInfo;
378
379    #[test]
380    fn test_filter_epoch() {
381        let table_change_log = TableChangeLogCommon::<SstableInfo>::new([
382            EpochNewChangeLog {
383                new_value: vec![],
384                old_value: vec![],
385                non_checkpoint_epochs: vec![],
386                checkpoint_epoch: 2,
387            },
388            EpochNewChangeLog {
389                new_value: vec![],
390                old_value: vec![],
391                non_checkpoint_epochs: vec![3],
392                checkpoint_epoch: 4,
393            },
394            EpochNewChangeLog {
395                new_value: vec![],
396                old_value: vec![],
397                non_checkpoint_epochs: vec![],
398                checkpoint_epoch: 6,
399            },
400            EpochNewChangeLog {
401                new_value: vec![],
402                old_value: vec![],
403                non_checkpoint_epochs: vec![8],
404                checkpoint_epoch: 10,
405            },
406        ]);
407
408        let epochs = (1..=11).collect_vec();
409        for i in 0..epochs.len() {
410            for j in i..epochs.len() {
411                let min_epoch = epochs[i];
412                let max_epoch = epochs[j];
413                let expected = table_change_log
414                    .0
415                    .iter()
416                    .filter(|log| {
417                        min_epoch <= log.checkpoint_epoch && log.first_epoch() <= max_epoch
418                    })
419                    .cloned()
420                    .collect_vec();
421                let actual = table_change_log
422                    .filter_epoch((min_epoch, max_epoch))
423                    .cloned()
424                    .collect_vec();
425                assert_eq!(expected, actual, "{:?}", (min_epoch, max_epoch));
426            }
427        }
428
429        let existing_epochs = table_change_log.epochs().collect_vec();
430        assert!(existing_epochs.is_sorted());
431        for &epoch in &epochs {
432            let expected = match existing_epochs
433                .iter()
434                .position(|existing_epoch| *existing_epoch >= epoch)
435            {
436                None => {
437                    // all existing epochs are less than epoch
438                    Ok(None)
439                }
440                Some(i) => {
441                    let this_epoch = existing_epochs[i];
442                    assert!(this_epoch >= epoch);
443                    if this_epoch == epoch {
444                        if i + 1 == existing_epochs.len() {
445                            // epoch is the latest epoch
446                            Ok(None)
447                        } else {
448                            Ok(Some(existing_epochs[i + 1]))
449                        }
450                    } else {
451                        // epoch not a existing epoch
452                        Err(())
453                    }
454                }
455            };
456            assert_eq!(expected, table_change_log.next_epoch(epoch));
457        }
458    }
459
460    #[test]
461    fn test_truncate() {
462        let mut table_change_log = TableChangeLogCommon::<SstableInfo>::new([
463            EpochNewChangeLog {
464                new_value: vec![],
465                old_value: vec![],
466                non_checkpoint_epochs: vec![],
467                checkpoint_epoch: 1,
468            },
469            EpochNewChangeLog {
470                new_value: vec![],
471                old_value: vec![],
472                non_checkpoint_epochs: vec![],
473                checkpoint_epoch: 2,
474            },
475            EpochNewChangeLog {
476                new_value: vec![],
477                old_value: vec![],
478                non_checkpoint_epochs: vec![3],
479                checkpoint_epoch: 4,
480            },
481            EpochNewChangeLog {
482                new_value: vec![],
483                old_value: vec![],
484                non_checkpoint_epochs: vec![],
485                checkpoint_epoch: 5,
486            },
487        ]);
488        let origin_table_change_log = table_change_log.clone();
489        for truncate_epoch in 0..6 {
490            table_change_log.truncate(truncate_epoch);
491            let expected_table_change_log = TableChangeLogCommon(
492                origin_table_change_log
493                    .0
494                    .iter()
495                    .filter_map(|epoch_change_log| {
496                        let mut epoch_change_log = epoch_change_log.clone();
497                        epoch_change_log
498                            .non_checkpoint_epochs
499                            .retain(|epoch| *epoch >= truncate_epoch);
500                        if epoch_change_log.non_checkpoint_epochs.is_empty()
501                            && epoch_change_log.checkpoint_epoch < truncate_epoch
502                        {
503                            None
504                        } else {
505                            Some(epoch_change_log)
506                        }
507                    })
508                    .collect(),
509            );
510            assert_eq!(expected_table_change_log, table_change_log);
511        }
512    }
513}
risingwave_hummock_sdk/change_log.rs

risingwave_hummock_sdk/
change_log.rs
