risingwave_meta/barrier/checkpoint/independent_job/
mod.rs

1// Copyright 2026 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, HashSet};
16use std::mem::take;
17
18use risingwave_common::catalog::TableId;
19use risingwave_common::util::epoch::Epoch;
20use risingwave_pb::id::FragmentId;
21use risingwave_pb::stream_plan::barrier::PbBarrierKind;
22
23pub(crate) mod batch_refresh_job;
24pub(crate) mod creating_job;
25
26pub(crate) use batch_refresh_job::{
27    BatchRefreshJobCheckpointControl, BatchRefreshJobTriggerContext, BatchRefreshLogicalFragments,
28    BatchRefreshRenderResult,
29};
30pub(crate) use creating_job::CreatingStreamingJobControl;
31
32use crate::barrier::info::BarrierInfo;
33use crate::barrier::notifier::Notifier;
34use crate::barrier::partial_graph::{CollectedBarrier, PartialGraphManager};
35use crate::barrier::{BackfillProgress, BarrierKind, FragmentBackfillProgress, TracedEpoch};
36use crate::controller::fragment::InflightFragmentInfo;
37
38/// Build a fake `BarrierInfo` for independent partial-graph barriers.
39///
40/// Shared by both `CreatingStreamingJobControl` and `BatchRefreshJobCheckpointControl`.
41fn new_fake_barrier(
42    prev_epoch_fake_physical_time: &mut u64,
43    pending_non_checkpoint_barriers: &mut Vec<u64>,
44    kind: PbBarrierKind,
45) -> BarrierInfo {
46    let prev_epoch = TracedEpoch::new(Epoch::from_physical_time(*prev_epoch_fake_physical_time));
47    *prev_epoch_fake_physical_time += 1;
48    let curr_epoch = TracedEpoch::new(Epoch::from_physical_time(*prev_epoch_fake_physical_time));
49    let kind = match kind {
50        PbBarrierKind::Unspecified => unreachable!(),
51        PbBarrierKind::Initial => {
52            assert!(pending_non_checkpoint_barriers.is_empty());
53            BarrierKind::Initial
54        }
55        PbBarrierKind::Barrier => {
56            pending_non_checkpoint_barriers.push(prev_epoch.value().0);
57            BarrierKind::Barrier
58        }
59        PbBarrierKind::Checkpoint => {
60            pending_non_checkpoint_barriers.push(prev_epoch.value().0);
61            BarrierKind::Checkpoint(take(pending_non_checkpoint_barriers))
62        }
63    };
64    BarrierInfo {
65        prev_epoch,
66        curr_epoch,
67        kind,
68    }
69}
70
71// ── Enum unifying independent checkpoint job types ──────────────────────────
72
73/// A streaming job that checkpoints independently from the database's main graph,
74/// using its own partial graph.
75pub(crate) enum IndependentCheckpointJobControl {
76    CreatingStreamingJob(CreatingStreamingJobControl),
77    BatchRefresh(BatchRefreshJobCheckpointControl),
78}
79
80impl IndependentCheckpointJobControl {
81    pub(crate) fn gen_backfill_progress(&self) -> Option<BackfillProgress> {
82        match self {
83            Self::CreatingStreamingJob(j) => Some(j.gen_backfill_progress()),
84            Self::BatchRefresh(j) => j.gen_backfill_progress(),
85        }
86    }
87
88    /// Collect a barrier and return whether a checkpoint should be forced in the next barrier.
89    pub(crate) fn collect(&mut self, collected_barrier: CollectedBarrier<'_>) -> bool {
90        match self {
91            Self::CreatingStreamingJob(j) => j.collect(collected_barrier),
92            Self::BatchRefresh(j) => j.collect(collected_barrier),
93        }
94    }
95
96    pub(crate) fn gen_fragment_backfill_progress(&self) -> Vec<FragmentBackfillProgress> {
97        match self {
98            Self::CreatingStreamingJob(j) => j.gen_fragment_backfill_progress(),
99            Self::BatchRefresh(j) => j.gen_fragment_backfill_progress(),
100        }
101    }
102
103    pub(crate) fn pinned_upstream_log_epoch(&self) -> (u64, HashSet<TableId>) {
104        match self {
105            Self::CreatingStreamingJob(j) => j.pinned_upstream_log_epoch(),
106            Self::BatchRefresh(j) => j.pinned_upstream_log_epoch(),
107        }
108    }
109
110    pub(crate) fn fragment_infos(&self) -> Option<&HashMap<FragmentId, InflightFragmentInfo>> {
111        match self {
112            Self::CreatingStreamingJob(j) => j.fragment_infos(),
113            Self::BatchRefresh(j) => j.fragment_infos(),
114        }
115    }
116
117    pub(crate) fn ack_completed(
118        &mut self,
119        partial_graph_manager: &mut PartialGraphManager,
120        epoch: u64,
121    ) {
122        match self {
123            Self::CreatingStreamingJob(j) => j.ack_completed(partial_graph_manager, epoch),
124            Self::BatchRefresh(j) => j.ack_completed(partial_graph_manager, epoch),
125        }
126    }
127
128    pub(crate) fn on_partial_graph_reset(self) {
129        match self {
130            Self::CreatingStreamingJob(j) => j.on_partial_graph_reset(),
131            Self::BatchRefresh(j) => j.on_partial_graph_reset(),
132        }
133    }
134
135    pub(crate) fn drop(
136        &mut self,
137        notifiers: &mut Vec<Notifier>,
138        partial_graph_manager: &mut PartialGraphManager,
139    ) -> bool {
140        match self {
141            Self::CreatingStreamingJob(j) => j.drop(notifiers, partial_graph_manager),
142            Self::BatchRefresh(j) => j.drop(notifiers, partial_graph_manager),
143        }
144    }
145
146    /// Reset during database recovery.
147    ///
148    /// Returns `true` if the partial graph was already resetting (from a prior drop),
149    /// meaning caller should not issue a new reset request.
150    pub(crate) fn reset(self) -> bool {
151        match self {
152            Self::CreatingStreamingJob(j) => j.reset(),
153            Self::BatchRefresh(j) => j.reset(),
154        }
155    }
156}
risingwave_meta/barrier/checkpoint/independent_job/mod.rs

risingwave_meta/barrier/checkpoint/independent_job/
mod.rs
