risingwave_meta/hummock/manager/
transaction.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::{BTreeMap, HashMap, HashSet};
16use std::ops::{Deref, DerefMut};
17use std::sync::Arc;
18
19use parking_lot::Mutex;
20use risingwave_common::catalog::TableId;
21use risingwave_hummock_sdk::change_log::{ChangeLogDelta, TableChangeLog};
22use risingwave_hummock_sdk::compaction_group::StaticCompactionGroupId;
23use risingwave_hummock_sdk::sstable_info::SstableInfo;
24use risingwave_hummock_sdk::table_watermark::TableWatermarks;
25use risingwave_hummock_sdk::vector_index::VectorIndexDelta;
26use risingwave_hummock_sdk::version::{GroupDelta, HummockVersion, HummockVersionDelta};
27use risingwave_hummock_sdk::{
28    CompactionGroupId, FrontendHummockVersionDelta, HummockSstableId, HummockVersionId,
29};
30use risingwave_meta_model::Epoch;
31use risingwave_pb::hummock::{
32    CompatibilityVersion, GroupConstruct, HummockVersionDeltas, HummockVersionStats,
33    StateTableInfoDelta,
34};
35use risingwave_pb::meta::subscribe_response::{Info, Operation};
36use sea_orm::{ConnectionTrait, EntityTrait};
37
38use super::TableCommittedEpochNotifiers;
39use crate::hummock::model::CompactionGroup;
40use crate::hummock::model::ext::to_table_change_log_meta_store_model;
41use crate::manager::{MetaOpts, NotificationManager};
42use crate::model::{
43    InMemValTransaction, MetadataModelResult, Transactional, ValTransaction, VarTransaction,
44};
45use crate::rpc::metrics::MetaMetrics;
46
47fn trigger_delta_log_stats(metrics: &MetaMetrics, total_number: usize) {
48    metrics.delta_log_count.set(total_number as _);
49}
50
51#[derive(Default)]
52struct TableChangeLogTransactionDelta {
53    updates: Vec<HashMap<TableId, ChangeLogDelta>>,
54    delete_all: HashSet<TableId>,
55}
56
57pub(super) fn trigger_version_stat(metrics: &MetaMetrics, current_version: &HummockVersion) {
58    metrics
59        .version_size
60        .set(current_version.estimated_encode_len() as i64);
61    metrics
62        .current_version_id
63        .set(current_version.id.as_i64_id());
64}
65
66pub(super) struct HummockVersionTransaction<'a> {
67    orig_version: &'a mut Arc<HummockVersion>,
68    orig_deltas: &'a mut BTreeMap<HummockVersionId, HummockVersionDelta>,
69    orig_table_change_log: &'a mut HashMap<TableId, TableChangeLog>,
70    notification_manager: &'a NotificationManager,
71    table_committed_epoch_notifiers: Option<&'a Mutex<TableCommittedEpochNotifiers>>,
72    meta_metrics: &'a MetaMetrics,
73    version_stat_tx: &'a tokio::sync::mpsc::UnboundedSender<Arc<HummockVersion>>,
74
75    pre_applied_version: Option<(
76        HummockVersion,
77        Vec<HummockVersionDelta>,
78        TableChangeLogTransactionDelta,
79    )>,
80    disable_apply_to_txn: bool,
81    opts: &'a MetaOpts,
82}
83
84impl<'a> HummockVersionTransaction<'a> {
85    pub(super) fn new(
86        version: &'a mut Arc<HummockVersion>,
87        deltas: &'a mut BTreeMap<HummockVersionId, HummockVersionDelta>,
88        table_change_log: &'a mut HashMap<TableId, TableChangeLog>,
89        notification_manager: &'a NotificationManager,
90        table_committed_epoch_notifiers: Option<&'a Mutex<TableCommittedEpochNotifiers>>,
91        meta_metrics: &'a MetaMetrics,
92        opts: &'a MetaOpts,
93        version_stat_tx: &'a tokio::sync::mpsc::UnboundedSender<Arc<HummockVersion>>,
94    ) -> Self {
95        Self {
96            orig_version: version,
97            orig_deltas: deltas,
98            orig_table_change_log: table_change_log,
99            pre_applied_version: None,
100            disable_apply_to_txn: false,
101            notification_manager,
102            table_committed_epoch_notifiers,
103            meta_metrics,
104            opts,
105            version_stat_tx,
106        }
107    }
108
109    pub(super) fn disable_apply_to_txn(&mut self) {
110        assert!(
111            self.pre_applied_version.is_none(),
112            "should only call disable at the beginning of txn"
113        );
114        self.disable_apply_to_txn = true;
115    }
116
117    pub(super) fn latest_version(&self) -> &HummockVersion {
118        if let Some((version, _, _)) = &self.pre_applied_version {
119            version
120        } else {
121            self.orig_version.as_ref()
122        }
123    }
124
125    pub(super) fn new_delta<'b>(&'b mut self) -> SingleDeltaTransaction<'a, 'b> {
126        let delta = self.latest_version().version_delta_after();
127        SingleDeltaTransaction {
128            version_txn: self,
129            delta: Some((delta, HashMap::new())),
130        }
131    }
132
133    fn pre_apply(
134        &mut self,
135        delta: HummockVersionDelta,
136        change_log_updates: HashMap<TableId, ChangeLogDelta>,
137    ) {
138        let (version, deltas, table_change_log_delta) =
139            self.pre_applied_version.get_or_insert_with(|| {
140                (
141                    self.orig_version.as_ref().clone(),
142                    Vec::with_capacity(1),
143                    TableChangeLogTransactionDelta::default(),
144                )
145            });
146        let changed_table_info = version.apply_version_delta(&delta);
147        // The in-memory change logs are only updated after the metastore transaction succeeds, so
148        // complete deletion is derived from the original state. A deletion that becomes eligible
149        // after multiple deltas in one transaction may be deferred to the next transaction.
150        let delete_all = HummockVersion::collect_gc_change_log_delta(
151            self.orig_table_change_log.keys(),
152            &change_log_updates,
153            &delta.removed_table_ids,
154            &delta.state_table_info_delta,
155            &changed_table_info,
156        );
157        table_change_log_delta.delete_all.extend(delete_all);
158        if !change_log_updates.is_empty() {
159            table_change_log_delta.updates.push(change_log_updates);
160        }
161        deltas.push(delta);
162    }
163
164    /// Returns a duplicate delta, used by time travel.
165    pub(super) fn pre_commit_epoch(
166        &mut self,
167        tables_to_commit: &HashMap<TableId, u64>,
168        new_compaction_groups: Vec<CompactionGroup>,
169        group_id_to_sub_levels: BTreeMap<CompactionGroupId, Vec<Vec<SstableInfo>>>,
170        new_table_ids: &HashMap<TableId, CompactionGroupId>,
171        new_table_watermarks: HashMap<TableId, TableWatermarks>,
172        change_log_delta: HashMap<TableId, ChangeLogDelta>,
173        vector_index_delta: HashMap<TableId, VectorIndexDelta>,
174        group_id_to_truncate_tables: HashMap<CompactionGroupId, HashSet<TableId>>,
175    ) -> HummockVersionDelta {
176        let mut new_version_delta = self.new_delta();
177        new_version_delta.new_table_watermarks = new_table_watermarks;
178        new_version_delta.set_change_log_delta(change_log_delta);
179        new_version_delta.vector_index_delta = vector_index_delta;
180
181        for compaction_group in &new_compaction_groups {
182            let group_deltas = &mut new_version_delta
183                .group_deltas
184                .entry(compaction_group.group_id())
185                .or_default()
186                .group_deltas;
187
188            #[expect(deprecated)]
189            group_deltas.push(GroupDelta::GroupConstruct(Box::new(GroupConstruct {
190                group_config: Some(compaction_group.compaction_config().as_ref().clone()),
191                group_id: compaction_group.group_id(),
192                parent_group_id: StaticCompactionGroupId::NewCompactionGroup as CompactionGroupId,
193                new_sst_start_id: HummockSstableId::default(), // No need to set it when `NewCompactionGroup`
194                table_ids: vec![],
195                version: CompatibilityVersion::LATEST as _,
196                split_key: None,
197            })));
198        }
199
200        // Append SSTs to a new version.
201        for (compaction_group_id, sub_levels) in group_id_to_sub_levels {
202            let group_deltas = &mut new_version_delta
203                .group_deltas
204                .entry(compaction_group_id)
205                .or_default()
206                .group_deltas;
207
208            for sub_level in sub_levels {
209                group_deltas.push(GroupDelta::NewL0SubLevel(sub_level));
210            }
211        }
212
213        for (compaction_group_id, table_ids) in group_id_to_truncate_tables {
214            let group_deltas = &mut new_version_delta
215                .group_deltas
216                .entry(compaction_group_id)
217                .or_default()
218                .group_deltas;
219
220            group_deltas.push(GroupDelta::PruneTableIdsFromSsts(
221                table_ids.into_iter().collect(),
222            ));
223        }
224
225        // update state table info
226        new_version_delta.with_latest_version(|version, delta| {
227            for (table_id, cg_id) in new_table_ids {
228                assert!(
229                    !version.state_table_info.info().contains_key(table_id),
230                    "newly added table exists previously: {:?}",
231                    table_id
232                );
233                let committed_epoch = *tables_to_commit.get(table_id).expect("newly added table must exist in tables_to_commit");
234                delta.state_table_info_delta.insert(
235                    *table_id,
236                    StateTableInfoDelta {
237                        committed_epoch,
238                        compaction_group_id: *cg_id,
239                    },
240                );
241            }
242
243            for (table_id, committed_epoch) in tables_to_commit {
244                if new_table_ids.contains_key(table_id) {
245                    continue;
246                }
247                let info = version.state_table_info.info().get(table_id).unwrap_or_else(|| {
248                    panic!("tables_to_commit {:?} contains table_id {} that is not newly added but not exists previously", tables_to_commit, table_id);
249                });
250                assert!(delta
251                    .state_table_info_delta
252                    .insert(
253                        *table_id,
254                        StateTableInfoDelta {
255                            committed_epoch: *committed_epoch,
256                            compaction_group_id: info.compaction_group_id,
257                        }
258                    )
259                    .is_none());
260            }
261        });
262
263        let time_travel_delta = (*new_version_delta).clone();
264        new_version_delta.pre_apply();
265        time_travel_delta
266    }
267}
268
269impl InMemValTransaction for HummockVersionTransaction<'_> {
270    fn commit(self) {
271        if let Some((version, deltas, table_change_log_delta)) = self.pre_applied_version {
272            *self.orig_version = Arc::new(version);
273            for change_log_delta in table_change_log_delta.updates {
274                HummockVersion::apply_change_log_delta(
275                    self.orig_table_change_log,
276                    &change_log_delta,
277                );
278            }
279            self.orig_table_change_log
280                .retain(|table_id, _| !table_change_log_delta.delete_all.contains(table_id));
281
282            if !self.disable_apply_to_txn {
283                let pb_deltas = deltas.iter().map(|delta| delta.to_protobuf()).collect();
284                self.notification_manager.notify_hummock_without_version(
285                    Operation::Add,
286                    Info::HummockVersionDeltas(risingwave_pb::hummock::HummockVersionDeltas {
287                        version_deltas: pb_deltas,
288                    }),
289                );
290                self.notification_manager.notify_frontend_without_version(
291                    Operation::Update,
292                    Info::HummockVersionDeltas(HummockVersionDeltas {
293                        version_deltas: deltas
294                            .iter()
295                            .map(|delta| {
296                                FrontendHummockVersionDelta::from_delta(delta).to_protobuf()
297                            })
298                            .collect(),
299                    }),
300                );
301                if let Some(table_committed_epoch_notifiers) = self.table_committed_epoch_notifiers
302                {
303                    table_committed_epoch_notifiers
304                        .lock()
305                        .notify_deltas(&deltas);
306                }
307            }
308
309            for delta in deltas {
310                assert!(self.orig_deltas.insert(delta.id, delta.clone()).is_none());
311            }
312
313            trigger_delta_log_stats(self.meta_metrics, self.orig_deltas.len());
314            let _ = self.version_stat_tx.send(self.orig_version.clone());
315        }
316    }
317}
318
319impl<TXN> ValTransaction<TXN> for HummockVersionTransaction<'_>
320where
321    TXN: ConnectionTrait,
322    HummockVersionDelta: Transactional<TXN>,
323    HummockVersionStats: Transactional<TXN>,
324{
325    async fn apply_to_txn(&self, txn: &mut TXN) -> MetadataModelResult<()> {
326        if self.disable_apply_to_txn {
327            return Ok(());
328        }
329        if let Some((_, deltas, table_change_log_delta)) = &self.pre_applied_version {
330            // These upsert_in_transaction can be batched. However, we know len(deltas) is always 1 currently.
331            for delta in deltas {
332                delta.upsert_in_transaction(txn).await?;
333            }
334
335            let insert_batch_size = self.opts.table_change_log_insert_batch_size as usize;
336            use futures::stream::{self, StreamExt};
337            use sea_orm::{ColumnTrait, Condition, QueryFilter};
338            let insert_iter = table_change_log_delta
339                .updates
340                .iter()
341                .flat_map(|updates| updates.iter())
342                .map(|(table_id, change_log_delta)| (*table_id, &change_log_delta.new_log));
343            let mut stream = stream::iter(insert_iter).chunks(insert_batch_size);
344            while let Some(change_log_batch) = stream.next().await {
345                let insert_many = change_log_batch
346                    .into_iter()
347                    .map(|(table_id, change_log)| {
348                        to_table_change_log_meta_store_model(table_id, change_log)
349                    })
350                    .collect::<Vec<_>>();
351                risingwave_meta_model::hummock_table_change_log::Entity::insert_many(insert_many)
352                    .on_empty_do_nothing()
353                    .exec(txn)
354                    .await?;
355            }
356
357            let delete_batch_size = self.opts.table_change_log_delete_batch_size as usize;
358            // `None` means deleting the whole table change log. Do not encode this as
359            // `u64::MAX`: the meta store epoch type is signed, so casting the sentinel would
360            // overflow and make the SQL predicate match nothing.
361            let delete_iter = table_change_log_delta
362                .updates
363                .iter()
364                .flat_map(|updates| updates.iter())
365                .map(|(table_id, change_log_delta)| {
366                    (*table_id, Some(change_log_delta.truncate_epoch))
367                })
368                .chain(
369                    table_change_log_delta
370                        .delete_all
371                        .iter()
372                        .map(|table_id| (*table_id, None)),
373                );
374
375            let mut stream = stream::iter(delete_iter).chunks(delete_batch_size);
376            while let Some(change_log_batch) = stream.next().await {
377                let mut condition = Condition::any();
378                for (table_id, truncate_epoch) in change_log_batch {
379                    let mut table_condition = Condition::all().add(
380                        risingwave_meta_model::hummock_table_change_log::Column::TableId
381                            .eq(table_id),
382                    );
383                    if let Some(truncate_epoch) = truncate_epoch {
384                        table_condition = table_condition.add(
385                            risingwave_meta_model::hummock_table_change_log::Column::CheckpointEpoch
386                                .lt(truncate_epoch as Epoch),
387                        );
388                    }
389                    condition = condition.add(table_condition);
390                }
391                risingwave_meta_model::hummock_table_change_log::Entity::delete_many()
392                    .filter(condition)
393                    .exec(txn)
394                    .await?;
395            }
396        }
397        Ok(())
398    }
399}
400
401pub(super) struct SingleDeltaTransaction<'a, 'b> {
402    version_txn: &'b mut HummockVersionTransaction<'a>,
403    delta: Option<(HummockVersionDelta, HashMap<TableId, ChangeLogDelta>)>,
404}
405
406impl SingleDeltaTransaction<'_, '_> {
407    pub(super) fn latest_version(&self) -> &HummockVersion {
408        self.version_txn.latest_version()
409    }
410
411    fn set_change_log_delta(&mut self, change_log_delta: HashMap<TableId, ChangeLogDelta>) {
412        self.delta.as_mut().expect("should exist").1 = change_log_delta;
413    }
414
415    pub(super) fn pre_apply(mut self) {
416        let (delta, change_log_delta) = self.delta.take().unwrap();
417        self.version_txn.pre_apply(delta, change_log_delta);
418    }
419
420    pub(super) fn with_latest_version(
421        &mut self,
422        f: impl FnOnce(&HummockVersion, &mut HummockVersionDelta),
423    ) {
424        f(
425            self.version_txn.latest_version(),
426            &mut self.delta.as_mut().expect("should exist").0,
427        )
428    }
429}
430
431impl Deref for SingleDeltaTransaction<'_, '_> {
432    type Target = HummockVersionDelta;
433
434    fn deref(&self) -> &Self::Target {
435        &self.delta.as_ref().expect("should exist").0
436    }
437}
438
439impl DerefMut for SingleDeltaTransaction<'_, '_> {
440    fn deref_mut(&mut self) -> &mut Self::Target {
441        &mut self.delta.as_mut().expect("should exist").0
442    }
443}
444
445impl Drop for SingleDeltaTransaction<'_, '_> {
446    fn drop(&mut self) {
447        if let Some((delta, change_log_delta)) = self.delta.take() {
448            self.version_txn.pre_apply(delta, change_log_delta);
449        }
450    }
451}
452
453pub(super) struct HummockVersionStatsTransaction<'a> {
454    stats: VarTransaction<'a, HummockVersionStats>,
455    notification_manager: &'a NotificationManager,
456}
457
458impl<'a> HummockVersionStatsTransaction<'a> {
459    pub(super) fn new(
460        stats: &'a mut HummockVersionStats,
461        notification_manager: &'a NotificationManager,
462    ) -> Self {
463        Self {
464            stats: VarTransaction::new(stats),
465            notification_manager,
466        }
467    }
468}
469
470impl InMemValTransaction for HummockVersionStatsTransaction<'_> {
471    fn commit(self) {
472        if self.stats.has_new_value() {
473            let stats = self.stats.clone();
474            self.stats.commit();
475            self.notification_manager
476                .notify_frontend_without_version(Operation::Update, Info::HummockStats(stats));
477        }
478    }
479}
480
481impl<TXN> ValTransaction<TXN> for HummockVersionStatsTransaction<'_>
482where
483    TXN: ConnectionTrait,
484    HummockVersionStats: Transactional<TXN>,
485{
486    async fn apply_to_txn(&self, txn: &mut TXN) -> MetadataModelResult<()> {
487        self.stats.apply_to_txn(txn).await
488    }
489}
490
491impl Deref for HummockVersionStatsTransaction<'_> {
492    type Target = HummockVersionStats;
493
494    fn deref(&self) -> &Self::Target {
495        self.stats.deref()
496    }
497}
498
499impl DerefMut for HummockVersionStatsTransaction<'_> {
500    fn deref_mut(&mut self) -> &mut Self::Target {
501        self.stats.deref_mut()
502    }
503}
504
505#[cfg(test)]
506mod tests {
507    use itertools::Itertools;
508    use risingwave_hummock_sdk::change_log::EpochNewChangeLog;
509    use risingwave_pb::hummock::StateTableInfo;
510
511    use super::*;
512
513    fn change_log_delta(checkpoint_epoch: u64, truncate_epoch: u64) -> ChangeLogDelta {
514        ChangeLogDelta {
515            truncate_epoch,
516            new_log: EpochNewChangeLog {
517                new_value: vec![],
518                old_value: vec![],
519                non_checkpoint_epochs: vec![],
520                checkpoint_epoch,
521            },
522        }
523    }
524
525    #[test]
526    fn test_apply_change_log_delta() {
527        let table_id = TableId::new(1);
528        let mut table_change_logs = HashMap::from([(
529            table_id,
530            TableChangeLog::new([
531                change_log_delta(1, 0).new_log,
532                change_log_delta(2, 0).new_log,
533            ]),
534        )]);
535        HummockVersion::apply_change_log_delta(
536            &mut table_change_logs,
537            &HashMap::from([(table_id, change_log_delta(3, 2))]),
538        );
539
540        assert_eq!(
541            table_change_logs[&table_id].epochs().collect_vec(),
542            vec![2, 3]
543        );
544    }
545
546    #[test]
547    fn test_collect_gc_change_log_delta() {
548        let table_id = TableId::new(1);
549        let removed_table_id = TableId::new(2);
550        let current_table_ids = HashSet::from([table_id, removed_table_id]);
551        let state_table_info_delta = HashMap::from([(
552            table_id,
553            StateTableInfoDelta {
554                committed_epoch: 2,
555                compaction_group_id: 1.into(),
556            },
557        )]);
558        let changed_table_info = HashMap::from([(
559            table_id,
560            Some(StateTableInfo {
561                committed_epoch: 1,
562                compaction_group_id: 1.into(),
563            }),
564        )]);
565
566        assert_eq!(
567            HummockVersion::collect_gc_change_log_delta(
568                current_table_ids.iter(),
569                &HashMap::<TableId, ChangeLogDelta>::new(),
570                &HashSet::from([removed_table_id]),
571                &state_table_info_delta,
572                &changed_table_info,
573            ),
574            HashSet::from([table_id, removed_table_id])
575        );
576        assert_eq!(
577            HummockVersion::collect_gc_change_log_delta(
578                current_table_ids.iter(),
579                &HashMap::from([(table_id, change_log_delta(2, 0))]),
580                &HashSet::new(),
581                &state_table_info_delta,
582                &changed_table_info,
583            ),
584            HashSet::new()
585        );
586    }
587}
risingwave_meta/hummock/manager/transaction.rs

risingwave_meta/hummock/manager/
transaction.rs
