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::{DatabaseId, TableId};
19use risingwave_common::id::JobId;
20use risingwave_common::util::epoch::Epoch;
21use risingwave_meta_model::WorkerId;
22use risingwave_pb::hummock::HummockVersionStats;
23use risingwave_pb::id::{ActorId, FragmentId, PartialGraphId};
24use risingwave_pb::source::ConnectorSplits;
25use risingwave_pb::stream_plan::barrier::PbBarrierKind;
26use risingwave_pb::stream_plan::barrier_mutation::Mutation;
27use risingwave_pb::stream_plan::{
28    AddMutation, PbSubscriptionUpstreamInfo, StartFragmentBackfillMutation,
29};
30use risingwave_pb::stream_service::barrier_complete_response::CreateMviewProgress;
31
32pub(crate) mod batch_refresh_job;
33pub(crate) mod creating_job;
34pub(crate) mod iceberg_v3_job;
35pub(crate) use batch_refresh_job::{
36    BatchRefreshJobCheckpointControl, BatchRefreshJobTriggerContext, BatchRefreshLogicalFragments,
37    BatchRefreshRenderResult, RenderedIndependentJobActors,
38};
39pub(crate) use creating_job::CreatingStreamingJobControl;
40pub(crate) use iceberg_v3_job::{
41    IcebergV3JobCheckpointControl, IcebergV3RenderResult, is_iceberg_v3_fragment_nodes,
42};
43
44use crate::MetaResult;
45use crate::barrier::backfill_order_control::get_nodes_with_backfill_dependencies;
46use crate::barrier::command::{CreateStreamingJobCommandInfo, ThrottleConfigMap};
47use crate::barrier::context::CreateIndependentStreamingJobCommandInfo;
48use crate::barrier::info::BarrierInfo;
49use crate::barrier::partial_graph::{
50    CollectedBarrier, PartialGraphAdder, PartialGraphBarrierInfo, PartialGraphManager,
51};
52use crate::barrier::progress::CreateMviewProgressTracker;
53use crate::barrier::rpc::to_partial_graph_id;
54use crate::barrier::{
55    BackfillOrderState, BackfillProgress, BarrierKind, FragmentBackfillProgress, TracedEpoch,
56};
57use crate::controller::fragment::InflightFragmentInfo;
58use crate::manager::MetaOpts;
59use crate::model::StreamJobActorsToCreate;
60use crate::notification::{CollectionNotifier, NotifierStarter};
61use crate::stream::ExtendedFragmentBackfillOrder;
62
63#[derive(Debug)]
64struct SnapshotPhaseControl {
65    /// Duplicated from `IndependentJobInfo` so snapshot-phase transitions are self-contained.
66    snapshot_epoch: u64,
67    prev_epoch_fake_physical_time: u64,
68    version_stats: HummockVersionStats,
69    create_mview_tracker: CreateMviewProgressTracker,
70    pending_non_checkpoint_barriers: Vec<u64>,
71}
72
73impl SnapshotPhaseControl {
74    fn for_new_job(
75        snapshot_epoch: u64,
76        fragment_infos: &HashMap<FragmentId, InflightFragmentInfo>,
77        create_info: &CreateStreamingJobCommandInfo,
78        version_stats: &HummockVersionStats,
79    ) -> (Self, BarrierInfo) {
80        let job_id = create_info.stream_job_fragments.stream_job_id();
81        let backfill_order_state = BackfillOrderState::new(
82            &create_info.fragment_backfill_ordering,
83            fragment_infos,
84            create_info.locality_fragment_state_table_mapping.clone(),
85        );
86        let create_mview_tracker = CreateMviewProgressTracker::recover(
87            job_id,
88            fragment_infos,
89            backfill_order_state,
90            version_stats,
91        );
92        let mut snapshot = Self {
93            snapshot_epoch,
94            prev_epoch_fake_physical_time: 0,
95            version_stats: version_stats.clone(),
96            create_mview_tracker,
97            pending_non_checkpoint_barriers: vec![],
98        };
99        let initial_barrier_info = new_fake_barrier(
100            &mut snapshot.prev_epoch_fake_physical_time,
101            &mut snapshot.pending_non_checkpoint_barriers,
102            PbBarrierKind::Checkpoint,
103        );
104        assert!(snapshot.pending_non_checkpoint_barriers.is_empty());
105        (snapshot, initial_barrier_info)
106    }
107
108    fn for_recovery(
109        job_id: JobId,
110        snapshot_epoch: u64,
111        committed_epoch: u64,
112        fragment_infos: &HashMap<FragmentId, InflightFragmentInfo>,
113        backfill_order: &ExtendedFragmentBackfillOrder,
114        version_stats: &HummockVersionStats,
115    ) -> (Self, BarrierInfo) {
116        let backfill_order_state =
117            BackfillOrderState::recover_from_fragment_infos(backfill_order, fragment_infos);
118        let create_mview_tracker = CreateMviewProgressTracker::recover(
119            job_id,
120            fragment_infos,
121            backfill_order_state,
122            version_stats,
123        );
124        let mut snapshot = Self {
125            snapshot_epoch,
126            prev_epoch_fake_physical_time: Epoch(committed_epoch).physical_time(),
127            version_stats: version_stats.clone(),
128            create_mview_tracker,
129            pending_non_checkpoint_barriers: vec![],
130        };
131        let initial_barrier_info = new_fake_barrier(
132            &mut snapshot.prev_epoch_fake_physical_time,
133            &mut snapshot.pending_non_checkpoint_barriers,
134            PbBarrierKind::Initial,
135        );
136        assert!(snapshot.pending_non_checkpoint_barriers.is_empty());
137        (snapshot, initial_barrier_info)
138    }
139
140    fn apply_progress<'a>(
141        &mut self,
142        progress: impl IntoIterator<Item = &'a CreateMviewProgress>,
143    ) -> bool {
144        for progress in progress {
145            self.create_mview_tracker
146                .apply_progress(progress, &self.version_stats);
147        }
148        self.create_mview_tracker.is_finished()
149    }
150
151    fn take_start_backfill_mutation(&mut self) -> Option<Mutation> {
152        let fragment_ids = self
153            .create_mview_tracker
154            .take_pending_backfill_nodes()
155            .collect::<Vec<_>>();
156        (!fragment_ids.is_empty()).then_some(Mutation::StartFragmentBackfill(
157            StartFragmentBackfillMutation { fragment_ids },
158        ))
159    }
160
161    fn next_fake_barrier(&mut self, upstream_kind: &BarrierKind) -> BarrierInfo {
162        let kind = match upstream_kind {
163            BarrierKind::Barrier => PbBarrierKind::Barrier,
164            BarrierKind::Checkpoint(_) => PbBarrierKind::Checkpoint,
165            BarrierKind::Initial => unreachable!("upstream new epoch should not be initial"),
166        };
167        new_fake_barrier(
168            &mut self.prev_epoch_fake_physical_time,
169            &mut self.pending_non_checkpoint_barriers,
170            kind,
171        )
172    }
173
174    fn finish_snapshot_barrier(&mut self) -> BarrierInfo {
175        self.pending_non_checkpoint_barriers
176            .push(self.snapshot_epoch);
177        BarrierInfo {
178            curr_epoch: TracedEpoch::new(Epoch(self.snapshot_epoch)),
179            prev_epoch: TracedEpoch::new(Epoch::from_physical_time(
180                self.prev_epoch_fake_physical_time,
181            )),
182            kind: BarrierKind::Checkpoint(take(&mut self.pending_non_checkpoint_barriers)),
183        }
184    }
185}
186
187fn build_initial_add_mutation(
188    fragment_infos: &HashMap<FragmentId, InflightFragmentInfo>,
189    backfill_ordering: &ExtendedFragmentBackfillOrder,
190    actor_splits: HashMap<ActorId, ConnectorSplits>,
191) -> Mutation {
192    Mutation::Add(AddMutation {
193        actor_dispatchers: Default::default(),
194        added_actors: fragment_infos
195            .values()
196            .flat_map(|fragment| fragment.actors.keys().copied())
197            .collect(),
198        actor_splits,
199        pause: false,
200        subscriptions_to_add: Default::default(),
201        backfill_nodes_to_pause: get_nodes_with_backfill_dependencies(backfill_ordering)
202            .into_iter()
203            .collect(),
204        actor_cdc_table_snapshot_splits: None,
205        new_upstream_sinks: Default::default(),
206        dropped_actors: Default::default(),
207        sink_log_store_flush: Default::default(),
208    })
209}
210
211struct InitialPartialGraphRequest<'a> {
212    node_actors: &'a HashMap<WorkerId, HashSet<ActorId>>,
213    state_table_ids: &'a HashSet<TableId>,
214    barrier_info: BarrierInfo,
215    actors_to_create: StreamJobActorsToCreate,
216    mutation: Mutation,
217    notifier: Option<&'a mut NotifierStarter>,
218    create_info: CreateIndependentStreamingJobCommandInfo,
219}
220
221fn add_initial_partial_graph(
222    graph_adder: &mut PartialGraphAdder<'_>,
223    partial_graph_id: PartialGraphId,
224    request: InitialPartialGraphRequest<'_>,
225) -> MetaResult<()> {
226    let InitialPartialGraphRequest {
227        node_actors,
228        state_table_ids,
229        barrier_info,
230        actors_to_create,
231        mutation,
232        notifier,
233        create_info,
234    } = request;
235    graph_adder.manager().inject_barrier(
236        partial_graph_id,
237        Some(mutation),
238        node_actors,
239        state_table_ids.iter().copied(),
240        node_actors.keys().copied(),
241        Some(actors_to_create),
242        PartialGraphBarrierInfo::new(
243            create_info.into_post_collect(),
244            barrier_info,
245            notifier,
246            state_table_ids.clone(),
247        ),
248    )
249}
250
251fn snapshot_backfill_max_pending_barrier_num(opts: &MetaOpts) -> usize {
252    opts.in_flight_barrier_nums
253        .saturating_mul(opts.snapshot_backfill_barrier_amplification_factor.max(1))
254}
255
256#[derive(Debug)]
257pub(crate) struct IndependentJobInfo {
258    pub(crate) job_id: JobId,
259    pub(crate) partial_graph_id: PartialGraphId,
260    pub(crate) snapshot_backfill_upstream_tables: HashSet<TableId>,
261    pub(crate) snapshot_epoch: u64,
262    pub(crate) state_table_ids: HashSet<TableId>,
263}
264
265impl IndependentJobInfo {
266    pub(crate) fn from_fragment_infos<'a>(
267        database_id: DatabaseId,
268        job_id: JobId,
269        snapshot_epoch: u64,
270        snapshot_backfill_upstream_tables: HashSet<TableId>,
271        fragment_infos: impl IntoIterator<Item = &'a InflightFragmentInfo> + 'a,
272    ) -> Self {
273        Self {
274            job_id,
275            partial_graph_id: to_partial_graph_id(database_id, Some(job_id)),
276            snapshot_backfill_upstream_tables,
277            snapshot_epoch,
278            state_table_ids: InflightFragmentInfo::existing_table_ids(fragment_infos).collect(),
279        }
280    }
281}
282
283#[derive(Debug)]
284pub(crate) struct IndependentJobControl {
285    /// Immutable metadata shared by every phase of the independent job.
286    pub(crate) info: IndependentJobInfo,
287    /// Latest checkpoint persisted through the common complete-barrier path.
288    pub(crate) max_committed_epoch: Option<u64>,
289}
290
291impl IndependentJobControl {
292    pub(crate) fn new(info: IndependentJobInfo) -> Self {
293        Self {
294            info,
295            max_committed_epoch: None,
296        }
297    }
298
299    pub(crate) fn recovered(info: IndependentJobInfo, committed_epoch: u64) -> Self {
300        Self {
301            info,
302            max_committed_epoch: Some(committed_epoch),
303        }
304    }
305
306    pub(crate) fn ack_completed(&mut self, completed_epoch: u64) {
307        if let Some(previous_epoch) = self.max_committed_epoch.replace(completed_epoch) {
308            assert!(completed_epoch > previous_epoch);
309        }
310    }
311}
312
313/// Build a fake `BarrierInfo` for independent partial-graph barriers.
314///
315/// Shared by both `CreatingStreamingJobControl` and `BatchRefreshJobCheckpointControl`.
316fn new_fake_barrier(
317    prev_epoch_fake_physical_time: &mut u64,
318    pending_non_checkpoint_barriers: &mut Vec<u64>,
319    kind: PbBarrierKind,
320) -> BarrierInfo {
321    let prev_epoch = TracedEpoch::new(Epoch::from_physical_time(*prev_epoch_fake_physical_time));
322    *prev_epoch_fake_physical_time += 1;
323    let curr_epoch = TracedEpoch::new(Epoch::from_physical_time(*prev_epoch_fake_physical_time));
324    let kind = match kind {
325        PbBarrierKind::Unspecified => unreachable!(),
326        PbBarrierKind::Initial => {
327            assert!(pending_non_checkpoint_barriers.is_empty());
328            BarrierKind::Initial
329        }
330        PbBarrierKind::Barrier => {
331            pending_non_checkpoint_barriers.push(prev_epoch.value().0);
332            BarrierKind::Barrier
333        }
334        PbBarrierKind::Checkpoint => {
335            pending_non_checkpoint_barriers.push(prev_epoch.value().0);
336            BarrierKind::Checkpoint(take(pending_non_checkpoint_barriers))
337        }
338    };
339    BarrierInfo {
340        prev_epoch,
341        curr_epoch,
342        kind,
343    }
344}
345
346// ── Enum unifying independent checkpoint job types ──────────────────────────
347
348/// The type-specific running state of a streaming job that checkpoints independently from the
349/// database's main graph.
350pub(crate) enum IndependentCheckpointJob {
351    CreatingStreamingJob(CreatingStreamingJobControl),
352    BatchRefresh(BatchRefreshJobCheckpointControl),
353    IcebergV3(IcebergV3JobCheckpointControl),
354}
355
356#[derive(Debug, PartialEq, Eq)]
357pub(crate) enum IndependentCheckpointJobStatus {
358    /// The initial barrier cannot be completed until the database has committed the snapshot
359    /// epoch.
360    Initial { snapshot_epoch: u64 },
361    /// The upstream database has committed the snapshot epoch, so barriers may complete.
362    Ready,
363}
364
365/// The lifecycle shared by all independent checkpoint jobs.
366pub(crate) enum IndependentCheckpointJobControl {
367    Running {
368        status: IndependentCheckpointJobStatus,
369        job_id: JobId,
370        partial_graph_id: PartialGraphId,
371        job: IndependentCheckpointJob,
372    },
373    Resetting {
374        /// Keep changelog pins until compute acknowledges the partial-graph reset. Before that
375        /// acknowledgment, snapshot-backfill executors may still be stopping and reading logs.
376        pinned_upstream_tables: HashSet<TableId>,
377        subscriptions_to_drop: Vec<PbSubscriptionUpstreamInfo>,
378        notifiers: Vec<CollectionNotifier>,
379    },
380}
381
382impl IndependentCheckpointJob {
383    fn can_drop_independently(&self) -> bool {
384        match self {
385            Self::CreatingStreamingJob(j) => j.can_drop_independently(),
386            Self::BatchRefresh(_) => true,
387            Self::IcebergV3(_) => true,
388        }
389    }
390
391    fn can_drop(&self) -> bool {
392        match self {
393            Self::CreatingStreamingJob(j) => j.can_drop(),
394            Self::BatchRefresh(_) => true,
395            Self::IcebergV3(_) => true,
396        }
397    }
398
399    fn pinned_upstream_tables(&self) -> &HashSet<TableId> {
400        match self {
401            Self::CreatingStreamingJob(j) => j.pinned_upstream_tables(),
402            Self::BatchRefresh(j) => j.pinned_upstream_tables(),
403            Self::IcebergV3(j) => j.pinned_upstream_tables(),
404        }
405    }
406
407    pub(crate) fn pre_apply_throttle(
408        &mut self,
409        config: &mut ThrottleConfigMap,
410    ) -> Option<Mutation> {
411        match self {
412            Self::CreatingStreamingJob(job) => job.pre_apply_throttle(config),
413            Self::BatchRefresh(job) => job.pre_apply_throttle(config),
414            Self::IcebergV3(job) => job.pre_apply_throttle(config),
415        }
416    }
417
418    pub(crate) fn on_new_upstream_barrier(
419        &mut self,
420        partial_graph_manager: &mut PartialGraphManager,
421        barrier_info: &BarrierInfo,
422        mutation: Option<(Mutation, Option<&mut NotifierStarter>)>,
423    ) -> MetaResult<()> {
424        match self {
425            Self::CreatingStreamingJob(job) => {
426                job.on_new_upstream_barrier(partial_graph_manager, barrier_info, mutation)
427            }
428            Self::BatchRefresh(job) => {
429                job.on_new_upstream_barrier(partial_graph_manager, barrier_info, mutation)
430            }
431            Self::IcebergV3(job) => {
432                job.on_new_upstream_barrier(partial_graph_manager, barrier_info, mutation)
433            }
434        }
435    }
436}
437
438impl IndependentCheckpointJobStatus {
439    fn on_upstream_database_ack_completed(&mut self, committed_epoch: u64) {
440        if let Self::Initial { snapshot_epoch } = self
441            && committed_epoch >= *snapshot_epoch
442        {
443            *self = Self::Ready;
444        }
445    }
446}
447
448impl IndependentCheckpointJobControl {
449    pub(crate) fn resetting(pinned_upstream_tables: HashSet<TableId>) -> Self {
450        Self::Resetting {
451            pinned_upstream_tables,
452            subscriptions_to_drop: vec![],
453            notifiers: vec![],
454        }
455    }
456
457    pub(crate) fn creating_streaming_job(
458        job_id: JobId,
459        partial_graph_id: PartialGraphId,
460        status: IndependentCheckpointJobStatus,
461        job: CreatingStreamingJobControl,
462    ) -> Self {
463        Self::Running {
464            status,
465            job_id,
466            partial_graph_id,
467            job: IndependentCheckpointJob::CreatingStreamingJob(job),
468        }
469    }
470
471    pub(crate) fn batch_refresh(
472        job_id: JobId,
473        partial_graph_id: PartialGraphId,
474        status: IndependentCheckpointJobStatus,
475        job: BatchRefreshJobCheckpointControl,
476    ) -> Self {
477        Self::Running {
478            status,
479            job_id,
480            partial_graph_id,
481            job: IndependentCheckpointJob::BatchRefresh(job),
482        }
483    }
484
485    pub(crate) fn iceberg_v3(
486        job_id: JobId,
487        partial_graph_id: PartialGraphId,
488        status: IndependentCheckpointJobStatus,
489        job: IcebergV3JobCheckpointControl,
490    ) -> Self {
491        Self::Running {
492            status,
493            job_id,
494            partial_graph_id,
495            job: IndependentCheckpointJob::IcebergV3(job),
496        }
497    }
498
499    pub(crate) fn running(&self) -> Option<&IndependentCheckpointJob> {
500        match self {
501            Self::Running { job, .. } => Some(job),
502            Self::Resetting { .. } => None,
503        }
504    }
505
506    pub(crate) fn running_mut(&mut self) -> Option<&mut IndependentCheckpointJob> {
507        match self {
508            Self::Running { job, .. } => Some(job),
509            Self::Resetting { .. } => None,
510        }
511    }
512
513    pub(crate) fn ready_mut(&mut self) -> Option<&mut IndependentCheckpointJob> {
514        match self {
515            Self::Running {
516                status: IndependentCheckpointJobStatus::Ready,
517                job,
518                ..
519            } => Some(job),
520            Self::Running {
521                status: IndependentCheckpointJobStatus::Initial { .. },
522                ..
523            }
524            | Self::Resetting { .. } => None,
525        }
526    }
527
528    pub(crate) fn on_upstream_database_ack_completed(&mut self, committed_epoch: u64) {
529        if let Self::Running { status, .. } = self {
530            status.on_upstream_database_ack_completed(committed_epoch);
531        }
532    }
533
534    pub(crate) fn gen_backfill_progress(&self) -> Option<BackfillProgress> {
535        match self.running()? {
536            IndependentCheckpointJob::CreatingStreamingJob(j) => Some(j.gen_backfill_progress()),
537            IndependentCheckpointJob::BatchRefresh(j) => j.gen_backfill_progress(),
538            IndependentCheckpointJob::IcebergV3(j) => j.gen_backfill_progress(),
539        }
540    }
541
542    /// Collect a barrier and return whether a checkpoint should be forced in the next barrier.
543    pub(crate) fn collect(&mut self, collected_barrier: CollectedBarrier<'_>) -> bool {
544        let job = self
545            .running_mut()
546            .expect("barriers should only be collected from a running partial graph");
547        match job {
548            IndependentCheckpointJob::CreatingStreamingJob(j) => j.collect(collected_barrier),
549            IndependentCheckpointJob::BatchRefresh(j) => j.collect(collected_barrier),
550            IndependentCheckpointJob::IcebergV3(j) => j.collect(collected_barrier),
551        }
552    }
553
554    pub(crate) fn gen_fragment_backfill_progress(&self) -> Vec<FragmentBackfillProgress> {
555        match self.running() {
556            Some(IndependentCheckpointJob::CreatingStreamingJob(j)) => {
557                j.gen_fragment_backfill_progress()
558            }
559            Some(IndependentCheckpointJob::BatchRefresh(j)) => j.gen_fragment_backfill_progress(),
560            Some(IndependentCheckpointJob::IcebergV3(j)) => j.gen_fragment_backfill_progress(),
561            None => vec![],
562        }
563    }
564
565    pub(crate) fn pinned_upstream_tables(&self) -> &HashSet<TableId> {
566        match self {
567            Self::Running { job, .. } => job.pinned_upstream_tables(),
568            Self::Resetting {
569                pinned_upstream_tables,
570                ..
571            } => pinned_upstream_tables,
572        }
573    }
574
575    pub(crate) fn fragment_infos(&self) -> Option<&HashMap<FragmentId, InflightFragmentInfo>> {
576        match self.running()? {
577            IndependentCheckpointJob::CreatingStreamingJob(j) => j.fragment_infos(),
578            IndependentCheckpointJob::BatchRefresh(j) => j.fragment_infos(),
579            IndependentCheckpointJob::IcebergV3(j) => Some(j.fragment_infos()),
580        }
581    }
582
583    pub(crate) fn ack_completed(
584        &mut self,
585        partial_graph_manager: &mut PartialGraphManager,
586        epoch: u64,
587    ) -> MetaResult<()> {
588        match self {
589            Self::Running {
590                status: IndependentCheckpointJobStatus::Ready,
591                job: IndependentCheckpointJob::CreatingStreamingJob(j),
592                ..
593            } => {
594                j.ack_completed(partial_graph_manager, epoch);
595            }
596            Self::Running {
597                status: IndependentCheckpointJobStatus::Ready,
598                job: IndependentCheckpointJob::BatchRefresh(j),
599                ..
600            } => {
601                j.ack_completed(partial_graph_manager, epoch);
602            }
603            Self::Running {
604                status: IndependentCheckpointJobStatus::Ready,
605                job: IndependentCheckpointJob::IcebergV3(j),
606                ..
607            } => j.ack_completed(partial_graph_manager, epoch)?,
608            Self::Running {
609                status: IndependentCheckpointJobStatus::Initial { .. },
610                ..
611            } => {
612                panic!("an initial job should transition to ready before completing a barrier")
613            }
614            Self::Resetting { .. } => {
615                // The job was dropped while the completing task was running in the background.
616                // The partial graph has already been reset, so skip the ack.
617            }
618        }
619        Ok(())
620    }
621
622    pub(crate) fn on_partial_graph_reset(self) -> Vec<PbSubscriptionUpstreamInfo> {
623        match self {
624            Self::Resetting {
625                subscriptions_to_drop,
626                notifiers,
627                ..
628            } => {
629                for notifier in notifiers {
630                    notifier.notify_collected();
631                }
632                subscriptions_to_drop
633            }
634            Self::Running { .. } => {
635                panic!("should be resetting when receiving reset partial graph resp")
636            }
637        }
638    }
639
640    pub(crate) fn can_drop(&self) -> bool {
641        match self {
642            Self::Running { job, .. } => job.can_drop(),
643            Self::Resetting { .. } => true,
644        }
645    }
646
647    pub(crate) fn drop(
648        &mut self,
649        notifier: Option<&mut NotifierStarter>,
650        partial_graph_manager: &mut PartialGraphManager,
651    ) -> bool {
652        match self {
653            Self::Resetting { notifiers, .. } => {
654                notifiers.extend(notifier.map(NotifierStarter::add_notify));
655                true
656            }
657            Self::Running { job, .. } if !job.can_drop_independently() => false,
658            Self::Running {
659                job_id,
660                partial_graph_id,
661                job,
662                ..
663            } => {
664                let subscriptions_to_drop = job
665                    .pinned_upstream_tables()
666                    .iter()
667                    .map(|upstream_mv_table_id| PbSubscriptionUpstreamInfo {
668                        subscriber_id: job_id.as_subscriber_id(),
669                        upstream_mv_table_id: *upstream_mv_table_id,
670                    })
671                    .collect();
672                // Resetting must keep owning the pins after the concrete job is replaced.
673                let pinned_upstream_tables = job.pinned_upstream_tables().clone();
674                partial_graph_manager.reset_partial_graphs([*partial_graph_id]);
675                *self = Self::Resetting {
676                    pinned_upstream_tables,
677                    subscriptions_to_drop,
678                    notifiers: notifier
679                        .map(NotifierStarter::add_notify)
680                        .into_iter()
681                        .collect(),
682                };
683                true
684            }
685        }
686    }
687
688    /// Reset during database recovery.
689    ///
690    /// Returns `true` if the partial graph was already resetting (from a prior drop),
691    /// meaning caller should not issue a new reset request.
692    pub(crate) fn reset(self) -> bool {
693        match self {
694            // Running jobs have no reset state to drain. Recovery will issue the partial-graph
695            // reset after rebuilding its reset plan.
696            Self::Running { .. } => false,
697            Self::Resetting { notifiers, .. } => {
698                for notifier in notifiers {
699                    notifier.notify_collected();
700                }
701                true
702            }
703        }
704    }
705}
706
707#[cfg(test)]
708mod tests {
709    use super::*;
710
711    #[test]
712    fn test_resetting_has_shared_inactive_behavior_and_keeps_pin() {
713        let pinned_upstream_tables = HashSet::from([TableId::new(7)]);
714        let job = IndependentCheckpointJobControl::Resetting {
715            pinned_upstream_tables: pinned_upstream_tables.clone(),
716            subscriptions_to_drop: vec![],
717            notifiers: vec![],
718        };
719
720        assert!(job.running().is_none());
721        assert!(job.gen_backfill_progress().is_none());
722        assert!(job.gen_fragment_backfill_progress().is_empty());
723        assert_eq!(job.pinned_upstream_tables(), &pinned_upstream_tables);
724        assert!(job.fragment_infos().is_none());
725        assert!(job.reset());
726    }
727
728    #[test]
729    fn test_reset_completion_returns_subscriptions_to_drop() {
730        let subscriptions_to_drop = vec![PbSubscriptionUpstreamInfo {
731            subscriber_id: JobId::new(8).as_subscriber_id(),
732            upstream_mv_table_id: TableId::new(7),
733        }];
734        let job = IndependentCheckpointJobControl::Resetting {
735            pinned_upstream_tables: HashSet::from([TableId::new(7)]),
736            subscriptions_to_drop: subscriptions_to_drop.clone(),
737            notifiers: vec![],
738        };
739
740        assert_eq!(job.on_partial_graph_reset(), subscriptions_to_drop);
741    }
742
743    #[test]
744    fn test_initial_status_transitions_on_upstream_database_ack() {
745        let mut status = IndependentCheckpointJobStatus::Initial { snapshot_epoch: 10 };
746
747        status.on_upstream_database_ack_completed(9);
748        assert_eq!(
749            status,
750            IndependentCheckpointJobStatus::Initial { snapshot_epoch: 10 }
751        );
752        status.on_upstream_database_ack_completed(10);
753        assert_eq!(status, IndependentCheckpointJobStatus::Ready);
754        status.on_upstream_database_ack_completed(11);
755        assert_eq!(status, IndependentCheckpointJobStatus::Ready);
756    }
757}
risingwave_meta/barrier/checkpoint/independent_job/mod.rs

risingwave_meta/barrier/checkpoint/independent_job/
mod.rs
