Struct BarrierInner

pub struct BarrierInner<M> {
    pub epoch: EpochPair,
    pub mutation: M,
    pub kind: BarrierKind,
    pub tracing_context: TracingContext,
    pub passed_actors: Vec<ActorId>,
}

The generic type M is the mutation type of the barrier.

For barrier of in the dispatcher, M is (), which means the mutation is erased. For barrier flowing within the streaming actor, M is the normal BarrierMutationType.

Fields

epoch: EpochPair

mutation: M

kind: BarrierKind

tracing_context: TracingContext

Tracing context for the current epoch of this barrier.

passed_actors: Vec<ActorId>

The actors that this barrier has passed locally. Used for debugging only.

Implementations

impl<M: Default> BarrierInner<M>

pub fn new_test_barrier(epoch: u64) -> Self

Create a plain barrier.

pub fn with_prev_epoch_for_test(epoch: u64, prev_epoch: u64) -> Self

impl BarrierInner<Option<Arc<Mutation>>>

pub fn into_dispatcher(self) -> DispatcherBarrier

pub fn with_mutation(self, mutation: Mutation) -> Self

pub fn with_stop(self) -> Self

pub fn is_with_stop_mutation(&self) -> bool

Whether this barrier carries stop mutation.

pub fn is_stop(&self, actor_id: ActorId) -> bool

Whether this barrier is to stop the actor with actor_id.

pub fn initial_split_assignment( &self, actor_id: ActorId, ) -> Option<&[SplitImpl]>

Get the initial split assignments for the actor with actor_id.

This should only be called on the initial barrier received by the executor. It must be

• Add mutation when it’s a new streaming job, or recovery.
• Update mutation when it’s created for scaling.
• AddAndUpdate mutation when it’s created for sink-into-table.

Note that SourceChangeSplit is not included, because it’s only used for changing splits of existing executors.

pub fn all_stop_actors(&self) -> Option<&HashSet<ActorId>>

Get all actors that to be stopped (dropped) by this barrier.

pub fn is_newly_added(&self, actor_id: ActorId) -> bool

Whether this barrier is to newly add the actor with actor_id. This is used for Chain and Values to decide whether to output the existing (historical) data.

By “newly”, we mean the actor belongs to a subgraph of a new streaming job. That is, actors added for scaling are not included.

pub fn has_more_downstream_fragments(&self, upstream_actor_id: ActorId) -> bool

Whether this barrier adds new downstream fragment for the actor with upstream_actor_id.

Use case

Some optimizations are applied when an actor doesn’t have any downstreams (“standalone” actors).

• Pause a standalone shared SourceExecutor.
• Disable a standalone MaterializeExecutor’s conflict check.

This is implemented by checking actor_context.initial_dispatch_num on startup, and check has_more_downstream_fragments on barrier to see whether the optimization needs to be turned off.

Some special cases not included

Note that this is not has_new_downstream_actor/fragment. For our use case, we only care about number of downstream fragments (more precisely, existence).

• When scaling, the number of downstream actors is changed, and they are “new”, but downstream fragments is not changed.
• When ALTER TABLE sink_into_table, the fragment is replaced with a “new” one, but the number is not changed.

pub fn is_pause_on_startup(&self) -> bool

Whether this barrier requires the executor to pause its data stream on startup.

pub fn is_resume(&self) -> bool

Whether this barrier is for resume.

pub fn as_update_merge( &self, actor_id: ActorId, upstream_fragment_id: FragmentId, ) -> Option<&MergeUpdate>

Returns the MergeUpdate if this barrier is to update the merge executors for the actor with actor_id.

pub fn as_update_vnode_bitmap(&self, actor_id: ActorId) -> Option<Arc<Bitmap>>

Returns the new vnode bitmap if this barrier is to update the vnode bitmap for the actor with actor_id.

Actually, this vnode bitmap update is only useful for the record accessing validation for distributed executors, since the read/write pattern will never be across multiple vnodes.

pub fn get_curr_epoch(&self) -> Epoch

pub fn tracing_context(&self) -> &TracingContext

Retrieve the tracing context for the current epoch of this barrier.

pub fn added_subscriber_on_mv_table( &self, mv_table_id: TableId, ) -> impl Iterator<Item = u32> + '_

impl<M> BarrierInner<M>

fn to_protobuf_inner( &self, barrier_fn: impl FnOnce(&M) -> Option<PbMutation>, ) -> PbBarrier

fn from_protobuf_inner( prost: &PbBarrier, mutation_from_pb: impl FnOnce(Option<&PbMutation>) -> StreamExecutorResult<M>, ) -> StreamExecutorResult<Self>

pub fn map_mutation<M2>(self, f: impl FnOnce(M) -> M2) -> BarrierInner<M2>

impl BarrierInner<()>

pub fn to_protobuf(&self) -> PbBarrier

impl BarrierInner<Option<Arc<Mutation>>>

pub fn to_protobuf(&self) -> PbBarrier

pub fn from_protobuf(prost: &PbBarrier) -> StreamExecutorResult<Self>

Trait Implementations

impl<M: Clone> Clone for BarrierInner<M>

fn clone(&self) -> BarrierInner<M>

Returns a copy of the value.

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

Performs copy-assignment from source.

impl<M: Debug> Debug for BarrierInner<M>

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

Formats the value using the given formatter.

impl<M: PartialEq> PartialEq for BarrierInner<M>

fn eq(&self, other: &Self) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.