Struct CompleteStreamFragmentGraph

pub struct CompleteStreamFragmentGraph {
    building_graph: StreamFragmentGraph,
    existing_fragments: HashMap<GlobalId<{ IdCategory::Fragment }>, Fragment>,
    existing_actor_location: HashMap<ActorId, WorkerId>,
    extra_downstreams: HashMap<GlobalId<{ IdCategory::Fragment }>, HashMap<GlobalId<{ IdCategory::Fragment }>, StreamFragmentEdge>>,
    extra_upstreams: HashMap<GlobalId<{ IdCategory::Fragment }>, HashMap<GlobalId<{ IdCategory::Fragment }>, StreamFragmentEdge>>,
}

A wrapper of StreamFragmentGraph that contains the additional information of pre-existing fragments, which are connected to the graph’s top-most or bottom-most fragments.

For example,

• if we’re going to build a mview on an existing mview, the upstream fragment containing the Materialize node will be included in this structure.
• if we’re going to replace the plan of a table with downstream mviews, the downstream fragments containing the StreamScan nodes will be included in this structure.

Fields

building_graph: StreamFragmentGraph

The fragment graph of the streaming job being built.

existing_fragments: HashMap<GlobalId<{ IdCategory::Fragment }>, Fragment>

The required information of existing fragments.

existing_actor_location: HashMap<ActorId, WorkerId>

The location of the actors in the existing fragments.

extra_downstreams: HashMap<GlobalId<{ IdCategory::Fragment }>, HashMap<GlobalId<{ IdCategory::Fragment }>, StreamFragmentEdge>>

Extra edges between existing fragments and the building fragments.

extra_upstreams: HashMap<GlobalId<{ IdCategory::Fragment }>, HashMap<GlobalId<{ IdCategory::Fragment }>, StreamFragmentEdge>>

Extra edges between existing fragments and the building fragments.

Implementations

impl CompleteStreamFragmentGraph

pub fn with_upstreams( graph: StreamFragmentGraph, upstream_root_fragments: HashMap<TableId, Fragment>, existing_actor_location: HashMap<ActorId, WorkerId>, ddl_type: DdlType, ) -> MetaResult<Self>

Create a new CompleteStreamFragmentGraph for MV on MV and CDC/Source Table with the upstream existing Materialize or Source fragments.

pub fn with_downstreams( graph: StreamFragmentGraph, original_table_fragment_id: FragmentId, downstream_fragments: Vec<(DispatchStrategy, Fragment)>, existing_actor_location: HashMap<ActorId, WorkerId>, ddl_type: DdlType, ) -> MetaResult<Self>

Create a new CompleteStreamFragmentGraph for replacing an existing table, with the downstream existing StreamScan fragments.

pub fn with_upstreams_and_downstreams( graph: StreamFragmentGraph, upstream_root_fragments: HashMap<TableId, Fragment>, upstream_actor_location: HashMap<ActorId, WorkerId>, original_table_fragment_id: FragmentId, downstream_fragments: Vec<(DispatchStrategy, Fragment)>, downstream_actor_location: HashMap<ActorId, WorkerId>, ddl_type: DdlType, ) -> MetaResult<Self>

For replacing an existing table based on shared cdc source

fn build_helper( graph: StreamFragmentGraph, upstream_ctx: Option<FragmentGraphUpstreamContext>, downstream_ctx: Option<FragmentGraphDownstreamContext>, ddl_type: DdlType, ) -> MetaResult<Self>

The core logic of building a CompleteStreamFragmentGraph, i.e., adding extra upstream/downstream fragments.

impl CompleteStreamFragmentGraph

pub(super) fn all_fragment_ids( &self, ) -> impl Iterator<Item = GlobalId<{ IdCategory::Fragment }>> + '_

Returns all fragment IDs in the complete graph, including the ones that are not in the building graph.

pub(super) fn all_edges( &self, ) -> impl Iterator<Item = (GlobalId<{ IdCategory::Fragment }>, GlobalId<{ IdCategory::Fragment }>, &StreamFragmentEdge)> + '_

Returns an iterator of all edges in the complete graph, including the external edges.

pub(super) fn existing_distribution( &self, ) -> HashMap<GlobalId<{ IdCategory::Fragment }>, Distribution>

Returns the distribution of the existing fragments.

pub(super) fn topo_order( &self, ) -> MetaResult<Vec<GlobalId<{ IdCategory::Fragment }>>>

Generate topological order of all fragments in this graph, including the ones that are not in the building graph. Returns error if the graph is not a DAG and topological sort can not be done.

For MV on MV, the first fragment popped out from the heap will be the top-most node, or the Sink / Materialize in stream graph.

pub(super) fn seal_fragment( &self, id: GlobalId<{ IdCategory::Fragment }>, actors: Vec<StreamActor>, distribution: Distribution, ) -> Fragment

Seal a BuildingFragment from the graph into a Fragment, which will be further used to build actors on the compute nodes and persist into meta store.

pub(super) fn get_fragment( &self, fragment_id: GlobalId<{ IdCategory::Fragment }>, ) -> EitherFragment

Get a fragment from the complete graph, which can be either a building fragment or an existing fragment.

pub(super) fn get_downstreams( &self, fragment_id: GlobalId<{ IdCategory::Fragment }>, ) -> impl Iterator<Item = (GlobalId<{ IdCategory::Fragment }>, &StreamFragmentEdge)>

Get all downstreams of a fragment, including the ones that are not in the building graph.

pub(super) fn get_upstreams( &self, fragment_id: GlobalId<{ IdCategory::Fragment }>, ) -> impl Iterator<Item = (GlobalId<{ IdCategory::Fragment }>, &StreamFragmentEdge)>

Get all upstreams of a fragment, including the ones that are not in the building graph.

pub(super) fn building_fragments( &self, ) -> &HashMap<GlobalId<{ IdCategory::Fragment }>, BuildingFragment>

Returns all building fragments in the graph.

pub(super) fn building_fragments_mut( &mut self, ) -> &mut HashMap<GlobalId<{ IdCategory::Fragment }>, BuildingFragment>

Returns all building fragments in the graph, mutable.

pub(super) fn max_parallelism(&self) -> usize

Get the expected vnode count of the building graph. See documentation of the field for more details.