risingwave_meta/barrier/checkpoint/independent_job/batch_refresh_job/

mod.rs

// Copyright 2026 RisingWave Labs
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Batch refresh job checkpoint control for periodically-refreshed materialized views.
//!
//! It lives permanently in `DatabaseCheckpointControl.independent_checkpoint_job_controls`
//! as a running batch-refresh independent job for its entire lifetime.
//!
//! Lifecycle:
//!   DDL → `ConsumingSnapshot` → `FinishingSnapshot` → `Idle`
//!                                                        ↕  (periodic trigger)
//!                                        `Idle` ← `ConsumingLogStore`

use std::collections::{HashMap, HashSet, hash_map};
use std::mem::{replace, take};
use std::sync::atomic::AtomicU32;

use anyhow::anyhow;
use itertools::Itertools;
use risingwave_common::catalog::{DatabaseId, TableId};
use risingwave_common::id::JobId;
use risingwave_common::metrics::{LabelGuardedHistogram, LabelGuardedIntGauge};
use risingwave_common::util::epoch::{Epoch, EpochPair};
use risingwave_meta_model::{DispatcherType, WorkerId, streaming_job};
use risingwave_pb::common::WorkerNode;
use risingwave_pb::ddl_service::PbBackfillType;
use risingwave_pb::hummock::HummockVersionStats;
use risingwave_pb::id::{ActorId, FragmentId, PartialGraphId};
use risingwave_pb::stream_plan::barrier_mutation::Mutation;
use risingwave_pb::stream_plan::{AddMutation, StopMutation};
use risingwave_pb::stream_service::BarrierCompleteResponse;
use tracing::{debug, info};

use crate::MetaResult;
use crate::barrier::checkpoint::independent_job::{
    IndependentCheckpointJob, IndependentCheckpointJobControl, IndependentCheckpointJobStatus,
    IndependentJobControl, IndependentJobInfo, InitialPartialGraphRequest, SnapshotPhaseControl,
    add_initial_partial_graph, build_initial_add_mutation,
};
use crate::barrier::command::{PostCollectCommand, ThrottleConfigMap, extract_throttle_config};
use crate::barrier::context::CreateIndependentStreamingJobCommandInfo;
use crate::barrier::edge_builder::FragmentEdgeBuilder;
use crate::barrier::info::BarrierInfo;
use crate::barrier::partial_graph::{
    CollectedBarrier, PartialGraphBarrierInfo, PartialGraphManager, PartialGraphStat,
};
use crate::barrier::progress::{TrackingJob, collect_done_fragments};
use crate::barrier::rpc::{ControlStreamManager, to_partial_graph_id};
use crate::barrier::{BackfillProgress, BarrierKind, FragmentBackfillProgress, TracedEpoch};
use crate::controller::fragment::InflightFragmentInfo;
use crate::controller::scale::{
    ComponentFragmentAligner, EnsembleActorTemplate, LoadedFragment, NoShuffleEnsemble,
    build_no_shuffle_fragment_graph_edges, find_no_shuffle_graphs,
};
use crate::model::{
    FragmentDownstreamRelation, StreamActor, StreamJobActorsToCreate, StreamingJobModelContextExt,
};
use crate::notification::NotifierStarter;
use crate::rpc::metrics::GLOBAL_META_METRICS;
use crate::stream::ExtendedFragmentBackfillOrder;

// ── Public types ──────────────────────────────────────────────────────────────

/// Logical fragment metadata for a batch refresh job.
///
/// Contains only catalog-level information: fragment structure, stream plan nodes,
/// distribution, and downstream relations. No actor IDs, no worker placement.
///
/// Used as the uniform input for `render_actors_and_build_job_info()`, which performs
/// actor rendering (ID allocation, worker placement, vnode assignment) internally.
/// No-shuffle ensembles are derived from `downstreams` internally.
#[derive(Debug)]
pub(crate) struct BatchRefreshLogicalFragments {
    /// Logical fragments of this job. Keyed by `fragment_id`.
    pub fragments: HashMap<FragmentId, LoadedFragment>,
    /// Internal downstream relations (intra-job only; no upstream edges).
    pub downstreams: FragmentDownstreamRelation,
}

/// Result of the unified actor rendering for a batch refresh job.
///
/// Produced by `render_actors_and_build_job_info()` and consumed by both
/// `new()` (create) and `recover()`.
#[derive(Debug)]
pub(crate) struct BatchRefreshRenderResult {
    pub fragment_infos: HashMap<FragmentId, InflightFragmentInfo>,
    pub node_actors: HashMap<WorkerId, HashSet<ActorId>>,
    pub actors_to_create: StreamJobActorsToCreate,
}

/// Static actor metadata rendered for a closed independent-job graph.
///
/// This does not build edges or decide which fragments are active. In particular,
/// Iceberg V3 can render all of its actor partitions once and activate only the
/// input or compaction-resolver partition for a given phase.
#[derive(Debug)]
pub(crate) struct RenderedIndependentJobActors {
    pub fragment_infos: HashMap<FragmentId, InflightFragmentInfo>,
    pub stream_actors: HashMap<FragmentId, Vec<StreamActor>>,
}

// ── Batch refresh job metadata ───────────────────────────────────────────────

/// Lightweight metadata for re-rendering actors on each periodic refresh run.
///
/// Loaded asynchronously on every trigger via `load_batch_refresh_trigger_context()`.
/// Contains the pieces consumed by `from_context()` and `render_actors_and_build_job_info()`,
/// as well as the resolved upstream log epochs and target epoch for this trigger.
#[derive(Debug)]
pub(crate) struct BatchRefreshJobTriggerContext {
    pub fragments: HashMap<FragmentId, LoadedFragment>,
    pub downstreams: FragmentDownstreamRelation,
    pub streaming_job_model: streaming_job::Model,
    pub definition: String,
    pub database_resource_group: String,
    /// Changelog entries per upstream table, used to derive log barriers.
    pub upstream_table_log_epochs: HashMap<TableId, Vec<(Vec<u64>, u64)>>,
    /// The upstream committed epoch to catch up to.
    pub target_upstream_epoch: u64,
}

// ── Status ────────────────────────────────────────────────────────────────────

#[derive(Debug)]
enum BatchRefreshJobStatus {
    /// The job is consuming upstream snapshot.
    ///
    /// Once snapshot consumption finishes, the final checkpoint + stop barriers are injected
    /// and the status transitions to `FinishingSnapshot`.
    ConsumingSnapshot {
        snapshot: SnapshotPhaseControl,
        fragment_infos: HashMap<FragmentId, InflightFragmentInfo>,
        node_actors: HashMap<WorkerId, HashSet<ActorId>>,
    },
    /// The job has finished consuming the snapshot.
    ///
    /// The final checkpoint barrier (at `snapshot_epoch`) and the stop barrier have been
    /// injected. Once the stop epoch is committed the job transitions to `Idle`.
    /// The committed epoch is expected to be the snapshot epoch when the snapshot
    /// consumption finishes.
    FinishingSnapshot {
        tracking_job: Option<TrackingJob>,
        fragment_infos: HashMap<FragmentId, InflightFragmentInfo>,
    },
    /// The job is idle, waiting for the next trigger. No partial graph is held.
    Idle,
    /// The job has created a partial graph for periodic refresh and is waiting for
    /// the initial barrier to bootstrap the newly-created actors.
    InitializingBatchRefresh {
        fragment_infos: HashMap<FragmentId, InflightFragmentInfo>,
        node_actors: HashMap<WorkerId, HashSet<ActorId>>,
        /// Log barriers to inject after the partial graph is initialized. The
        /// last one is the checkpoint stop barrier with `curr_epoch = u64::MAX`.
        pending_log_barriers: Vec<BarrierInfo>,
        target_upstream_epoch: u64,
    },
    /// The job is consuming upstream log store changes (periodic refresh).
    ///
    /// All replay barriers have been pre-injected (last with `StopMutation` at
    /// `curr_epoch = u64::MAX`). When `target_upstream_epoch` commits,
    /// the partial graph is removed and the job transitions to `Idle`.
    ConsumingLogStore {
        fragment_infos: HashMap<FragmentId, InflightFragmentInfo>,
        /// `prev_epoch` of the stop barrier; becomes `last_committed_epoch` when transitioning to Idle.
        target_upstream_epoch: u64,
    },
}

// ── Complete type ─────────────────────────────────────────────────────────────

// ── Main checkpoint control ───────────────────────────────────────────────────

/// Self-contained checkpoint control for a batch refresh MV.
///
/// Unlike `CreatingStreamingJobControl`, this struct handles the full lifecycle
/// (snapshot → idle → re-run → idle → ...). Both types are stored together in
/// `DatabaseCheckpointControl.independent_checkpoint_job_controls` as
/// `IndependentCheckpointJobControl` variants.
#[derive(Debug)]
pub(crate) struct BatchRefreshJobCheckpointControl {
    control: IndependentJobControl,
    /// Batch refresh interval in seconds. Used to determine when to trigger a refresh run.
    batch_refresh_seconds: u64,

    status: BatchRefreshJobStatus,
}

// ── Unified actor rendering ───────────────────────────────────────────────────

/// Render actor IDs, placement, vnode bitmaps, and actor definitions for a closed
/// independent-job graph.
///
/// Every fragment in a no-shuffle ensemble must belong to `fragments`. Independent
/// jobs consume upstream data through the log store, so their actor placement does
/// not need to align with an existing upstream fragment.
impl RenderedIndependentJobActors {
    pub(crate) fn render(
        fragments: &HashMap<FragmentId, LoadedFragment>,
        downstreams: &FragmentDownstreamRelation,
        definition: &str,
        actor_id_generator: &AtomicU32,
        worker_nodes: &HashMap<WorkerId, WorkerNode>,
        database_resource_group: &str,
        streaming_job_model: &streaming_job::Model,
    ) -> MetaResult<Self> {
        let ensembles =
            BatchRefreshJobCheckpointControl::resolve_ensembles(fragments, downstreams)?;

        let mut actor_assignments: HashMap<
            FragmentId,
            HashMap<ActorId, (WorkerId, Option<risingwave_common::bitmap::Bitmap>)>,
        > = HashMap::new();

        for ensemble in &ensembles {
            let first_component = ensemble
                .component_fragments()
                .next()
                .expect("ensemble must have at least one component");
            let fragment = &fragments[&first_component];
            let distribution_type = fragment.distribution_type;
            let vnode_count = fragment.vnode_count;

            for fid in ensemble.component_fragments() {
                let f = &fragments[&fid];
                assert_eq!(
                    vnode_count, f.vnode_count,
                    "fragments {} and {} in same ensemble have different vnode counts",
                    first_component, fid,
                );
            }

            let entry_fragment_parallelism = Itertools::exactly_one(
                ensemble
                    .entry_fragments()
                    .map(|fid| fragments[&fid].parallelism.clone())
                    .dedup(),
            )
            .map_err(|_| {
                anyhow!("entry fragments have inconsistent parallelism settings in independent job")
            })?;

            let actor_template = EnsembleActorTemplate::render_new(
                streaming_job_model,
                worker_nodes,
                entry_fragment_parallelism,
                database_resource_group.to_owned(),
                distribution_type,
                vnode_count,
            )?;

            for fid in ensemble.component_fragments() {
                let f = &fragments[&fid];
                let aligner =
                    ComponentFragmentAligner::new_persistent(&actor_template, actor_id_generator);
                let assignments = aligner.align_component_actor(f.distribution_type);
                actor_assignments.insert(fid, assignments);
            }
        }

        let stream_context = streaming_job_model.stream_context();
        let mut stream_actors: HashMap<FragmentId, Vec<StreamActor>> = HashMap::new();
        let mut actor_location: HashMap<ActorId, WorkerId> = HashMap::new();

        for (fragment_id, assignments) in &actor_assignments {
            let mut actors = Vec::with_capacity(assignments.len());
            for (&actor_id, (worker_id, vnode_bitmap)) in assignments {
                actor_location.insert(actor_id, *worker_id);
                actors.push(StreamActor {
                    actor_id,
                    fragment_id: *fragment_id,
                    vnode_bitmap: vnode_bitmap.clone(),
                    mview_definition: definition.to_owned(),
                    expr_context: Some(stream_context.to_expr_context()),
                    config_override: stream_context.config_override.clone(),
                });
            }
            stream_actors.insert(*fragment_id, actors);
        }

        let fragment_infos = fragments
            .iter()
            .map(|(fragment_id, loaded)| {
                let actors = stream_actors
                    .get(fragment_id)
                    .into_iter()
                    .flatten()
                    .map(|actor| {
                        (
                            actor.actor_id,
                            crate::controller::fragment::InflightActorInfo {
                                worker_id: actor_location[&actor.actor_id],
                                vnode_bitmap: actor.vnode_bitmap.clone(),
                                splits: vec![],
                            },
                        )
                    })
                    .collect();
                (
                    *fragment_id,
                    InflightFragmentInfo {
                        fragment_id: *fragment_id,
                        distribution_type: loaded.distribution_type,
                        fragment_type_mask: loaded.fragment_type_mask,
                        vnode_count: loaded.vnode_count,
                        nodes: loaded.nodes.clone(),
                        actors,
                        state_table_ids: loaded.state_table_ids.clone(),
                    },
                )
            })
            .collect();

        Ok(Self {
            fragment_infos,
            stream_actors,
        })
    }

    /// Build the runtime information for the active part of an already-rendered independent job.
    ///
    /// Keeping this separate from [`Self::render`] lets jobs with dormant fragments partition the
    /// rendered graph before any runtime edges are connected.
    pub(crate) fn build_job_info(
        self,
        downstreams: &FragmentDownstreamRelation,
        partial_graph_id: PartialGraphId,
        control_stream_manager: &ControlStreamManager,
    ) -> MetaResult<BatchRefreshRenderResult> {
        let Self {
            fragment_infos,
            stream_actors,
        } = self;

        let (mut edges, _) = FragmentEdgeBuilder::new()
            .add_new_fragments(
                fragment_infos.values(),
                partial_graph_id,
                control_stream_manager,
            )
            .finish_fragments()
            .add_relations(downstreams)?
            .build();

        let actors_to_create =
            edges.collect_actors_to_create(fragment_infos.values().map(|fragment| {
                (
                    fragment.fragment_id,
                    &fragment.nodes,
                    fragment.actors.iter().map(|(actor_id, actor)| {
                        let stream_actor = stream_actors[&fragment.fragment_id]
                            .iter()
                            .find(|stream_actor| stream_actor.actor_id == *actor_id)
                            .expect("rendered actor should exist");
                        (stream_actor, actor.worker_id)
                    }),
                    vec![], // Independent jobs have no actor-level subscribers.
                )
            }));

        let node_actors = InflightFragmentInfo::actor_ids_to_collect(fragment_infos.values());
        Ok(BatchRefreshRenderResult {
            fragment_infos,
            node_actors,
            actors_to_create,
        })
    }
}

impl BatchRefreshJobCheckpointControl {
    /// Render actors for a batch refresh job from logical metadata only.
    ///
    /// Performs the full pipeline:
    /// 1. Derive no-shuffle ensembles from `downstreams`
    /// 2. Render actor assignments (ID allocation, worker placement, vnode bitmap)
    /// 3. Build `StreamActor` structs
    /// 4. Build internal-only edges (no upstream dispatcher edges)
    /// 5. Produce `fragment_infos`, `node_actors`, and `actors_to_create`
    ///
    /// Shared by both the DDL create path and the recovery path.
    pub(crate) fn render_actors_and_build_job_info(
        fragments: &HashMap<FragmentId, LoadedFragment>,
        downstreams: &FragmentDownstreamRelation,
        definition: &str,
        // Actor rendering context:
        actor_id_generator: &AtomicU32,
        worker_nodes: &HashMap<WorkerId, WorkerNode>,
        control_stream_manager: &ControlStreamManager,
        database_resource_group: &str,
        streaming_job_model: &streaming_job::Model,
        // Edge building context:
        partial_graph_id: PartialGraphId,
    ) -> MetaResult<BatchRefreshRenderResult> {
        let rendered = RenderedIndependentJobActors::render(
            fragments,
            downstreams,
            definition,
            actor_id_generator,
            worker_nodes,
            database_resource_group,
            streaming_job_model,
        )?;
        rendered.build_job_info(downstreams, partial_graph_id, control_stream_manager)
    }

    /// Derive no-shuffle ensembles from fragment downstreams.
    fn resolve_ensembles(
        fragments: &HashMap<FragmentId, LoadedFragment>,
        downstreams: &FragmentDownstreamRelation,
    ) -> MetaResult<Vec<NoShuffleEnsemble>> {
        let mut new_no_shuffle: HashMap<_, HashSet<_>> = HashMap::new();
        for (upstream_fid, relations) in downstreams {
            for rel in relations {
                if rel.dispatcher_type == DispatcherType::NoShuffle {
                    new_no_shuffle
                        .entry(*upstream_fid)
                        .or_default()
                        .insert(rel.downstream_fragment_id);
                }
            }
        }

        let mut ensembles = if new_no_shuffle.is_empty() {
            Vec::new()
        } else {
            let no_shuffle_edges: Vec<(FragmentId, FragmentId)> = new_no_shuffle
                .iter()
                .flat_map(|(u, ds)| ds.iter().map(move |d| (*u, *d)))
                .collect();
            let all_fragment_ids: Vec<FragmentId> = no_shuffle_edges
                .iter()
                .flat_map(|(u, d)| [*u, *d])
                .collect::<HashSet<_>>()
                .into_iter()
                .collect();
            let (fwd, bwd) = build_no_shuffle_fragment_graph_edges(no_shuffle_edges);
            find_no_shuffle_graphs(&all_fragment_ids, &fwd, &bwd)?
        };

        // Add standalone fragments as single-fragment ensembles.
        let covered: HashSet<FragmentId> = ensembles
            .iter()
            .flat_map(|e| e.component_fragments())
            .collect();
        for fragment_id in fragments.keys() {
            if !covered.contains(fragment_id) {
                ensembles.push(NoShuffleEnsemble::singleton(*fragment_id));
            }
        }

        Ok(ensembles)
    }
}

// ── Construction ──────────────────────────────────────────────────────────────

impl BatchRefreshJobCheckpointControl {
    /// Create from DDL command. Starts in `ConsumingSnapshot`.
    ///
    /// Internally calls `render_actors_and_build_job_info()` and injects the
    /// partial-graph initial barrier.
    #[expect(clippy::too_many_arguments)]
    pub(crate) fn new<'a>(
        entry: hash_map::VacantEntry<'a, JobId, IndependentCheckpointJobControl>,
        database_id: DatabaseId,
        job_id: JobId,
        create_info: CreateIndependentStreamingJobCommandInfo,
        notifier: Option<&mut NotifierStarter>,
        snapshot_backfill_upstream_tables: HashSet<TableId>,
        snapshot_epoch: u64,
        version_stat: &HummockVersionStats,
        term_id: &str,
        partial_graph_manager: &mut PartialGraphManager,
        logical: &BatchRefreshLogicalFragments,
        worker_nodes: &HashMap<WorkerId, WorkerNode>,
        batch_refresh_seconds: u64,
    ) -> MetaResult<&'a mut Self> {
        debug!(
            %job_id,
            "new batch refresh job"
        );

        let partial_graph_id = to_partial_graph_id(database_id, Some(job_id));
        let backfill_ordering = &create_info.info.fragment_backfill_ordering;
        let actor_id_generator = partial_graph_manager
            .control_stream_manager()
            .env
            .actor_id_generator();

        let render_result = Self::render_actors_and_build_job_info(
            &logical.fragments,
            &logical.downstreams,
            &create_info.info.definition,
            actor_id_generator,
            worker_nodes,
            partial_graph_manager.control_stream_manager(),
            &create_info.info.database_resource_group,
            &create_info.info.streaming_job_model,
            partial_graph_id,
        )?;
        let initial_partial_graph_mutation = build_initial_add_mutation(
            &render_result.fragment_infos,
            backfill_ordering,
            Default::default(), // No splits: batch refresh jobs reject source fragments.
        );

        let (snapshot, initial_barrier_info) = SnapshotPhaseControl::for_new_job(
            snapshot_epoch,
            &render_result.fragment_infos,
            &create_info.info,
            version_stat,
        );

        let BatchRefreshRenderResult {
            fragment_infos,
            node_actors,
            actors_to_create,
        } = render_result;
        let control = IndependentJobControl::new(IndependentJobInfo::from_fragment_infos(
            database_id,
            job_id,
            snapshot_epoch,
            snapshot_backfill_upstream_tables,
            fragment_infos.values(),
        ));

        let mut graph_adder = partial_graph_manager.add_partial_graph(
            partial_graph_id,
            term_id,
            BatchRefreshBarrierStats::new(job_id, snapshot_epoch),
        );
        if let Err(e) = add_initial_partial_graph(
            &mut graph_adder,
            partial_graph_id,
            InitialPartialGraphRequest {
                node_actors: &node_actors,
                state_table_ids: &control.info.state_table_ids,
                barrier_info: initial_barrier_info,
                actors_to_create,
                mutation: initial_partial_graph_mutation,
                notifier,
                create_info,
            },
        ) {
            graph_adder.failed();
            entry.insert(IndependentCheckpointJobControl::resetting(
                control.info.snapshot_backfill_upstream_tables,
            ));
            return Err(e);
        }
        graph_adder.added();
        let this = Self {
            control,
            batch_refresh_seconds,
            status: BatchRefreshJobStatus::ConsumingSnapshot {
                snapshot,
                fragment_infos,
                node_actors,
            },
        };
        let control = entry.insert(IndependentCheckpointJobControl::batch_refresh(
            job_id,
            partial_graph_id,
            IndependentCheckpointJobStatus::Initial { snapshot_epoch },
            this,
        ));
        let Some(IndependentCheckpointJob::BatchRefresh(job)) = control.running_mut() else {
            unreachable!()
        };
        Ok(job)
    }

    /// Recover from a persistent state during recovery.
    ///
    /// - If `committed_epoch >= snapshot_epoch` → Idle (snapshot completed before crash).
    /// - If `committed_epoch < snapshot_epoch` → `ConsumingSnapshot` using pre-rendered actors.
    pub(crate) fn recover(
        control: IndependentJobControl,
        backfill_order: ExtendedFragmentBackfillOrder,
        version_stat: &HummockVersionStats,
        initial_mutation: Mutation,
        render_result: BatchRefreshRenderResult,
        term_id: &str,
        partial_graph_recoverer: &mut crate::barrier::partial_graph::PartialGraphRecoverer<'_>,
        batch_refresh_seconds: u64,
    ) -> MetaResult<Self> {
        let job_id = control.info.job_id;
        let partial_graph_id = control.info.partial_graph_id;
        let snapshot_epoch = control.info.snapshot_epoch;
        let committed_epoch = control
            .max_committed_epoch
            .expect("recovered independent job should have committed");

        if committed_epoch >= snapshot_epoch {
            // Snapshot completed; recover to Idle.
            info!(
                %job_id,
                committed_epoch,
                snapshot_epoch,
                "recovered idle batch refresh job (no partial graph)"
            );
            return Ok(Self {
                control,
                batch_refresh_seconds,
                status: BatchRefreshJobStatus::Idle,
            });
        }

        // Snapshot still in-progress; recover to ConsumingSnapshot.
        info!(
            %job_id,
            committed_epoch,
            snapshot_epoch,
            "recovered batch refresh job to consuming snapshot"
        );

        let (snapshot, first_barrier_info) = SnapshotPhaseControl::for_recovery(
            job_id,
            snapshot_epoch,
            committed_epoch,
            &render_result.fragment_infos,
            &backfill_order,
            version_stat,
        );

        partial_graph_recoverer.recover_graph(
            partial_graph_id,
            term_id,
            initial_mutation,
            &first_barrier_info,
            &render_result.node_actors,
            control.info.state_table_ids.iter().copied(),
            render_result.actors_to_create,
            BatchRefreshBarrierStats::new(job_id, snapshot_epoch),
        )?;

        Ok(Self {
            control,
            batch_refresh_seconds,
            status: BatchRefreshJobStatus::ConsumingSnapshot {
                snapshot,
                fragment_infos: render_result.fragment_infos,
                node_actors: render_result.node_actors,
            },
        })
    }
}

// ── Barrier injection ─────────────────────────────────────────────────────────

impl BatchRefreshJobCheckpointControl {
    fn inject_barrier(
        partial_graph_id: PartialGraphId,
        partial_graph_manager: &mut PartialGraphManager,
        node_actors: &HashMap<WorkerId, HashSet<ActorId>>,
        state_table_ids: &HashSet<TableId>,
        barrier_info: BarrierInfo,
        new_actors: Option<StreamJobActorsToCreate>,
        mutation: Option<Mutation>,
        notifier: Option<&mut NotifierStarter>,
        first_create_info: Option<CreateIndependentStreamingJobCommandInfo>,
        is_stop: bool,
    ) -> MetaResult<()> {
        if is_stop {
            assert!(
                matches!(&mutation, Some(Mutation::Stop(_))),
                "stop barrier must carry a Stop mutation"
            );
        }
        partial_graph_manager.inject_barrier(
            partial_graph_id,
            mutation,
            node_actors,
            state_table_ids.iter().copied(),
            if is_stop {
                // Stop barrier: data already synced by the prior checkpoint.
                itertools::Either::Left(std::iter::empty())
            } else {
                itertools::Either::Right(node_actors.keys().copied())
            },
            new_actors,
            PartialGraphBarrierInfo::new(
                first_create_info.map_or_else(
                    PostCollectCommand::barrier,
                    CreateIndependentStreamingJobCommandInfo::into_post_collect,
                ),
                barrier_info,
                notifier,
                state_table_ids.clone(),
            ),
        )?;
        Ok(())
    }
}

// ── Barrier forwarding and collection ─────────────────────────────────────────

impl BatchRefreshJobCheckpointControl {
    pub(crate) fn on_new_upstream_barrier(
        &mut self,
        partial_graph_manager: &mut PartialGraphManager,
        barrier_info: &BarrierInfo,
        mutation: Option<(Mutation, Option<&mut NotifierStarter>)>,
    ) -> MetaResult<()> {
        if !matches!(self.status, BatchRefreshJobStatus::ConsumingSnapshot { .. }) {
            // ConsumingLogStore has all barriers pre-injected; no forwarding needed.
            // Idle has no partial graph.
            return Ok(());
        }
        let (mutation, notifier) = match mutation {
            Some((mutation, notifier)) => (Some(mutation), notifier),
            None => (None, None),
        };
        // Check if snapshot consumption is finished and we need to inject stop barriers.
        let is_finished = matches!(
            &self.status,
            BatchRefreshJobStatus::ConsumingSnapshot { snapshot, .. }
            if snapshot.create_mview_tracker.is_finished()
        );

        if is_finished {
            // Take the status out to destructure and transition to `FinishingSnapshot`.
            // Use a placeholder; will be overwritten below.
            let old_status = replace(&mut self.status, BatchRefreshJobStatus::Idle);
            let BatchRefreshJobStatus::ConsumingSnapshot {
                mut snapshot,
                fragment_infos,
                node_actors,
            } = old_status
            else {
                unreachable!()
            };

            // Inject final checkpoint at snapshot epoch.
            let final_checkpoint = snapshot.finish_snapshot_barrier();
            let tracking_job = snapshot.create_mview_tracker.into_tracking_job();

            // Inject stop barrier with u64::MAX as curr_epoch and empty nodes_to_sync_table.
            let stop_barrier = BarrierInfo {
                prev_epoch: TracedEpoch::new(Epoch(snapshot.snapshot_epoch)),
                curr_epoch: TracedEpoch::new(Epoch(u64::MAX)),
                kind: BarrierKind::Checkpoint(vec![snapshot.snapshot_epoch]),
            };

            let stop_actors: Vec<ActorId> = fragment_infos
                .values()
                .flat_map(|f| f.actors.keys().copied())
                .collect();

            Self::inject_barrier(
                self.control.info.partial_graph_id,
                partial_graph_manager,
                &node_actors,
                &self.control.info.state_table_ids,
                final_checkpoint,
                None,
                None,
                notifier,
                None,
                false,
            )?;
            Self::inject_barrier(
                self.control.info.partial_graph_id,
                partial_graph_manager,
                &node_actors,
                &self.control.info.state_table_ids,
                stop_barrier,
                None,
                Some(Mutation::Stop(StopMutation {
                    actors: stop_actors,
                    dropped_sink_fragments: vec![],
                })),
                None,
                None,
                true,
            )?;

            self.status = BatchRefreshJobStatus::FinishingSnapshot {
                tracking_job: Some(tracking_job),
                fragment_infos,
            };
        } else {
            // Normal barrier — still consuming snapshot.
            let BatchRefreshJobStatus::ConsumingSnapshot {
                snapshot,
                node_actors,
                ..
            } = &mut self.status
            else {
                unreachable!("is_finished was false, status must be ConsumingSnapshot")
            };

            // Forward a fake barrier to the partial graph.
            let mutation = mutation.or_else(|| snapshot.take_start_backfill_mutation());
            let barrier_to_inject = snapshot.next_fake_barrier(&barrier_info.kind);
            Self::inject_barrier(
                self.control.info.partial_graph_id,
                partial_graph_manager,
                node_actors,
                &self.control.info.state_table_ids,
                barrier_to_inject,
                None,
                mutation,
                notifier,
                None,
                false,
            )?;
        }
        Ok(())
    }

    pub(crate) fn collect(&mut self, collected_barrier: CollectedBarrier<'_>) -> bool {
        match &mut self.status {
            BatchRefreshJobStatus::ConsumingSnapshot { snapshot, .. } => snapshot.apply_progress(
                collected_barrier
                    .resps
                    .values()
                    .flat_map(|resp| &resp.create_mview_progress),
            ),
            BatchRefreshJobStatus::InitializingBatchRefresh { .. }
            | BatchRefreshJobStatus::ConsumingLogStore { .. } => {
                // All barriers are pre-injected; no progress tracking needed.
                false
            }
            _ => false,
        }
    }
}

// ── Completing ────────────────────────────────────────────────────────────────

impl BatchRefreshJobCheckpointControl {
    #[expect(clippy::type_complexity)]
    pub(crate) fn start_completing(
        &mut self,
        partial_graph_manager: &mut PartialGraphManager,
    ) -> Option<(
        u64,
        HashMap<WorkerId, BarrierCompleteResponse>,
        PartialGraphBarrierInfo,
        Option<TrackingJob>,
    )> {
        match &self.status {
            BatchRefreshJobStatus::ConsumingSnapshot { .. }
            | BatchRefreshJobStatus::FinishingSnapshot { .. }
            | BatchRefreshJobStatus::ConsumingLogStore { .. } => {}
            BatchRefreshJobStatus::Idle
            | BatchRefreshJobStatus::InitializingBatchRefresh { .. } => {
                return None;
            }
        };

        partial_graph_manager
            .start_completing(
                self.control.info.partial_graph_id,
                std::ops::Bound::Unbounded,
                |_non_checkpoint_epoch, _resps, _| {
                    // Progress already applied in `collect()`.
                },
            )
            .map(|(epoch, resps, info)| {
                // Take tracking job only when the snapshot stop barrier completes
                // (i.e., we are in FinishingSnapshot and the epoch matches snapshot_epoch).
                // Note: ConsumingLogStore's stop barrier also has prev_epoch == target_upstream_epoch,
                // which may coincidentally equal snapshot_epoch if no new upstream commits occurred.
                // We must check the status, not just the epoch, to avoid a false positive.
                let tracking_job = match &mut self.status {
                    BatchRefreshJobStatus::FinishingSnapshot { tracking_job, .. }
                        if epoch == self.control.info.snapshot_epoch =>
                    {
                        Some(
                            tracking_job
                                .take()
                                .expect("tracking job should not have been taken yet"),
                        )
                    }
                    _ => None,
                };
                (epoch, resps, info, tracking_job)
            })
    }

    pub(super) fn ack_completed(
        &mut self,
        partial_graph_manager: &mut PartialGraphManager,
        completed_epoch: u64,
    ) {
        match &self.status {
            BatchRefreshJobStatus::ConsumingSnapshot { .. } => {
                partial_graph_manager
                    .ack_completed(self.control.info.partial_graph_id, completed_epoch);
                self.control.ack_completed(completed_epoch);
            }
            BatchRefreshJobStatus::FinishingSnapshot { tracking_job, .. }
                if completed_epoch == self.control.info.snapshot_epoch =>
            {
                partial_graph_manager
                    .ack_completed(self.control.info.partial_graph_id, completed_epoch);
                self.control.ack_completed(completed_epoch);
                assert!(
                    tracking_job.is_none(),
                    "tracking job should have been taken at start_completing"
                );
                info!(
                    job_id = %self.control.info.job_id,
                    completed_epoch,
                    "batch refresh job: snapshot done, transitioned to idle, removing partial graph"
                );
                partial_graph_manager
                    .remove_partial_graphs(vec![self.control.info.partial_graph_id]);
                self.status = BatchRefreshJobStatus::Idle;
            }
            BatchRefreshJobStatus::FinishingSnapshot { .. } => {
                partial_graph_manager
                    .ack_completed(self.control.info.partial_graph_id, completed_epoch);
                self.control.ack_completed(completed_epoch);
            }
            BatchRefreshJobStatus::ConsumingLogStore {
                target_upstream_epoch,
                ..
            } if completed_epoch == *target_upstream_epoch => {
                let target = *target_upstream_epoch;
                partial_graph_manager
                    .ack_completed(self.control.info.partial_graph_id, completed_epoch);
                self.control.ack_completed(completed_epoch);
                info!(
                    job_id = %self.control.info.job_id,
                    completed_epoch,
                    target_upstream_epoch = target,
                    "batch refresh job: logstore done, transitioned to idle, removing partial graph"
                );
                partial_graph_manager
                    .remove_partial_graphs(vec![self.control.info.partial_graph_id]);
                self.status = BatchRefreshJobStatus::Idle;
            }
            BatchRefreshJobStatus::ConsumingLogStore { .. } => {
                partial_graph_manager
                    .ack_completed(self.control.info.partial_graph_id, completed_epoch);
                self.control.ack_completed(completed_epoch);
            }
            BatchRefreshJobStatus::Idle
            | BatchRefreshJobStatus::InitializingBatchRefresh { .. } => {
                unreachable!("batch refresh job should not be completing in this state")
            }
        }
    }
}

// ── Query methods ─────────────────────────────────────────────────────────────

impl BatchRefreshJobCheckpointControl {
    pub(crate) fn gen_backfill_progress(&self) -> Option<BackfillProgress> {
        match &self.status {
            BatchRefreshJobStatus::ConsumingSnapshot { snapshot, .. } => {
                let progress = if snapshot.create_mview_tracker.is_finished() {
                    "Snapshot finished".to_owned()
                } else {
                    let progress = snapshot.create_mview_tracker.gen_backfill_progress();
                    format!("BatchRefresh Snapshot [{}]", progress)
                };
                Some(BackfillProgress {
                    progress,
                    backfill_type: PbBackfillType::SnapshotBackfill,
                })
            }
            BatchRefreshJobStatus::FinishingSnapshot { .. } => Some(BackfillProgress {
                progress: "BatchRefresh Stopping".to_owned(),
                backfill_type: PbBackfillType::SnapshotBackfill,
            }),
            BatchRefreshJobStatus::InitializingBatchRefresh { .. }
            | BatchRefreshJobStatus::ConsumingLogStore { .. } => Some(BackfillProgress {
                progress: "BatchRefresh LogStore".to_owned(),
                backfill_type: PbBackfillType::SnapshotBackfill,
            }),
            BatchRefreshJobStatus::Idle => None,
        }
    }

    pub(super) fn gen_fragment_backfill_progress(&self) -> Vec<FragmentBackfillProgress> {
        match &self.status {
            BatchRefreshJobStatus::ConsumingSnapshot {
                snapshot,
                fragment_infos,
                ..
            } => snapshot
                .create_mview_tracker
                .collect_fragment_progress(fragment_infos, true),
            BatchRefreshJobStatus::FinishingSnapshot { fragment_infos, .. } => {
                collect_done_fragments(self.control.info.job_id, fragment_infos)
            }
            _ => vec![],
        }
    }

    pub(super) fn pinned_upstream_tables(&self) -> &HashSet<TableId> {
        match &self.status {
            BatchRefreshJobStatus::ConsumingSnapshot { .. }
            | BatchRefreshJobStatus::FinishingSnapshot { .. }
            | BatchRefreshJobStatus::ConsumingLogStore { .. }
            | BatchRefreshJobStatus::InitializingBatchRefresh { .. }
            | BatchRefreshJobStatus::Idle => &self.control.info.snapshot_backfill_upstream_tables,
        }
    }

    pub(crate) fn fragment_infos(&self) -> Option<&HashMap<FragmentId, InflightFragmentInfo>> {
        match &self.status {
            BatchRefreshJobStatus::ConsumingSnapshot { fragment_infos, .. } => Some(fragment_infos),
            BatchRefreshJobStatus::InitializingBatchRefresh { fragment_infos, .. } => {
                Some(fragment_infos)
            }
            BatchRefreshJobStatus::ConsumingLogStore { fragment_infos, .. } => Some(fragment_infos),
            BatchRefreshJobStatus::FinishingSnapshot { .. } | BatchRefreshJobStatus::Idle => None,
        }
    }

    pub(crate) fn pre_apply_throttle(
        &mut self,
        config: &mut ThrottleConfigMap,
    ) -> Option<Mutation> {
        let BatchRefreshJobStatus::ConsumingSnapshot {
            fragment_infos,
            snapshot,
            ..
        } = &mut self.status
        else {
            return None;
        };
        if snapshot.create_mview_tracker.is_finished() {
            return None;
        }

        extract_throttle_config(config, |fragment_id, stream_node| {
            if let Some(fragment_info) = fragment_infos.get_mut(&fragment_id) {
                fragment_info.nodes = stream_node.clone();
                true
            } else {
                false
            }
        })
    }

    /// Whether this idle job should start a refresh run.
    ///
    /// Returns `true` if the job is idle and the upstream committed epoch is
    /// far enough ahead of the job's last committed epoch (by `batch_refresh_seconds`).
    pub(crate) fn should_start_refresh(&self, upstream_committed_epoch: u64) -> bool {
        if let BatchRefreshJobStatus::Idle = &self.status {
            let last_committed_epoch = self
                .control
                .max_committed_epoch
                .expect("batch refresh job should have committed when entering Idle");
            let job_physical_ms = Epoch(last_committed_epoch).physical_time();
            let upstream_physical_ms = Epoch(upstream_committed_epoch).physical_time();
            let threshold_ms = self.batch_refresh_seconds * 1000;
            upstream_physical_ms.saturating_sub(job_physical_ms) >= threshold_ms
        } else {
            false
        }
    }

    /// Returns the last committed epoch if the job is idle.
    pub(crate) fn last_committed_epoch(&self) -> Option<u64> {
        if let BatchRefreshJobStatus::Idle = &self.status {
            self.control.max_committed_epoch
        } else {
            None
        }
    }
}

// ── Logstore refresh run ──────────────────────────────────────────────────────

impl BatchRefreshJobCheckpointControl {
    /// Start a logstore consumption run.
    ///
    /// Preconditions: the job must be `Idle`.
    ///
    /// 1. Resolves log epochs from the hummock changelog
    /// 2. Re-renders actors using the cached context
    /// 3. Injects all barriers at once (first with `AddMutation`, last with `StopMutation`)
    /// 4. Transitions to `ConsumingLogStore`
    ///
    /// Returns `true` if a refresh run was started, `false` if there are no
    /// log epochs to consume (early return, stays idle).
    pub(crate) fn start_refresh_run(
        &mut self,
        context: &BatchRefreshJobTriggerContext,
        worker_nodes: &HashMap<WorkerId, WorkerNode>,
        actor_id_counter: &AtomicU32,
        term_id: &str,
        partial_graph_manager: &mut PartialGraphManager,
    ) -> MetaResult<bool> {
        let last_committed_epoch = match &self.status {
            BatchRefreshJobStatus::Idle => self
                .control
                .max_committed_epoch
                .expect("batch refresh job should have committed when entering Idle"),
            _ => panic!(
                "batch refresh job {}: start_refresh_run called in non-Idle state {:?}",
                self.control.info.job_id, self.status
            ),
        };

        // Resolve log epochs into barrier infos.
        let target_upstream_epoch = context.target_upstream_epoch;
        let Some((first_epoch, pending_log_barriers)) = Self::resolve_log_epoch_barriers(
            &self.control.info.snapshot_backfill_upstream_tables,
            &context.upstream_table_log_epochs,
            last_committed_epoch,
        )?
        else {
            info!(
                job_id = %self.control.info.job_id,
                last_committed_epoch,
                target_upstream_epoch,
                "batch refresh job: no log epochs to consume, staying idle"
            );
            return Ok(false);
        };

        let log_target_epoch = pending_log_barriers.last().expect("non-empty").prev_epoch();
        if target_upstream_epoch != log_target_epoch {
            info!(
                job_id = %self.control.info.job_id,
                last_committed_epoch,
                target_upstream_epoch,
                log_target_epoch,
                "batch refresh job: upstream target has no resolved changelog yet, staying idle"
            );
            return Ok(false);
        }

        // Build logical fragments from cached context.
        let logical = BatchRefreshLogicalFragments::from_context(context);

        // Re-render actors.
        let render_result = Self::render_actors_and_build_job_info(
            &logical.fragments,
            &logical.downstreams,
            &context.definition,
            actor_id_counter,
            worker_nodes,
            partial_graph_manager.control_stream_manager(),
            &context.database_resource_group,
            &context.streaming_job_model,
            self.control.info.partial_graph_id,
        )?;

        // Build actors_to_create and initial mutation.
        let added_actors: Vec<ActorId> = render_result
            .fragment_infos
            .values()
            .flat_map(|fragment| fragment.actors.keys().copied())
            .collect();

        let initial_mutation = Mutation::Add(AddMutation {
            actor_dispatchers: Default::default(),
            added_actors,
            actor_splits: Default::default(),
            pause: false,
            subscriptions_to_add: Default::default(),
            backfill_nodes_to_pause: Default::default(),
            actor_cdc_table_snapshot_splits: None,
            new_upstream_sinks: Default::default(),
            dropped_actors: Default::default(),
            sink_log_store_flush: Default::default(),
        });

        let node_actors = &render_result.node_actors;
        let initial_barrier = BarrierInfo {
            prev_epoch: TracedEpoch::new(Epoch(last_committed_epoch)),
            curr_epoch: TracedEpoch::new(Epoch(first_epoch)),
            kind: BarrierKind::Initial,
        };
        let mut partial_graph_recoverer = partial_graph_manager.start_recover();
        let recover_result = partial_graph_recoverer.recover_graph(
            self.control.info.partial_graph_id,
            term_id,
            initial_mutation,
            &initial_barrier,
            node_actors,
            self.control.info.state_table_ids.iter().copied(),
            render_result.actors_to_create,
            BatchRefreshBarrierStats::new(
                self.control.info.job_id,
                self.control.info.snapshot_epoch,
            ),
        );
        match recover_result {
            Ok(()) => {
                let initializing_partial_graphs = partial_graph_recoverer.all_initializing();
                debug_assert_eq!(initializing_partial_graphs.len(), 1);
                debug_assert!(
                    initializing_partial_graphs.contains(&self.control.info.partial_graph_id)
                );
            }
            Err(e) => {
                partial_graph_recoverer.failed();
                return Err(e);
            }
        }

        info!(
            job_id = %self.control.info.job_id,
            last_committed_epoch,
            target_upstream_epoch,
            num_log_barriers = pending_log_barriers.len(),
            "batch refresh job: initialized logstore consumption partial graph"
        );

        self.status = BatchRefreshJobStatus::InitializingBatchRefresh {
            fragment_infos: render_result.fragment_infos,
            node_actors: render_result.node_actors,
            pending_log_barriers,
            target_upstream_epoch,
        };

        Ok(true)
    }

    pub(crate) fn on_log_store_initialized(
        &mut self,
        partial_graph_manager: &mut PartialGraphManager,
    ) -> MetaResult<()> {
        let old_status = replace(&mut self.status, BatchRefreshJobStatus::Idle);
        let BatchRefreshJobStatus::InitializingBatchRefresh {
            fragment_infos,
            node_actors,
            pending_log_barriers,
            target_upstream_epoch,
        } = old_status
        else {
            panic!(
                "batch refresh job {}: logstore initialized in unexpected status {:?}",
                self.control.info.job_id, old_status
            );
        };

        let final_barrier_idx = pending_log_barriers.len() - 1;
        let mut stop_mutation = Some(Mutation::Stop(StopMutation {
            actors: fragment_infos
                .values()
                .flat_map(|fragment| fragment.actors.keys().copied())
                .collect(),
            dropped_sink_fragments: vec![],
        }));
        for (idx, barrier) in pending_log_barriers.into_iter().enumerate() {
            let is_stop_barrier = idx == final_barrier_idx;
            let mutation = is_stop_barrier.then(|| stop_mutation.take().expect("unused"));
            Self::inject_barrier(
                self.control.info.partial_graph_id,
                partial_graph_manager,
                &node_actors,
                &self.control.info.state_table_ids,
                barrier,
                None,
                mutation,
                None,
                None,
                is_stop_barrier,
            )?;
        }

        self.status = BatchRefreshJobStatus::ConsumingLogStore {
            fragment_infos,
            target_upstream_epoch,
        };
        Ok(())
    }

    /// Resolve upstream log epochs from the hummock changelog into barrier infos.
    ///
    /// Returns `(first_epoch, log_barriers)`. `first_epoch` is consumed by the
    /// initial barrier. `log_barriers` contains all barriers to inject after
    /// initialization, ending with the final checkpoint stop barrier.
    fn resolve_log_epoch_barriers(
        snapshot_backfill_upstream_tables: &HashSet<TableId>,
        upstream_table_log_epochs: &HashMap<TableId, Vec<(Vec<u64>, u64)>>,
        exclusive_start_log_epoch: u64,
    ) -> MetaResult<Option<(u64, Vec<BarrierInfo>)>> {
        let table_id = snapshot_backfill_upstream_tables
            .iter()
            .next()
            .expect("snapshot backfill job should have upstream");
        let Some(epochs) = upstream_table_log_epochs.get(table_id) else {
            return Ok(None);
        };

        // Find the starting point: skip entries up to and including exclusive_start_log_epoch.
        let mut epochs_iter = epochs.iter().peekable();
        loop {
            match epochs_iter.peek() {
                Some((_, checkpoint_epoch)) if *checkpoint_epoch <= exclusive_start_log_epoch => {
                    epochs_iter.next();
                }
                _ => break,
            }
        }

        let mut epoch_infos = vec![];
        for (non_checkpoint_epochs, checkpoint_epoch) in epochs_iter {
            epoch_infos.extend(
                non_checkpoint_epochs
                    .iter()
                    .copied()
                    .map(|epoch| (epoch, false)),
            );
            epoch_infos.push((*checkpoint_epoch, true));
        }
        if epoch_infos.is_empty() {
            return Ok(None);
        }

        let first_epoch = epoch_infos[0].0;
        let mut pending_non_checkpoint_epochs = vec![];
        let mut replay_barriers = vec![];
        for window in epoch_infos.windows(2) {
            let (prev_epoch, is_checkpoint) = window[0];
            let curr_epoch = window[1].0;
            assert!(prev_epoch > exclusive_start_log_epoch);
            assert!(curr_epoch > prev_epoch);
            pending_non_checkpoint_epochs.push(prev_epoch);
            let kind = if is_checkpoint {
                BarrierKind::Checkpoint(take(&mut pending_non_checkpoint_epochs))
            } else {
                BarrierKind::Barrier
            };
            replay_barriers.push(BarrierInfo {
                prev_epoch: TracedEpoch::new(Epoch(prev_epoch)),
                curr_epoch: TracedEpoch::new(Epoch(curr_epoch)),
                kind,
            });
        }

        let (last_epoch, _) = *epoch_infos.last().expect("non-empty");
        assert!(last_epoch > exclusive_start_log_epoch);
        pending_non_checkpoint_epochs.push(last_epoch);
        replay_barriers.push(BarrierInfo {
            prev_epoch: TracedEpoch::new(Epoch(last_epoch)),
            curr_epoch: TracedEpoch::new(Epoch(u64::MAX)),
            kind: BarrierKind::Checkpoint(pending_non_checkpoint_epochs),
        });

        Ok(Some((first_epoch, replay_barriers)))
    }
}

impl BatchRefreshLogicalFragments {
    /// Build logical fragments from a trigger context.
    pub(crate) fn from_context(ctx: &BatchRefreshJobTriggerContext) -> Self {
        Self {
            fragments: ctx.fragments.clone(),
            downstreams: ctx.downstreams.clone(),
        }
    }
}

// ── Barrier stats ─────────────────────────────────────────────────────────────

struct BatchRefreshBarrierStats {
    barrier_latency: LabelGuardedHistogram,
    inflight_barrier_num: LabelGuardedIntGauge,
}

impl BatchRefreshBarrierStats {
    fn new(job_id: JobId, _snapshot_epoch: u64) -> Self {
        let table_id_str = format!("{}", job_id);
        Self {
            barrier_latency: GLOBAL_META_METRICS
                .snapshot_backfill_barrier_latency
                .with_guarded_label_values(&[table_id_str.as_str(), "batch_refresh_snapshot"]),
            inflight_barrier_num: GLOBAL_META_METRICS
                .snapshot_backfill_inflight_barrier_num
                .with_guarded_label_values(&[&table_id_str]),
        }
    }
}

impl PartialGraphStat for BatchRefreshBarrierStats {
    fn observe_barrier_latency(&self, _epoch: EpochPair, barrier_latency_secs: f64) {
        self.barrier_latency.observe(barrier_latency_secs);
    }

    fn observe_barrier_num(&self, inflight_barrier_num: usize, _collected_barrier_num: usize) {
        self.inflight_barrier_num.set(inflight_barrier_num as _);
    }
}