Struct DatabaseCheckpointControl 

pub(in barrier) struct DatabaseCheckpointControl {
    pub(super) database_id: DatabaseId,
    pub(super) state: BarrierWorkerState,
    command_ctx_queue: BTreeMap<u64, EpochNode>,
    completing_barrier: Option<u64>,
    committed_epoch: Option<u64>,
    pub(super) database_info: InflightDatabaseInfo,
    pub creating_streaming_job_controls: HashMap<JobId, CreatingStreamingJobControl>,
    metrics: DatabaseCheckpointControlMetrics,
}

Controls the concurrent execution of commands.

Fields

database_id: DatabaseId

state: BarrierWorkerState

command_ctx_queue: BTreeMap<u64, EpochNode>

Save the state and message of barrier in order. Key is the prev_epoch.

completing_barrier: Option<u64>

The barrier that are completing. Some(prev_epoch)

committed_epoch: Option<u64>

database_info: InflightDatabaseInfo

creating_streaming_job_controls: HashMap<JobId, CreatingStreamingJobControl>

metrics: DatabaseCheckpointControlMetrics

Implementations

impl DatabaseCheckpointControl

fn new(database_id: DatabaseId, shared_actor_infos: SharedActorInfos) -> Self

pub(crate) fn recovery( database_id: DatabaseId, state: BarrierWorkerState, committed_epoch: u64, database_info: InflightDatabaseInfo, creating_streaming_job_controls: HashMap<JobId, CreatingStreamingJobControl>, ) -> Self

pub(crate) fn is_valid_after_worker_err(&self, worker_id: WorkerId) -> bool

fn total_command_num(&self) -> usize

fn update_barrier_nums_metrics(&self)

Update the metrics of barrier nums.

fn enqueue_command( &mut self, command_ctx: CommandContext, notifiers: Vec<Notifier>, creating_jobs_to_wait: HashSet<JobId>, )

Enqueue a barrier command

fn barrier_collected( &mut self, partial_graph_id: PartialGraphId, collected_barrier: CollectedBarrier, periodic_barriers: &mut PeriodicBarriers, ) -> MetaResult<()>

Change the state of this prev_epoch to Completed. Return continuous nodes with Completed starting from first node [Completed..InFlight) and remove them.

fn can_inject_barrier(&self, in_flight_barrier_nums: usize) -> bool

Pause inject barrier until True.

impl DatabaseCheckpointControl

fn collect_backfill_pinned_upstream_log_epoch( &self, ) -> HashMap<JobId, (u64, HashSet<TableId>)>

return creating job table fragment id -> (backfill progress epoch , {upstream_mv_table_id})

fn collect_no_shuffle_fragment_relations_for_reschedule_check( &self, ) -> Vec<(FragmentId, FragmentId)>

fn collect_reschedule_blocked_jobs_for_creating_jobs_inflight( &self, ) -> MetaResult<HashSet<JobId>>

fn collect_reschedule_blocked_job_ids( &self, reschedules: &HashMap<FragmentId, Reschedule>, fragment_actors: &HashMap<FragmentId, HashSet<ActorId>>, blocked_job_ids: &HashSet<JobId>, ) -> HashSet<JobId>

fn next_complete_barrier_task( &mut self, task: &mut Option<CompleteBarrierTask>, context: Option<(&mut PeriodicBarriers, &mut PartialGraphManager)>, hummock_version_stats: &HummockVersionStats, )

fn ack_completed( &mut self, command_prev_epoch: Option<u64>, creating_job_epochs: Vec<(JobId, u64)>, )

fn handle_refresh_table_info( &self, task: &mut Option<CompleteBarrierTask>, node: &EpochNode, )

impl DatabaseCheckpointControl

fn handle_new_barrier( &mut self, command: Option<(Command, Vec<Notifier>)>, checkpoint: bool, span: Span, partial_graph_manager: &mut PartialGraphManager, hummock_version_stats: &HummockVersionStats, ) -> MetaResult<()>

Handle the new barrier from the scheduled queue and inject it.

impl DatabaseCheckpointControl

fn collect_base_info( &self, ) -> (HashSet<TableId>, HashMap<WorkerId, HashSet<ActorId>>)

Collect table IDs to commit and actor IDs to collect from current fragment infos.

fn apply_simple_command( &self, mutation: Option<Mutation>, command_name: &'static str, ) -> (Option<Mutation>, HashSet<TableId>, Option<StreamJobActorsToCreate>, HashMap<WorkerId, HashSet<ActorId>>, PostCollectCommand)

Helper for the simplest command variants: those that only need a pre-computed mutation and a command name, with no actors to create and no additional side effects on self.

pub(super) fn apply_command( &mut self, command: Option<Command>, notifiers: &mut Vec<Notifier>, barrier_info: &BarrierInfo, partial_graph_manager: &mut PartialGraphManager, hummock_version_stats: &HummockVersionStats, ) -> MetaResult<ApplyCommandInfo>

Returns the inflight actor infos that have included the newly added actors in the given command. The dropped actors will be removed from the state after the info get resolved.