Enum Command 

pub enum Command {
    Flush,
    Pause,
    Resume,
    DropStreamingJobs {
        streaming_job_ids: HashSet<JobId>,
        actors: Vec<ActorId>,
        unregistered_state_table_ids: HashSet<TableId>,
        unregistered_fragment_ids: HashSet<FragmentId>,
        dropped_sink_fragment_by_targets: HashMap<FragmentId, Vec<FragmentId>>,
    },
    CreateStreamingJob {
        info: CreateStreamingJobCommandInfo,
        job_type: CreateStreamingJobType,
        cross_db_snapshot_backfill_info: SnapshotBackfillInfo,
    },
    MergeSnapshotBackfillStreamingJobs(HashMap<JobId, (HashSet<TableId>, InflightStreamingJobInfo)>),
    RescheduleFragment {
        reschedules: HashMap<FragmentId, Reschedule>,
        fragment_actors: HashMap<FragmentId, HashSet<ActorId>>,
    },
    ReplaceStreamJob(ReplaceStreamJobPlan),
    SourceChangeSplit(SplitState),
    Throttle(ThrottleConfig),
    CreateSubscription {
        subscription_id: u32,
        upstream_mv_table_id: TableId,
        retention_second: u64,
    },
    DropSubscription {
        subscription_id: u32,
        upstream_mv_table_id: TableId,
    },
    ConnectorPropsChange(ConnectorPropsChange),
    StartFragmentBackfill {
        fragment_ids: Vec<FragmentId>,
    },
    Refresh {
        table_id: TableId,
        associated_source_id: TableId,
    },
    ListFinish {
        table_id: TableId,
        associated_source_id: TableId,
    },
    LoadFinish {
        table_id: TableId,
        associated_source_id: TableId,
    },
}

Command is the input of crate::barrier::worker::GlobalBarrierWorker. For different commands, it will build different barriers to send, and may do different stuffs after the barrier is collected.

Variants

Flush

Flush command will generate a checkpoint barrier. After the barrier is collected and committed all messages before the checkpoint barrier should have been committed.

Pause

Pause command generates a Pause barrier only if the cluster is not already paused. Otherwise, a barrier with no mutation will be generated.

Resume

Resume command generates a Resume barrier only if the cluster is paused with the same reason. Otherwise, a barrier with no mutation will be generated.

DropStreamingJobs

DropStreamingJobs command generates a Stop barrier to stop the given Vec<ActorId>. The catalog has ensured that these streaming jobs are safe to be dropped by reference counts before.

Barriers from the actors to be dropped will STILL be collected. After the barrier is collected, it notifies the local stream manager of compute nodes to drop actors, and then delete the job fragments info from meta store.

Fields

streaming_job_ids: HashSet<JobId>

actors: Vec<ActorId>

unregistered_state_table_ids: HashSet<TableId>

unregistered_fragment_ids: HashSet<FragmentId>

dropped_sink_fragment_by_targets: HashMap<FragmentId, Vec<FragmentId>>

CreateStreamingJob

CreateStreamingJob command generates a Add barrier by given info.

Barriers from the actors to be created, which is marked as Inactive at first, will STILL be collected since the barrier should be passthrough.

After the barrier is collected, these newly created actors will be marked as Running. And it adds the job fragments info to meta store. However, the creating progress will last for a while until the finish channel is signaled, then the state of TableFragments will be set to Created.

Fields

info: CreateStreamingJobCommandInfo

job_type: CreateStreamingJobType

cross_db_snapshot_backfill_info: SnapshotBackfillInfo

MergeSnapshotBackfillStreamingJobs(HashMap<JobId, (HashSet<TableId>, InflightStreamingJobInfo)>)

RescheduleFragment

Reschedule command generates a Update barrier by the Reschedule of each fragment. Mainly used for scaling and migration.

Barriers from which actors should be collected, and the post behavior of this command are very similar to Create and Drop commands, for added and removed actors, respectively.

Fields

reschedules: HashMap<FragmentId, Reschedule>

fragment_actors: HashMap<FragmentId, HashSet<ActorId>>

ReplaceStreamJob(ReplaceStreamJobPlan)

ReplaceStreamJob command generates a Update barrier with the given replace_upstream. This is essentially switching the downstream of the old job fragments to the new ones, and dropping the old job fragments. Used for schema change.

This can be treated as a special case of RescheduleFragment, while the upstream fragment of the Merge executors are changed additionally.

SourceChangeSplit(SplitState)

SourceChangeSplit generates a Splits barrier for pushing initialized splits or changed splits.

Throttle(ThrottleConfig)

Throttle command generates a Throttle barrier with the given throttle config to change the rate_limit of FlowControl Executor after StreamScan or Source.

CreateSubscription

CreateSubscription command generates a CreateSubscriptionMutation to notify materialize executor to start storing old value for subscription.

Fields

subscription_id: u32

upstream_mv_table_id: TableId

retention_second: u64

DropSubscription

DropSubscription command generates a DropSubscriptionsMutation to notify materialize executor to stop storing old value when there is no subscription depending on it.

Fields

subscription_id: u32

upstream_mv_table_id: TableId

ConnectorPropsChange(ConnectorPropsChange)

StartFragmentBackfill

StartFragmentBackfill command will trigger backfilling for specified scans by fragment_id.

Fields

fragment_ids: Vec<FragmentId>

Refresh

Refresh command generates a barrier to refresh a table by truncating state and reloading data from source.

Fields

table_id: TableId

associated_source_id: TableId

ListFinish

Fields

table_id: TableId

associated_source_id: TableId

LoadFinish

Fields

table_id: TableId

associated_source_id: TableId

Implementations

impl Command

pub fn pause() -> Self

pub fn resume() -> Self

pub(crate) fn fragment_changes( &self, ) -> Option<(Option<JobId>, HashMap<FragmentId, CommandFragmentChanges>)>

pub fn need_checkpoint(&self) -> bool

impl Command

pub(super) fn to_mutation( &self, is_currently_paused: bool, edges: &mut Option<FragmentEdgeBuildResult>, control_stream_manager: &ControlStreamManager, ) -> Option<Mutation>

Generate a mutation for the given command.

edges contains the information of dispatchers of DispatchExecutor and actor_upstreamss of MergeNode

pub(super) fn actors_to_create( &self, graph_info: &InflightDatabaseInfo, edges: &mut Option<FragmentEdgeBuildResult>, control_stream_manager: &ControlStreamManager, ) -> Option<StreamJobActorsToCreate>

fn generate_update_mutation_for_replace_table( dropped_actors: impl IntoIterator<Item = ActorId>, merge_updates: HashMap<FragmentId, Vec<MergeUpdate>>, dispatchers: FragmentActorDispatchers, init_split_assignment: &SplitAssignment, cdc_table_snapshot_split_assignment: CdcTableSnapshotSplitAssignmentWithGeneration, auto_refresh_schema_sinks: Option<&Vec<AutoRefreshSchemaSinkContext>>, ) -> Option<Mutation>

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

For CancelStreamingJob, returns the table id of the target table.

impl Command

pub(super) fn collect_actor_upstreams( actor_dispatchers: impl Iterator<Item = (FragmentId, impl Iterator<Item = (ActorId, &[Dispatcher])>)>, reschedule_dispatcher_update: Option<(&HashMap<FragmentId, Reschedule>, &HashMap<FragmentId, HashSet<ActorId>>)>, graph_info: &InflightDatabaseInfo, control_stream_manager: &ControlStreamManager, ) -> HashMap<ActorId, ActorUpstreams>

Trait Implementations

impl Debug for Command

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

Formats the value using the given formatter.

impl Display for Command

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

Formats the value using the given formatter.