Struct StreamJobFragments

pub struct StreamJobFragments {
    stream_job_id: TableId,
    state: State,
    pub fragments: BTreeMap<FragmentId, Fragment>,
    pub actor_status: BTreeMap<ActorId, ActorStatus>,
    pub actor_splits: HashMap<ActorId, Vec<SplitImpl>>,
    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

actor_splits: HashMap<ActorId, Vec<SplitImpl>>

The splits of actors, incl. both Source and SourceBackfill 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 fn to_protobuf(&self) -> PbTableFragments

pub fn from_protobuf(prost: PbTableFragments) -> Self

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, WorkerSlotId>, 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 stream_job_id(&self) -> TableId

Returns the table id.

pub fn state(&self) -> State

Returns the state of the table fragments.

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 is_initial(&self) -> bool

Returns whether the table fragments is in Initial state.

pub fn set_state(&mut self, state: State)

Set the state of the table fragments.

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

Update state of all actors

pub fn set_actor_splits_by_split_assignment( &mut self, split_assignment: SplitAssignment, )

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 actors(&self) -> Vec<StreamActor>

Returns actors associated with this table.

pub fn filter_actor_ids( &self, check_type: impl Fn(u32) -> bool + 'static, ) -> impl Iterator<Item = ActorId> + '_

Returns the actor ids with the given fragment type.

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

Returns mview actor ids.

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

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

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 snapshot_backfill_actor_ids(&self) -> HashSet<ActorId>

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, ) -> MetadataModelResult<HashMap<SourceId, BTreeSet<(FragmentId, FragmentId)>>>

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

Resolve dependent table

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

Returns a mapping of dependent table ids of the TableFragments to their corresponding count.

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) -> HashMap<WorkerId, Vec<StreamActor>>

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

pub fn internal_tables(&self) -> BTreeMap<u32, Table>

Retrieve the complete internal tables map of the whole graph.

Compared to crate::stream::StreamFragmentGraph::incomplete_internal_tables, the table catalogs returned here are complete, with all fields filled.

pub fn all_tables(&self) -> BTreeMap<u32, Table>

internal_tables() with additional table in Materialize node.

fn collect_tables_inner( &self, internal_tables_only: bool, ) -> 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 copy 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.