Struct ScaleController

pub struct ScaleController {
    pub metadata_manager: MetadataManager,
    pub source_manager: SourceManagerRef,
    pub env: MetaSrvEnv,
    pub reschedule_lock: RwLock<()>,
}

Fields

metadata_manager: MetadataManager

source_manager: SourceManagerRef

env: MetaSrvEnv

reschedule_lock: RwLock<()>

We will acquire lock during DDL to prevent scaling operations on jobs that are in the creating state. e.g., a MV cannot be rescheduled during foreground backfill.

Implementations

impl ScaleController

pub fn new( metadata_manager: &MetadataManager, source_manager: SourceManagerRef, env: MetaSrvEnv, ) -> Self

pub async fn integrity_check(&self) -> MetaResult<()>

async fn build_reschedule_context( &self, reschedule: &mut HashMap<FragmentId, WorkerReschedule>, options: RescheduleOptions, table_parallelisms: Option<&mut HashMap<TableId, TableParallelism>>, ) -> MetaResult<RescheduleContext>

Build the context for rescheduling and do some validation for the request.

pub(crate) async fn analyze_reschedule_plan( &self, reschedules: HashMap<FragmentId, WorkerReschedule>, options: RescheduleOptions, table_parallelisms: Option<&mut HashMap<TableId, TableParallelism>>, ) -> MetaResult<HashMap<FragmentId, Reschedule>>

From the high-level WorkerReschedule to the low-level reschedule plan Reschedule.

Returns (reschedule_fragment, applied_reschedules)

• reschedule_fragment: the generated reschedule plan
• applied_reschedules: the changes that need to be updated to the meta store (pre_apply_reschedules, only for V1).

In normal process of scaling, we use the returned values to build a Command::RescheduleFragment, which will then flows through the barrier mechanism to perform scaling. Meta store is updated after the barrier is collected.

During recovery, we don’t need the barrier mechanism, and can directly use the returned values to update meta.

fn arrange_reschedules( &self, reschedule: &HashMap<FragmentId, WorkerReschedule>, ctx: &RescheduleContext, ) -> MetaResult<(HashMap<FragmentId, BTreeMap<ActorId, WorkerId>>, HashMap<FragmentId, BTreeMap<ActorId, WorkerId>>)>

fn modify_actor_upstream_and_downstream( ctx: &RescheduleContext, fragment_actors_to_remove: &HashMap<FragmentId, BTreeMap<ActorId, WorkerId>>, fragment_actors_to_create: &HashMap<FragmentId, BTreeMap<ActorId, WorkerId>>, fragment_actor_bitmap: &HashMap<FragmentId, HashMap<ActorId, Bitmap>>, no_shuffle_upstream_actor_map: &HashMap<ActorId, HashMap<FragmentId, ActorId>>, no_shuffle_downstream_actors_map: &HashMap<ActorId, HashMap<FragmentId, ActorId>>, new_actor: &mut PbStreamActor, ) -> MetaResult<()>

Modifies the upstream and downstream actors of the new created actor according to the overall changes, and is used to handle cascading updates

pub async fn post_apply_reschedule( &self, reschedules: &HashMap<FragmentId, Reschedule>, table_parallelism: &HashMap<TableId, TableParallelism>, ) -> MetaResult<()>

pub async fn generate_table_resize_plan( &self, policy: TableResizePolicy, ) -> MetaResult<HashMap<FragmentId, WorkerReschedule>>

pub(crate) fn filter_unschedulable_workers( workers: &[WorkerNode], ) -> HashSet<WorkerId>

fn diff_worker_slot_changes( fragment_worker_slots: &BTreeMap<WorkerId, usize>, target_worker_slots: &BTreeMap<WorkerId, usize>, ) -> WorkerReschedule

fn build_no_shuffle_relation_index( actor_map: &HashMap<ActorId, CustomActorInfo>, no_shuffle_source_fragment_ids: &mut HashSet<FragmentId>, no_shuffle_target_fragment_ids: &mut HashSet<FragmentId>, )

fn build_fragment_dispatcher_index( actor_map: &HashMap<ActorId, CustomActorInfo>, fragment_dispatcher_map: &mut HashMap<FragmentId, HashMap<FragmentId, DispatcherType>>, )

pub fn resolve_no_shuffle_upstream_tables( fragment_ids: HashSet<FragmentId>, fragment_map: &HashMap<FragmentId, CustomFragmentInfo>, no_shuffle_source_fragment_ids: &HashSet<FragmentId>, no_shuffle_target_fragment_ids: &HashSet<FragmentId>, fragment_to_table: &HashMap<FragmentId, TableId>, table_parallelisms: &mut HashMap<TableId, TableParallelism>, ) -> MetaResult<()>

pub fn resolve_no_shuffle_upstream_fragments<T>( reschedule: &mut HashMap<FragmentId, T>, fragment_map: &HashMap<FragmentId, CustomFragmentInfo>, no_shuffle_source_fragment_ids: &HashSet<FragmentId>, no_shuffle_target_fragment_ids: &HashSet<FragmentId>, ) -> MetaResult<()>

where T: Clone + Eq,