Struct CreatingStreamingJobControl 

pub(crate) struct CreatingStreamingJobControl {
    control: IndependentJobControl,
    node_actors: HashMap<WorkerId, HashSet<ActorId>>,
    status: CreatingStreamingJobStatus,
    max_lagged_barrier_num: usize,
    max_pending_barrier_num: usize,
    upstream_lag: LabelGuardedIntGauge,
}

Fields

control: IndependentJobControl

node_actors: HashMap<WorkerId, HashSet<ActorId>>

status: CreatingStreamingJobStatus

max_lagged_barrier_num: usize

max_pending_barrier_num: usize

upstream_lag: LabelGuardedIntGauge

Implementations

impl CreatingStreamingJobControl

pub(crate) fn new<'a>( entry: VacantEntry<'a, JobId, IndependentCheckpointJobControl>, create_info: CreateIndependentStreamingJobCommandInfo, notifier: Option<&mut NotifierStarter>, snapshot_backfill_upstream_tables: HashSet<TableId>, snapshot_epoch: u64, since_timestamp_upstream_log_epochs: Option<(&Vec<(Vec<u64>, u64)>, PartialGraphId, u64)>, version_stat: &HummockVersionStats, term_id: &str, partial_graph_manager: &mut PartialGraphManager, edges: &mut FragmentEdgeBuildResult, split_assignment: &SplitAssignment, actors: &RenderResult, ) -> MetaResult<&'a mut Self>

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

pub(super) fn resolve_upstream_log_epochs<'a>( snapshot_backfill_upstream_tables: &HashSet<TableId>, upstream_table_log_epochs: &'a HashMap<TableId, Vec<(Vec<u64>, u64)>>, upstream_lag_start_epoch: u64, upstream_barrier_info: &BarrierInfo, ) -> MetaResult<(u64, &'a [(Vec<u64>, u64)])>

Resolve upstream log epoch groups after the epoch where this job starts lagging upstream.

The start epoch may be synthetic and fall between an upstream checkpoint and the next upstream epoch. The returned checkpoint is the last upstream checkpoint at or before the start epoch, so callers that require exact checkpoint alignment can validate that policy before composing the returned raw epoch groups into barriers.

pub(super) fn resolve_checkpoint_aligned_upstream_log_barriers( snapshot_backfill_upstream_tables: &HashSet<TableId>, upstream_table_log_epochs: &HashMap<TableId, Vec<(Vec<u64>, u64)>>, upstream_lag_start_epoch: u64, upstream_barrier_info: &BarrierInfo, ) -> MetaResult<Vec<BarrierInfo>>

pub(super) fn compose_upstream_log_barriers( upstream_lag_start_epoch: u64, upstream_log_epochs: &[(Vec<u64>, u64)], upstream_barrier_info: &BarrierInfo, ) -> MetaResult<Vec<BarrierInfo>>

fn resolve_since_timestamp_upstream_log_epochs( upstream_log_epochs: &Vec<(Vec<u64>, u64)>, pending_upstream_barriers: impl Iterator<Item = &BarrierInfo>, snapshot_epoch: u64, new_upstream_barrier_prev_epoch: u64, ) -> MetaResult<(BarrierInfo, Vec<BarrierInfo>)>

Resolves the log-store barriers that must be injected before the create barrier.

Example with pending upstream barriers:

snapshot epoch: 60
changelog after snapshot: [61, 62, 63, 64] + 65
pending upstream barriers: 66 -> 67 barrier, 67 -> 68 barrier,
                           68 -> 69 barrier, 69 -> 70 barrier
new create barrier: 70 -> 71

injected: 60 -> 61 checkpoint, 61 -> 62 barrier, ..., 64 -> 65 barrier,
          65 -> 66 checkpoint, 66 -> 67 barrier, 67 -> 68 barrier, ...,
          69 -> 70 barrier
current create barrier later injects: 70 -> 71 checkpoint

Example without pending upstream barriers:

snapshot epoch: 60
changelog after snapshot: [61, 62, 63, 64] + 65
new create barrier: 66 -> 67

injected: 60 -> 61 checkpoint, 61 -> 62 barrier, ..., 64 -> 65 barrier,
          65 -> 66 checkpoint
current create barrier later injects: 66 -> 67 checkpoint

fn recover_consuming_snapshot( job_id: JobId, upstream_table_log_epochs: &HashMap<TableId, Vec<(Vec<u64>, u64)>>, snapshot_epoch: u64, committed_epoch: u64, upstream_barrier_info: &BarrierInfo, info: CreatingJobInfo, backfill_order: &ExtendedFragmentBackfillOrder, version_stat: &HummockVersionStats, ) -> MetaResult<(CreatingStreamingJobStatus, BarrierInfo)>

fn recover_consuming_log_store( job_id: JobId, upstream_table_log_epochs: &HashMap<TableId, Vec<(Vec<u64>, u64)>>, committed_epoch: u64, upstream_barrier_info: &BarrierInfo, info: CreatingJobInfo, ) -> MetaResult<(CreatingStreamingJobStatus, BarrierInfo)>

pub(crate) fn recover( control: IndependentJobControl, upstream_table_log_epochs: &HashMap<TableId, Vec<(Vec<u64>, u64)>>, upstream_barrier_info: &BarrierInfo, fragment_infos: HashMap<FragmentId, InflightFragmentInfo>, backfill_order: ExtendedFragmentBackfillOrder, fragment_relations: &FragmentDownstreamRelation, version_stat: &HummockVersionStats, new_actors: StreamJobActorsToCreate, initial_mutation: Mutation, term_id: &str, partial_graph_recoverer: &mut PartialGraphRecoverer<'_>, ) -> MetaResult<Self>

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

pub(super) fn pinned_upstream_tables(&self) -> &HashSet<TableId>

fn inject_barrier( partial_graph_id: PartialGraphId, partial_graph_manager: &mut PartialGraphManager, node_actors: &HashMap<WorkerId, HashSet<ActorId>>, state_table_ids: &HashSet<TableId>, is_finishing: bool, barrier_info: BarrierInfo, new_actors: Option<StreamJobActorsToCreate>, mutation: Option<Mutation>, notifier: Option<&mut NotifierStarter>, first_create_info: Option<CreateIndependentStreamingJobCommandInfo>, ) -> MetaResult<()>

pub(crate) fn start_consume_upstream( &mut self, barrier_info: &BarrierInfo, ) -> &CreatingJobInfo

pub(crate) fn start_finishing( &mut self, partial_graph_manager: &mut PartialGraphManager, barrier_info: &BarrierInfo, ) -> MetaResult<CreatingJobInfo>

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

pub(crate) fn pre_apply_throttle( &mut self, config: &mut HashMap<FragmentId, (ThrottleConfig, PbStreamNode)>, ) -> Option<Mutation>

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

Returns whether the next barrier should be forced to a checkpoint.

fn is_ready_to_consume_upstream(&self) -> bool

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

pub(crate) fn should_start_consume_upstream( &self, partial_graph_manager: &PartialGraphManager, ) -> bool

impl CreatingStreamingJobControl

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

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

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

pub fn into_tracking_job(self) -> TrackingJob

pub(super) fn can_drop_independently(&self) -> bool

Whether the job can be dropped by resetting its independent partial graph.

A finishing job has already been merged into the database graph, so it must be handled by the database-graph drop command instead.

pub(super) fn can_drop(&self) -> bool

Trait Implementations

impl Debug for CreatingStreamingJobControl

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

Formats the value using the given formatter.