Struct DatabaseCheckpointControl 

pub(crate) struct DatabaseCheckpointControl {
    database_id: DatabaseId,
    state: BarrierWorkerState,
    command_ctx_queue: BTreeMap<u64, EpochNode>,
    completing_barrier: Option<u64>,
    committed_epoch: Option<u64>,
    creating_streaming_job_controls: HashMap<JobId, CreatingStreamingJobControl>,
    cdc_table_backfill_tracker: CdcTableBackfillTrackerRef,
    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>

creating_streaming_job_controls: HashMap<JobId, CreatingStreamingJobControl>

cdc_table_backfill_tracker: CdcTableBackfillTrackerRef

metrics: DatabaseCheckpointControlMetrics

Implementations

impl DatabaseCheckpointControl

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

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

fn total_command_num(&self) -> usize

fn update_barrier_nums_metrics(&self)

Update the metrics of barrier nums.

fn jobs_to_merge( &self, ) -> Option<HashMap<JobId, (HashSet<TableId>, HashMap<FragmentId, InflightFragmentInfo>)>>

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

Enqueue a barrier command

fn barrier_collected( &mut self, resp: BarrierCompleteResponse, 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.

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

Return whether the database can still work after worker failure

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 next_complete_barrier_task( &mut self, task: &mut Option<CompleteBarrierTask>, context: Option<(&mut PeriodicBarriers, &mut ControlStreamManager)>, 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, control_stream_manager: &mut ControlStreamManager, hummock_version_stats: &HummockVersionStats, ) -> MetaResult<()>

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