Struct GlobalStreamManager

pub struct GlobalStreamManager {
    pub env: MetaSrvEnv,
    pub metadata_manager: MetadataManager,
    pub barrier_scheduler: BarrierScheduler,
    pub source_manager: SourceManagerRef,
    creating_job_info: Arc<CreatingStreamingJobInfo>,
    pub scale_controller: ScaleControllerRef,
}

GlobalStreamManager manages all the streams in the system.

Fields

env: MetaSrvEnv

metadata_manager: MetadataManager

barrier_scheduler: BarrierScheduler

Broadcasts and collect barriers

source_manager: SourceManagerRef

Maintains streaming sources from external system like kafka

creating_job_info: Arc<CreatingStreamingJobInfo>

Creating streaming job info.

scale_controller: ScaleControllerRef

Implementations

impl GlobalStreamManager

pub async fn reschedule_lock_read_guard(&self) -> RwLockReadGuard<'_, ()>

pub async fn reschedule_lock_write_guard(&self) -> RwLockWriteGuard<'_, ()>

pub async fn reschedule_actors( &self, database_id: DatabaseId, reschedules: HashMap<FragmentId, WorkerReschedule>, options: RescheduleOptions, table_parallelism: Option<HashMap<TableId, TableParallelism>>, ) -> MetaResult<()>

The entrypoint of rescheduling actors.

Used by:

• The directly exposed low-level API risingwave_meta_service::scale_service::ScaleService (risectl meta reschedule)
• High-level parallelism control API
  • manual ALTER [TABLE | INDEX | MATERIALIZED VIEW | SINK] SET PARALLELISM
  • automatic parallelism control for TableParallelism::Adaptive when worker nodes changed

async fn trigger_parallelism_control(&self) -> MetaResult<bool>

When new worker nodes joined, or the parallelism of existing worker nodes changed, examines if there are any jobs can be scaled, and scales them if found.

This method will iterate over all CREATED jobs, and can be repeatedly called.

Returns

• Ok(false) if no jobs can be scaled;
• Ok(true) if some jobs are scaled, and it is possible that there are more jobs can be scaled.

async fn run(&self, shutdown_rx: Receiver<()>)

Handles notification of worker node activation and deletion, and triggers parallelism control.

pub fn start_auto_parallelism_monitor( self: Arc<Self>, ) -> (JoinHandle<()>, Sender<()>)

impl GlobalStreamManager

pub fn new( env: MetaSrvEnv, metadata_manager: MetadataManager, barrier_scheduler: BarrierScheduler, source_manager: SourceManagerRef, scale_controller: ScaleControllerRef, ) -> MetaResult<Self>

pub async fn create_streaming_job( self: &Arc<Self>, table_fragments: TableFragments, ctx: CreateStreamingJobContext, ) -> MetaResult<NotificationVersion>

Create streaming job, it works as follows:

1. Broadcast the actor info based on the scheduling result in the context, build the hanging channels in upstream worker nodes.
2. (optional) Get the split information of the StreamSource via source manager and patch actors.
3. Notify related worker nodes to update and build the actors.
4. Store related meta data.

async fn create_streaming_job_impl( &self, table_fragments: TableFragments, __arg2: CreateStreamingJobContext, ) -> MetaResult<NotificationVersion>

pub async fn replace_table( &self, table_fragments: TableFragments, __arg2: ReplaceTableContext, ) -> MetaResult<()>

pub async fn drop_streaming_jobs( &self, database_id: DatabaseId, removed_actors: Vec<ActorId>, streaming_job_ids: Vec<ObjectId>, state_table_ids: Vec<TableId>, fragment_ids: HashSet<FragmentId>, )

Drop streaming jobs by barrier manager, and clean up all related resources. The error will be ignored because the recovery process will take over it in cleaning part. Check Command::DropStreamingJobs for details.

pub async fn cancel_streaming_jobs( &self, table_ids: Vec<TableId>, ) -> Vec<TableId>

Cancel streaming jobs and return the canceled table ids.

1. Send cancel message to stream jobs (via cancel_jobs).
2. Send cancel message to recovered stream jobs (via barrier_scheduler).

Cleanup of their state will be cleaned up after the CancelStreamJob command succeeds, by the barrier manager for both of them.

pub(crate) async fn alter_table_parallelism( &self, table_id: u32, parallelism: TableParallelism, deferred: bool, ) -> MetaResult<()>

pub async fn create_subscription( self: &Arc<Self>, subscription: &Subscription, ) -> MetaResult<()>

pub async fn drop_subscription( self: &Arc<Self>, database_id: DatabaseId, subscription_id: u32, table_id: u32, )