Enum Command

pub enum Command {
    Flush,
    Pause(PausedReason),
    Resume(PausedReason),
    DropStreamingJobs {
        table_fragments_ids: HashSet<TableId>,
        actors: Vec<ActorId>,
        unregistered_state_table_ids: HashSet<TableId>,
        unregistered_fragment_ids: HashSet<FragmentId>,
    },
    CreateStreamingJob {
        info: CreateStreamingJobCommandInfo,
        job_type: CreateStreamingJobType,
    },
    MergeSnapshotBackfillStreamingJobs(HashMap<TableId, (SnapshotBackfillInfo, InflightStreamingJobInfo)>),
    RescheduleFragment {
        reschedules: HashMap<FragmentId, Reschedule>,
        table_parallelism: HashMap<TableId, TableParallelism>,
        fragment_actors: HashMap<FragmentId, HashSet<ActorId>>,
    },
    ReplaceTable(ReplaceTablePlan),
    SourceSplitAssignment(SplitAssignment),
    Throttle(ThrottleConfig),
    CreateSubscription {
        subscription_id: u32,
        upstream_mv_table_id: TableId,
        retention_second: u64,
    },
    DropSubscription {
        subscription_id: u32,
        upstream_mv_table_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(PausedReason)

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

Resume(PausedReason)

Resume command generates a Resume barrier with the provided PausedReason 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 table fragments info from meta store.

Fields

table_fragments_ids: HashSet<TableId>

actors: Vec<ActorId>

unregistered_state_table_ids: HashSet<TableId>

unregistered_fragment_ids: HashSet<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 table 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

MergeSnapshotBackfillStreamingJobs(HashMap<TableId, (SnapshotBackfillInfo, 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>

table_parallelism: HashMap<TableId, TableParallelism>

fragment_actors: HashMap<FragmentId, HashSet<ActorId>>

ReplaceTable(ReplaceTablePlan)

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

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

SourceSplitAssignment(SplitAssignment)

SourceSplitAssignment 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

Implementations

impl Command

pub fn pause(reason: PausedReason) -> Self

pub fn resume(reason: PausedReason) -> Self

pub fn cancel(table_fragments: &TableFragments) -> Self

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

pub fn should_pause_inject_barrier(&self) -> bool

If we need to send a barrier to modify actor configuration, we will pause the barrier injection. return true.

pub fn need_checkpoint(&self) -> bool

impl Command

pub fn to_mutation( &self, current_paused_reason: Option<PausedReason>, ) -> Option<Mutation>

Generate a mutation for the given command.

pub fn actors_to_create(&self) -> Option<HashMap<WorkerId, Vec<StreamActor>>>

fn generate_update_mutation_for_replace_table( old_table_fragments: &TableFragments, merge_updates: &[MergeUpdate], dispatchers: &HashMap<ActorId, Vec<Dispatcher>>, init_split_assignment: &SplitAssignment, ) -> Option<Mutation>

pub fn next_paused_reason( this: Option<&Self>, current_paused_reason: Option<PausedReason>, ) -> Option<PausedReason>

Returns the paused reason after executing the current command.

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

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

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<(), Error>

Formats the value using the given formatter.