Struct GlobalBarrierWorkerContextImpl 

pub(super) struct GlobalBarrierWorkerContextImpl {
    scheduled_barriers: ScheduledBarriers,
    status: Arc<ArcSwap<BarrierManagerStatus>>,
    pub(super) metadata_manager: MetadataManager,
    hummock_manager: HummockManagerRef,
    source_manager: SourceManagerRef,
    _scale_controller: ScaleControllerRef,
    pub(super) env: MetaSrvEnv,
    barrier_scheduler: BarrierScheduler,
}

Fields

scheduled_barriers: ScheduledBarriers

status: Arc<ArcSwap<BarrierManagerStatus>>

metadata_manager: MetadataManager

hummock_manager: HummockManagerRef

source_manager: SourceManagerRef

_scale_controller: ScaleControllerRef

env: MetaSrvEnv

barrier_scheduler: BarrierScheduler

Barrier scheduler for scheduling load finish commands

Implementations

impl GlobalBarrierWorkerContextImpl

fn set_status(&self, new_status: BarrierManagerStatus)

impl GlobalBarrierWorkerContextImpl

async fn clean_dirty_streaming_jobs( &self, database_id: Option<DatabaseId>, ) -> MetaResult<()>

Clean catalogs for creating streaming jobs that are in foreground mode or table fragments not persisted.

async fn purge_state_table_from_hummock( &self, all_state_table_ids: &HashSet<TableId>, ) -> MetaResult<()>

async fn list_background_job_progress( &self, database_id: Option<DatabaseId>, ) -> MetaResult<HashMap<TableId, String>>

async fn resolve_graph_info( &self, database_id: Option<DatabaseId>, worker_nodes: &ActiveStreamingWorkerNodes, ) -> MetaResult<HashMap<DatabaseId, HashMap<TableId, HashMap<FragmentId, InflightFragmentInfo>>>>

Resolve actor information from cluster, fragment manager and ChangedTableId. We use changed_table_id to modify the actors to be sent or collected. Because these actor will create or drop before this barrier flow through them.

fn resolve_hummock_version_epochs( background_jobs: impl Iterator<Item = (TableId, &HashMap<FragmentId, InflightFragmentInfo>)>, version: &HummockVersion, ) -> MetaResult<(HashMap<TableId, u64>, HashMap<TableId, Vec<(Vec<u64>, u64)>>)>

fn collect_cdc_table_backfill_actors<'a, I>( jobs: I, ) -> HashMap<u32, HashSet<ActorId>>

where I: Iterator<Item = (&'a TableId, &'a HashMap<FragmentId, InflightFragmentInfo>)>,

async fn recovery_table_with_upstream_sinks( &self, inflight_jobs: &mut HashMap<DatabaseId, HashMap<TableId, HashMap<FragmentId, InflightFragmentInfo>>>, ) -> MetaResult<()>

For normal DDL operations, the UpstreamSinkUnion operator is modified dynamically, and does not persist the newly added or deleted upstreams in meta-store. Therefore, when restoring jobs, we need to restore the information required by the operator based on the current state of the upstream (sink) and downstream (table) of the operator.

pub(super) async fn reload_runtime_info_impl( &self, ) -> MetaResult<BarrierWorkerRuntimeInfoSnapshot>

pub(super) async fn reload_database_runtime_info_impl( &self, database_id: DatabaseId, ) -> MetaResult<Option<DatabaseRuntimeInfoSnapshot>>

async fn load_stream_actors( &self, all_info: &HashMap<DatabaseId, HashMap<TableId, HashMap<FragmentId, InflightFragmentInfo>>>, ) -> MetaResult<HashMap<ActorId, StreamActor>>

impl GlobalBarrierWorkerContextImpl

pub(super) fn new( scheduled_barriers: ScheduledBarriers, status: Arc<ArcSwap<BarrierManagerStatus>>, metadata_manager: MetadataManager, hummock_manager: HummockManagerRef, source_manager: SourceManagerRef, scale_controller: ScaleControllerRef, env: MetaSrvEnv, barrier_scheduler: BarrierScheduler, ) -> Self

pub(super) fn status(&self) -> Arc<ArcSwap<BarrierManagerStatus>>

impl GlobalBarrierWorkerContextImpl

pub(super) async fn new_control_stream_impl( &self, node: &WorkerNode, init_request: &PbInitRequest, ) -> MetaResult<StreamingControlHandle>

Trait Implementations

impl GlobalBarrierWorkerContext for GlobalBarrierWorkerContextImpl

async fn commit_epoch( &self, commit_info: CommitEpochInfo, ) -> MetaResult<HummockVersionStats>

async fn next_scheduled(&self) -> Scheduled

fn abort_and_mark_blocked( &self, database_id: Option<DatabaseId>, recovery_reason: RecoveryReason, )

fn mark_ready(&self, options: MarkReadyOptions)

async fn post_collect_command<'a>( &'a self, command: &'a CommandContext, ) -> MetaResult<()>

async fn notify_creating_job_failed( &self, database_id: Option<DatabaseId>, err: String, )

async fn finish_creating_job(&self, job: TrackingJob) -> MetaResult<()>

async fn finish_cdc_table_backfill(&self, job: TableId) -> MetaResult<()>

async fn new_control_stream( &self, node: &WorkerNode, init_request: &PbInitRequest, ) -> MetaResult<StreamingControlHandle>

async fn reload_runtime_info( &self, ) -> MetaResult<BarrierWorkerRuntimeInfoSnapshot>

async fn reload_database_runtime_info( &self, database_id: DatabaseId, ) -> MetaResult<Option<DatabaseRuntimeInfoSnapshot>>

async fn handle_list_finished_source_ids( &self, list_finished_source_ids: Vec<u32>, ) -> MetaResult<()>

async fn handle_load_finished_source_ids( &self, load_finished_source_ids: Vec<u32>, ) -> MetaResult<()>

async fn handle_refresh_finished_table_ids( &self, refresh_finished_table_ids: Vec<u32>, ) -> MetaResult<()>