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
51pub(super) fn trigger_version_stat(metrics: &MetaMetrics, current_version: &HummockVersion) {
52    metrics
53        .version_size
54        .set(current_version.estimated_encode_len() as i64);
55    metrics
56        .current_version_id
57        .set(current_version.id.as_i64_id());
58}
59
60pub(super) struct HummockVersionTransaction<'a> {
61    orig_version: &'a mut Arc<HummockVersion>,
62    orig_deltas: &'a mut BTreeMap<HummockVersionId, HummockVersionDelta>,
63    orig_table_change_log: &'a mut HashMap<TableId, TableChangeLog>,
64    notification_manager: &'a NotificationManager,
65    table_committed_epoch_notifiers: Option<&'a Mutex<TableCommittedEpochNotifiers>>,
66    meta_metrics: &'a MetaMetrics,
67    version_stat_tx: &'a tokio::sync::mpsc::UnboundedSender<Arc<HummockVersion>>,
68
69    pre_applied_version: Option<(HummockVersion, Vec<HummockVersionDelta>, HashSet<TableId>)>,
70    disable_apply_to_txn: bool,
71    opts: &'a MetaOpts,
72}
73
74impl<'a> HummockVersionTransaction<'a> {
75    pub(super) fn new(
76        version: &'a mut Arc<HummockVersion>,
77        deltas: &'a mut BTreeMap<HummockVersionId, HummockVersionDelta>,
78        table_change_log: &'a mut HashMap<TableId, TableChangeLog>,
79        notification_manager: &'a NotificationManager,
80        table_committed_epoch_notifiers: Option<&'a Mutex<TableCommittedEpochNotifiers>>,
81        meta_metrics: &'a MetaMetrics,
82        opts: &'a MetaOpts,
83        version_stat_tx: &'a tokio::sync::mpsc::UnboundedSender<Arc<HummockVersion>>,
84    ) -> Self {
85        Self {
86            orig_version: version,
87            orig_deltas: deltas,
88            orig_table_change_log: table_change_log,
89            pre_applied_version: None,
90            disable_apply_to_txn: false,
91            notification_manager,
92            table_committed_epoch_notifiers,
93            meta_metrics,
94            opts,
95            version_stat_tx,
96        }
97    }
98
99    pub(super) fn disable_apply_to_txn(&mut self) {
100        assert!(
101            self.pre_applied_version.is_none(),
102            "should only call disable at the beginning of txn"
103        );
104        self.disable_apply_to_txn = true;
105    }
106
107    pub(super) fn latest_version(&self) -> &HummockVersion {
108        if let Some((version, _, _)) = &self.pre_applied_version {
109            version
110        } else {
111            self.orig_version.as_ref()
112        }
113    }
114
115    pub(super) fn new_delta<'b>(&'b mut self) -> SingleDeltaTransaction<'a, 'b> {
116        let delta = self.latest_version().version_delta_after();
117        SingleDeltaTransaction {
118            version_txn: self,
119            delta: Some(delta),
120        }
121    }
122
123    fn pre_apply(&mut self, delta: HummockVersionDelta) {
124        let (version, deltas, gc_change_log_deltas) =
125            self.pre_applied_version.get_or_insert_with(|| {
126                (
127                    self.orig_version.as_ref().clone(),
128                    Vec::with_capacity(1),
129                    HashSet::new(),
130                )
131            });
132        let changed_table_info = version.apply_version_delta(&delta);
133        // Ideally, the first parameter should be the cumulative state (orig_table_change_log + all applied deltas).
134        // However, currently, we use orig_table_change_log directly because deltas are only applied after a successful metastore write in the end of the transaction.
135        // Consequently, some table eligible for GC are not returned by collect_gc_change_log_delta and are deferred to the next transaction.
136        // This delay is acceptable and does not impact system correctness.
137        let gc_change_log_delta = HummockVersion::collect_gc_change_log_delta(
138            self.orig_table_change_log.keys(),
139            &delta.change_log_delta,
140            &delta.removed_table_ids,
141            &delta.state_table_info_delta,
142            &changed_table_info,
143        );
144        gc_change_log_deltas.extend(gc_change_log_delta);
145        deltas.push(delta);
146    }
147
148    /// Returns a duplicate delta, used by time travel.
149    pub(super) fn pre_commit_epoch(
150        &mut self,
151        tables_to_commit: &HashMap<TableId, u64>,
152        new_compaction_groups: Vec<CompactionGroup>,
153        group_id_to_sub_levels: BTreeMap<CompactionGroupId, Vec<Vec<SstableInfo>>>,
154        new_table_ids: &HashMap<TableId, CompactionGroupId>,
155        new_table_watermarks: HashMap<TableId, TableWatermarks>,
156        change_log_delta: HashMap<TableId, ChangeLogDelta>,
157        vector_index_delta: HashMap<TableId, VectorIndexDelta>,
158        group_id_to_truncate_tables: HashMap<CompactionGroupId, HashSet<TableId>>,
159    ) -> HummockVersionDelta {
160        let mut new_version_delta = self.new_delta();
161        new_version_delta.new_table_watermarks = new_table_watermarks;
162        new_version_delta.change_log_delta = change_log_delta;
163        new_version_delta.vector_index_delta = vector_index_delta;
164
165        for compaction_group in &new_compaction_groups {
166            let group_deltas = &mut new_version_delta
167                .group_deltas
168                .entry(compaction_group.group_id())
169                .or_default()
170                .group_deltas;
171
172            #[expect(deprecated)]
173            group_deltas.push(GroupDelta::GroupConstruct(Box::new(GroupConstruct {
174                group_config: Some(compaction_group.compaction_config().as_ref().clone()),
175                group_id: compaction_group.group_id(),
176                parent_group_id: StaticCompactionGroupId::NewCompactionGroup as CompactionGroupId,
177                new_sst_start_id: HummockSstableId::default(), // No need to set it when `NewCompactionGroup`
178                table_ids: vec![],
179                version: CompatibilityVersion::LATEST as _,
180                split_key: None,
181            })));
182        }
183
184        // Append SSTs to a new version.
185        for (compaction_group_id, sub_levels) in group_id_to_sub_levels {
186            let group_deltas = &mut new_version_delta
187                .group_deltas
188                .entry(compaction_group_id)
189                .or_default()
190                .group_deltas;
191
192            for sub_level in sub_levels {
193                group_deltas.push(GroupDelta::NewL0SubLevel(sub_level));
194            }
195        }
196
197        for (compaction_group_id, table_ids) in group_id_to_truncate_tables {
198            let group_deltas = &mut new_version_delta
199                .group_deltas
200                .entry(compaction_group_id)
201                .or_default()
202                .group_deltas;
203
204            group_deltas.push(GroupDelta::PruneTableIdsFromSsts(
205                table_ids.into_iter().collect(),
206            ));
207        }
208
209        // update state table info
210        new_version_delta.with_latest_version(|version, delta| {
211            for (table_id, cg_id) in new_table_ids {
212                assert!(
213                    !version.state_table_info.info().contains_key(table_id),
214                    "newly added table exists previously: {:?}",
215                    table_id
216                );
217                let committed_epoch = *tables_to_commit.get(table_id).expect("newly added table must exist in tables_to_commit");
218                delta.state_table_info_delta.insert(
219                    *table_id,
220                    StateTableInfoDelta {
221                        committed_epoch,
222                        compaction_group_id: *cg_id,
223                    },
224                );
225            }
226
227            for (table_id, committed_epoch) in tables_to_commit {
228                if new_table_ids.contains_key(table_id) {
229                    continue;
230                }
231                let info = version.state_table_info.info().get(table_id).unwrap_or_else(|| {
232                    panic!("tables_to_commit {:?} contains table_id {} that is not newly added but not exists previously", tables_to_commit, table_id);
233                });
234                assert!(delta
235                    .state_table_info_delta
236                    .insert(
237                        *table_id,
238                        StateTableInfoDelta {
239                            committed_epoch: *committed_epoch,
240                            compaction_group_id: info.compaction_group_id,
241                        }
242                    )
243                    .is_none());
244            }
245        });
246
247        let time_travel_delta = (*new_version_delta).clone();
248        new_version_delta.pre_apply();
249        time_travel_delta
250    }
251}
252
253impl InMemValTransaction for HummockVersionTransaction<'_> {
254    fn commit(self) {
255        if let Some((version, deltas, gc_change_log_deltas)) = self.pre_applied_version {
256            *self.orig_version = Arc::new(version);
257            for delta in &deltas {
258                HummockVersion::apply_change_log_delta(
259                    self.orig_table_change_log,
260                    &delta.change_log_delta,
261                );
262            }
263            self.orig_table_change_log
264                .retain(|table_id, _| !gc_change_log_deltas.contains(table_id));
265
266            if !self.disable_apply_to_txn {
267                let pb_deltas = deltas.iter().map(|delta| delta.to_protobuf()).collect();
268                self.notification_manager.notify_hummock_without_version(
269                    Operation::Add,
270                    Info::HummockVersionDeltas(risingwave_pb::hummock::HummockVersionDeltas {
271                        version_deltas: pb_deltas,
272                    }),
273                );
274                self.notification_manager.notify_frontend_without_version(
275                    Operation::Update,
276                    Info::HummockVersionDeltas(HummockVersionDeltas {
277                        version_deltas: deltas
278                            .iter()
279                            .map(|delta| {
280                                FrontendHummockVersionDelta::from_delta(delta).to_protobuf()
281                            })
282                            .collect(),
283                    }),
284                );
285                if let Some(table_committed_epoch_notifiers) = self.table_committed_epoch_notifiers
286                {
287                    table_committed_epoch_notifiers
288                        .lock()
289                        .notify_deltas(&deltas);
290                }
291            }
292
293            for delta in deltas {
294                assert!(self.orig_deltas.insert(delta.id, delta.clone()).is_none());
295            }
296
297            trigger_delta_log_stats(self.meta_metrics, self.orig_deltas.len());
298            let _ = self.version_stat_tx.send(self.orig_version.clone());
299        }
300    }
301}
302
303impl<TXN> ValTransaction<TXN> for HummockVersionTransaction<'_>
304where
305    TXN: ConnectionTrait,
306    HummockVersionDelta: Transactional<TXN>,
307    HummockVersionStats: Transactional<TXN>,
308{
309    async fn apply_to_txn(&self, txn: &mut TXN) -> MetadataModelResult<()> {
310        if self.disable_apply_to_txn {
311            return Ok(());
312        }
313        if let Some((_, deltas, gc_change_log_deltas)) = &self.pre_applied_version {
314            // These upsert_in_transaction can be batched. However, we know len(deltas) is always 1 currently.
315            for delta in deltas {
316                delta.upsert_in_transaction(txn).await?;
317            }
318
319            let insert_batch_size = self.opts.table_change_log_insert_batch_size as usize;
320            use futures::stream::{self, StreamExt};
321            use sea_orm::{ColumnTrait, Condition, QueryFilter};
322            let insert_iter = deltas
323                .iter()
324                .flat_map(|i| i.change_log_delta.iter())
325                .map(|(table_id, change_log_delta)| (*table_id, &change_log_delta.new_log));
326            let mut stream = stream::iter(insert_iter).chunks(insert_batch_size);
327            while let Some(change_log_batch) = stream.next().await {
328                let insert_many = change_log_batch
329                    .into_iter()
330                    .map(|(table_id, change_log)| {
331                        to_table_change_log_meta_store_model(table_id, change_log)
332                    })
333                    .collect::<Vec<_>>();
334                risingwave_meta_model::hummock_table_change_log::Entity::insert_many(insert_many)
335                    .on_empty_do_nothing()
336                    .exec(txn)
337                    .await?;
338            }
339
340            let delete_batch_size = self.opts.table_change_log_delete_batch_size as usize;
341            let delete_iter = deltas
342                .iter()
343                .flat_map(|i| i.change_log_delta.iter())
344                .map(|(table_id, change_log_delta)| (*table_id, change_log_delta.truncate_epoch))
345                .chain(
346                    gc_change_log_deltas
347                        .iter()
348                        .map(|table_id| (*table_id, u64::MAX)),
349                );
350
351            let mut stream = stream::iter(delete_iter).chunks(delete_batch_size);
352            while let Some(change_log_batch) = stream.next().await {
353                let mut condition = Condition::any();
354                for (table_id, truncate_epoch) in change_log_batch {
355                    condition = condition.add(
356                        Condition::all()
357                            .add(risingwave_meta_model::hummock_table_change_log::Column::TableId.eq(table_id))
358                            .add(risingwave_meta_model::hummock_table_change_log::Column::CheckpointEpoch.lt(truncate_epoch as Epoch))
359                    );
360                }
361                risingwave_meta_model::hummock_table_change_log::Entity::delete_many()
362                    .filter(condition)
363                    .exec(txn)
364                    .await?;
365            }
366        }
367        Ok(())
368    }
369}
370
371pub(super) struct SingleDeltaTransaction<'a, 'b> {
372    version_txn: &'b mut HummockVersionTransaction<'a>,
373    delta: Option<HummockVersionDelta>,
374}
375
376impl SingleDeltaTransaction<'_, '_> {
377    pub(super) fn latest_version(&self) -> &HummockVersion {
378        self.version_txn.latest_version()
379    }
380
381    pub(super) fn pre_apply(mut self) {
382        self.version_txn.pre_apply(self.delta.take().unwrap());
383    }
384
385    pub(super) fn with_latest_version(
386        &mut self,
387        f: impl FnOnce(&HummockVersion, &mut HummockVersionDelta),
388    ) {
389        f(
390            self.version_txn.latest_version(),
391            self.delta.as_mut().expect("should exist"),
392        )
393    }
394}
395
396impl Deref for SingleDeltaTransaction<'_, '_> {
397    type Target = HummockVersionDelta;
398
399    fn deref(&self) -> &Self::Target {
400        self.delta.as_ref().expect("should exist")
401    }
402}
403
404impl DerefMut for SingleDeltaTransaction<'_, '_> {
405    fn deref_mut(&mut self) -> &mut Self::Target {
406        self.delta.as_mut().expect("should exist")
407    }
408}
409
410impl Drop for SingleDeltaTransaction<'_, '_> {
411    fn drop(&mut self) {
412        if let Some(delta) = self.delta.take() {
413            self.version_txn.pre_apply(delta);
414        }
415    }
416}
417
418pub(super) struct HummockVersionStatsTransaction<'a> {
419    stats: VarTransaction<'a, HummockVersionStats>,
420    notification_manager: &'a NotificationManager,
421}
422
423impl<'a> HummockVersionStatsTransaction<'a> {
424    pub(super) fn new(
425        stats: &'a mut HummockVersionStats,
426        notification_manager: &'a NotificationManager,
427    ) -> Self {
428        Self {
429            stats: VarTransaction::new(stats),
430            notification_manager,
431        }
432    }
433}
434
435impl InMemValTransaction for HummockVersionStatsTransaction<'_> {
436    fn commit(self) {
437        if self.stats.has_new_value() {
438            let stats = self.stats.clone();
439            self.stats.commit();
440            self.notification_manager
441                .notify_frontend_without_version(Operation::Update, Info::HummockStats(stats));
442        }
443    }
444}
445
446impl<TXN> ValTransaction<TXN> for HummockVersionStatsTransaction<'_>
447where
448    TXN: ConnectionTrait,
449    HummockVersionStats: Transactional<TXN>,
450{
451    async fn apply_to_txn(&self, txn: &mut TXN) -> MetadataModelResult<()> {
452        self.stats.apply_to_txn(txn).await
453    }
454}
455
456impl Deref for HummockVersionStatsTransaction<'_> {
457    type Target = HummockVersionStats;
458
459    fn deref(&self) -> &Self::Target {
460        self.stats.deref()
461    }
462}
463
464impl DerefMut for HummockVersionStatsTransaction<'_> {
465    fn deref_mut(&mut self) -> &mut Self::Target {
466        self.stats.deref_mut()
467    }
468}
risingwave_meta/hummock/manager/transaction.rs

risingwave_meta/hummock/manager/
transaction.rs
