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, ) -> MetaResult<Vec<(String, StreamJobFragments)>>

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

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: &HashMap<TableId, (String, StreamJobFragments)>, version: &HummockVersion, ) -> MetaResult<(HashMap<TableId, u64>, HashMap<TableId, Vec<(Vec<u64>, u64)>>)>

async fn recovery_table_with_upstream_sinks( &self, inflight_jobs: &mut HashMap<DatabaseId, HashMap<TableId, InflightStreamingJobInfo>>, ) -> 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>>

impl GlobalBarrierWorkerContextImpl

const RECOVERY_FORCE_MIGRATION_TIMEOUT: Duration

async fn migrate_actors( &self, active_nodes: &mut ActiveStreamingWorkerNodes, ) -> MetaResult<HashMap<DatabaseId, HashMap<TableId, InflightStreamingJobInfo>>>

Migrate actors in expired CNs to newly joined ones, return true if any actor is migrated.

async fn scale_actors( &self, active_nodes: &ActiveStreamingWorkerNodes, ) -> MetaResult<()>

fn derive_target_parallelism( available_parallelism: usize, assigned_parallelism: TableParallelism, actual_fragment_parallelism: Option<usize>, default_parallelism: DefaultParallelism, ) -> TableParallelism

async fn load_all_actors(&self) -> MetaResult<HashMap<ActorId, StreamActor>>

Update all actors in compute nodes.

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_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<()>