Struct StreamJobFragments 

pub struct StreamJobFragments {
    pub stream_job_id: TableId,
    pub state: State,
    pub fragments: BTreeMap<FragmentId, Fragment>,
    pub actor_status: BTreeMap<ActorId, ActorStatus>,
    pub ctx: StreamContext,
    pub assigned_parallelism: TableParallelism,
    pub max_parallelism: usize,
}

Fragments of a streaming job. Corresponds to [PbTableFragments]. (It was previously called TableFragments due to historical reasons.)

We store whole fragments in a single column family as follow: stream_job_id => StreamJobFragments.

Fields

stream_job_id: TableId

The table id.

state: State

The state of the table fragments.

fragments: BTreeMap<FragmentId, Fragment>

The table fragments.

actor_status: BTreeMap<ActorId, ActorStatus>

The status of actors

ctx: StreamContext

The streaming context associated with this stream plan and its fragments

assigned_parallelism: TableParallelism

The parallelism assigned to this table fragments

max_parallelism: usize

The max parallelism specified when the streaming job was created, i.e., expected vnode count.

The reason for persisting this value is mainly to check if a parallelism change (via ALTER .. SET PARALLELISM) is valid, so that the behavior can be consistent with the creation of the streaming job.

Note that the actual vnode count, denoted by vnode_count in fragments, may be different from this value (see StreamFragmentGraph.max_parallelism for more details.). As a result, checking the parallelism change with this value can be inaccurate in some cases. However, when generating resizing plans, we still take the vnode_count of each fragment into account.

Implementations

impl StreamJobFragments

pub(super) fn new_fragment_info<'a>( &'a self, assignment: &'a SplitAssignment, ) -> impl Iterator<Item = (FragmentId, InflightFragmentInfo)> + 'a

impl StreamJobFragments

pub fn to_protobuf( &self, fragment_upstreams: &HashMap<FragmentId, HashSet<FragmentId>>, fragment_dispatchers: &FragmentActorDispatchers, ) -> PbTableFragments

impl StreamJobFragments

pub fn for_test( table_id: TableId, fragments: BTreeMap<FragmentId, Fragment>, ) -> Self

Create a new TableFragments with state of Initial, with other fields empty.

pub fn new( stream_job_id: TableId, fragments: BTreeMap<FragmentId, Fragment>, actor_locations: &BTreeMap<ActorId, ActorAlignmentId>, ctx: StreamContext, table_parallelism: TableParallelism, max_parallelism: usize, ) -> Self

Create a new TableFragments with state of Initial, with the status of actors set to Inactive on the given workers.

pub fn fragment_ids(&self) -> impl Iterator<Item = FragmentId> + '_

pub fn fragments(&self) -> impl Iterator<Item = &Fragment>

pub fn fragment_actors(&self, fragment_id: FragmentId) -> &[StreamActor]

pub fn stream_job_id(&self) -> TableId

Returns the table id.

pub fn timezone(&self) -> Option<String>

Returns the timezone of the table

pub fn is_created(&self) -> bool

Returns whether the table fragments is in Created state.

pub fn update_actors_state(&mut self, state: ActorState)

Update state of all actors

pub fn actor_ids(&self) -> Vec<ActorId>

Returns actor ids associated with this table.

pub fn actor_fragment_mapping(&self) -> HashMap<ActorId, FragmentId>

pub fn tracking_progress_actor_ids( &self, ) -> Vec<(ActorId, BackfillUpstreamType)>

pub fn tracking_progress_actor_ids_impl( fragments: impl IntoIterator<Item = (FragmentTypeMask, impl Iterator<Item = ActorId>)>, ) -> Vec<(ActorId, BackfillUpstreamType)>

Returns actor ids that need to be tracked when creating MV.

pub fn root_fragment(&self) -> Option<Fragment>

pub fn mview_fragment(&self) -> Option<Fragment>

Returns the fragment with the Mview type flag.

pub fn source_fragment(&self) -> Option<Fragment>

pub fn sink_fragment(&self) -> Option<Fragment>

pub fn stream_source_fragments(&self) -> HashMap<SourceId, BTreeSet<FragmentId>>

Extract the fragments that include source executors that contains an external stream source, grouping by source id.

pub fn source_backfill_fragments( &self, ) -> HashMap<SourceId, BTreeSet<(FragmentId, FragmentId)>>

pub fn source_backfill_fragments_impl( fragments: impl Iterator<Item = (FragmentId, &StreamNode)>, ) -> HashMap<SourceId, BTreeSet<(FragmentId, FragmentId)>>

Returns (source_id, -> (source_backfill_fragment_id, upstream_source_fragment_id)).

Note: the fragment source_backfill_fragment_id may actually have multiple upstream fragments, but only one of them is the upstream source fragment, which is what we return.

pub fn union_fragment_for_table(&mut self) -> &mut Fragment

Find the table job’s Union fragment. Panics if not found.

fn resolve_dependent_table( stream_node: &StreamNode, table_ids: &mut HashMap<TableId, usize>, )

Resolve dependent table

pub fn upstream_table_counts(&self) -> HashMap<TableId, usize>

pub fn upstream_table_counts_impl( fragment_nodes: impl Iterator<Item = &StreamNode>, ) -> HashMap<TableId, usize>

Returns upstream table counts.

pub fn worker_actor_states( &self, ) -> BTreeMap<WorkerId, Vec<(ActorId, ActorState)>>

Returns states of actors group by worker id.

pub fn worker_actor_ids(&self) -> BTreeMap<WorkerId, Vec<ActorId>>

Returns actor locations group by worker id.

pub fn active_actors(&self) -> Vec<StreamActor>

Returns the status of actors group by worker id.

pub fn actors_to_create( &self, ) -> impl Iterator<Item = (FragmentId, &StreamNode, impl Iterator<Item = (&StreamActor, WorkerId)> + '_)> + '_

pub fn mv_table_id(&self) -> Option<u32>

pub fn collect_tables( fragments: impl Iterator<Item = &Fragment>, ) -> BTreeMap<u32, Table>

pub fn internal_table_ids(&self) -> Vec<u32>

Returns the internal table ids without the mview table.

pub fn all_table_ids(&self) -> impl Iterator<Item = u32> + '_

Returns all internal table ids including the mview table.

pub fn fill_expr_context(self) -> Self

Fill the expr_context in StreamActor. Used for compatibility.

Trait Implementations

impl Clone for StreamJobFragments

fn clone(&self) -> StreamJobFragments

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for StreamJobFragments

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

Formats the value using the given formatter.