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

risingwave_hummock_sdk/
change_log.rs
