Struct BatchRefreshJobCheckpointControl 

pub(crate) struct BatchRefreshJobCheckpointControl {
    job_id: JobId,
    partial_graph_id: PartialGraphId,
    snapshot_backfill_upstream_tables: HashSet<TableId>,
    snapshot_epoch: u64,
    max_committed_epoch: Option<u64>,
    status: BatchRefreshJobStatus,
}

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.

Fields

job_id: JobId

partial_graph_id: PartialGraphId

snapshot_backfill_upstream_tables: HashSet<TableId>

snapshot_epoch: u64

max_committed_epoch: Option<u64>

status: BatchRefreshJobStatus

Implementations

impl BatchRefreshJobCheckpointControl

pub(crate) fn render_actors_and_build_job_info( fragments: &HashMap<FragmentId, LoadedFragment>, downstreams: &FragmentDownstreamRelation, definition: &str, actor_id_generator: &AtomicU32, worker_nodes: &HashMap<WorkerId, WorkerNode>, database_resource_group: &str, streaming_job_model: &Model, partial_graph_id: PartialGraphId, ) -> MetaResult<BatchRefreshRenderResult>

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, state_table_ids, actors_to_create

Shared by both the DDL create path and the recovery path.

pub(crate) fn build_initial_partial_graph_mutation( render_result: &BatchRefreshRenderResult, backfill_ordering: &ExtendedFragmentBackfillOrder, ) -> Mutation

Build the initial Add mutation for the partial graph’s first barrier.

The rendered actors come from a prior render_actors_and_build_job_info() call; backfill_nodes_to_pause is derived from the job’s backfill ordering.

fn resolve_ensembles( fragments: &HashMap<FragmentId, LoadedFragment>, downstreams: &FragmentDownstreamRelation, ) -> MetaResult<Vec<NoShuffleEnsemble>>

Derive no-shuffle ensembles from fragment downstreams.

impl BatchRefreshJobCheckpointControl

pub(crate) fn new( database_id: DatabaseId, job_id: JobId, create_info: CreateSnapshotBackfillJobCommandInfo, notifiers: Vec<Notifier>, snapshot_backfill_upstream_tables: HashSet<TableId>, snapshot_epoch: u64, version_stat: &HummockVersionStats, partial_graph_manager: &mut PartialGraphManager, logical: &BatchRefreshLogicalFragments, worker_nodes: &HashMap<WorkerId, WorkerNode>, ) -> MetaResult<Self>

Create from DDL command. Starts in ConsumingSnapshot.

Internally calls render_actors_and_build_job_info() and injects the partial-graph initial barrier.

pub(crate) fn recover( database_id: DatabaseId, job_id: JobId, snapshot_backfill_upstream_tables: HashSet<TableId>, snapshot_epoch: u64, committed_epoch: u64, backfill_order: ExtendedFragmentBackfillOrder, version_stat: &HummockVersionStats, initial_mutation: Mutation, render_result: BatchRefreshRenderResult, partial_graph_recoverer: &mut PartialGraphRecoverer<'_>, ) -> MetaResult<Self>

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.

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>, notifiers: Vec<Notifier>, first_create_info: Option<CreateSnapshotBackfillJobCommandInfo>, is_stop: bool, ) -> MetaResult<()>

impl BatchRefreshJobCheckpointControl

pub(crate) fn on_new_upstream_barrier( &mut self, partial_graph_manager: &mut PartialGraphManager, barrier_info: &BarrierInfo, mutation: Option<(Mutation, Vec<Notifier>)>, ) -> MetaResult<()>

pub(crate) fn collect( &mut self, collected_barrier: CollectedBarrier<'_>, ) -> bool

impl BatchRefreshJobCheckpointControl

pub(crate) fn start_completing( &mut self, partial_graph_manager: &mut PartialGraphManager, ) -> Option<(u64, HashMap<WorkerId, BarrierCompleteResponse>, PartialGraphBarrierInfo, Option<TrackingJob>)>

pub(super) fn ack_completed( &mut self, partial_graph_manager: &mut PartialGraphManager, completed_epoch: u64, )

pub(super) fn on_partial_graph_reset(self)

Called when the partial graph reset is confirmed (drop only).

impl BatchRefreshJobCheckpointControl

pub(crate) fn gen_backfill_progress(&self) -> Option<BackfillProgress>

pub(super) fn gen_fragment_backfill_progress( &self, ) -> Vec<FragmentBackfillProgress>

pub(super) fn pinned_upstream_log_epoch(&self) -> (u64, HashSet<TableId>)

Returns the pinned upstream log epoch and upstream table IDs.

pub(crate) fn fragment_infos( &self, ) -> Option<&HashMap<FragmentId, InflightFragmentInfo>>

pub(crate) fn is_snapshot_backfilling(&self) -> bool

impl BatchRefreshJobCheckpointControl

pub(super) fn drop( &mut self, notifiers: &mut Vec<Notifier>, partial_graph_manager: &mut PartialGraphManager, ) -> bool

Drop this batch refresh job.

pub(crate) fn reset(self) -> bool

Reset during database recovery.

Returns true if the partial graph was already resetting (from a prior drop), meaning we should not issue a new reset request.

Trait Implementations

impl Debug for BatchRefreshJobCheckpointControl

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.