Struct MetadataManager

pub struct MetadataManager {
    pub cluster_controller: ClusterControllerRef,
    pub catalog_controller: CatalogControllerRef,
}

Fields

cluster_controller: ClusterControllerRef

catalog_controller: CatalogControllerRef

Implementations

impl MetadataManager

pub fn new( cluster_controller: ClusterControllerRef, catalog_controller: CatalogControllerRef, ) -> Self

pub async fn get_worker_by_id( &self, worker_id: WorkerId, ) -> MetaResult<Option<PbWorkerNode>>

pub async fn count_worker_node(&self) -> MetaResult<HashMap<WorkerType, u64>>

pub async fn get_worker_info_by_id( &self, worker_id: WorkerId, ) -> Option<WorkerExtraInfo>

pub async fn add_worker_node( &self, type: PbWorkerType, host_address: HostAddress, property: AddNodeProperty, resource: PbResource, ) -> MetaResult<WorkerId>

pub async fn list_worker_node( &self, worker_type: Option<WorkerType>, worker_state: Option<State>, ) -> MetaResult<Vec<PbWorkerNode>>

pub async fn subscribe_active_streaming_compute_nodes( &self, ) -> MetaResult<(Vec<WorkerNode>, UnboundedReceiver<LocalNotification>)>

pub async fn list_active_streaming_compute_nodes( &self, ) -> MetaResult<Vec<PbWorkerNode>>

pub async fn list_active_serving_compute_nodes( &self, ) -> MetaResult<Vec<PbWorkerNode>>

pub async fn list_active_database_ids(&self) -> MetaResult<HashSet<DatabaseId>>

pub async fn split_fragment_map_by_database<T: Debug>( &self, fragment_map: HashMap<FragmentId, T>, ) -> MetaResult<HashMap<DatabaseId, HashMap<FragmentId, T>>>

pub async fn list_background_creating_jobs(&self) -> MetaResult<Vec<TableId>>

pub async fn list_sources(&self) -> MetaResult<Vec<PbSource>>

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

pub async fn running_fragment_parallelisms( &self, id_filter: Option<HashSet<FragmentId>>, ) -> MetaResult<HashMap<FragmentId, FragmentParallelismInfo>>

pub async fn get_upstream_root_fragments( &self, upstream_table_ids: &HashSet<TableId>, ) -> MetaResult<(HashMap<TableId, Fragment>, HashMap<ActorId, WorkerId>)>

Get and filter the “root” fragments of the specified relations. The root fragment is the bottom-most fragment of its fragment graph, and can be a MView or a Source.

What can be the root fragment

• For MV, it should have one MView fragment.
• For table, it should have one MView fragment and one or two Source fragments. MView should be the root.
• For source, it should have one Source fragment.

In other words, it’s the MView fragment if it exists, otherwise it’s the Source fragment.

What do we expect to get for different creating streaming job

• MV/Sink/Index should have MV upstream fragments for upstream MV/Tables, and Source upstream fragments for upstream shared sources.
• CDC Table has a Source upstream fragment.
• Sources and other Tables shouldn’t have an upstream fragment.

pub async fn get_streaming_cluster_info( &self, ) -> MetaResult<StreamingClusterInfo>

pub async fn get_all_table_options( &self, ) -> MetaResult<HashMap<u32, TableOption>>

pub async fn get_table_name_type_mapping( &self, ) -> MetaResult<HashMap<u32, (String, String)>>

pub async fn get_created_table_ids(&self) -> MetaResult<Vec<u32>>

pub async fn get_table_catalog_by_ids( &self, ids: Vec<u32>, ) -> MetaResult<Vec<PbTable>>

pub async fn get_sink_catalog_by_ids( &self, ids: &[u32], ) -> MetaResult<Vec<PbSink>>

pub async fn get_table_catalog_by_cdc_table_id( &self, cdc_table_id: &String, ) -> MetaResult<Vec<PbTable>>

pub async fn get_downstream_chain_fragments( &self, job_id: u32, ) -> MetaResult<(Vec<(PbDispatchStrategy, PbFragment)>, HashMap<ActorId, WorkerId>)>

pub async fn get_worker_actor_ids( &self, job_ids: HashSet<TableId>, ) -> MetaResult<BTreeMap<WorkerId, Vec<ActorId>>>

pub async fn get_job_id_to_internal_table_ids_mapping( &self, ) -> Option<Vec<(u32, Vec<u32>)>>

pub async fn get_job_fragments_by_id( &self, id: &TableId, ) -> MetaResult<TableFragments>

pub async fn get_running_actors_of_fragment( &self, id: FragmentId, ) -> MetaResult<HashSet<ActorId>>

pub async fn get_running_actors_and_upstream_actors_of_fragment( &self, id: FragmentId, ) -> MetaResult<HashSet<(ActorId, Vec<ActorId>)>>

pub async fn get_job_fragments_by_ids( &self, ids: &[TableId], ) -> MetaResult<Vec<TableFragments>>

pub async fn all_active_actors( &self, ) -> MetaResult<HashMap<ActorId, StreamActor>>

pub async fn worker_actor_count(&self) -> MetaResult<HashMap<WorkerId, usize>>

pub async fn count_streaming_job(&self) -> MetaResult<usize>

pub async fn list_stream_job_desc( &self, ) -> MetaResult<Vec<MetaTelemetryJobDesc>>

pub async fn update_source_rate_limit_by_source_id( &self, source_id: SourceId, rate_limit: Option<u32>, ) -> MetaResult<HashMap<FragmentId, Vec<ActorId>>>

pub async fn update_backfill_rate_limit_by_table_id( &self, table_id: TableId, rate_limit: Option<u32>, ) -> MetaResult<HashMap<FragmentId, Vec<ActorId>>>

pub async fn update_actor_splits_by_split_assignment( &self, split_assignment: &SplitAssignment, ) -> MetaResult<()>

pub async fn get_mv_depended_subscriptions( &self, ) -> MetaResult<HashMap<DatabaseId, HashMap<TableId, HashMap<SubscriptionId, u64>>>>

pub async fn get_job_max_parallelism( &self, table_id: TableId, ) -> MetaResult<usize>

pub fn cluster_id(&self) -> &ClusterId

pub async fn list_rate_limits(&self) -> MetaResult<Vec<RateLimitInfo>>

impl MetadataManager

pub(crate) async fn wait_streaming_job_finished( &self, id: ObjectId, ) -> MetaResult<NotificationVersion>

Wait for job finishing notification in TrackingJob::finish. The progress is updated per barrier.

pub(crate) async fn notify_finish_failed(&self, err: &MetaError)

Trait Implementations

impl Clone for MetadataManager

fn clone(&self) -> MetadataManager

Returns a copy of the value.

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

Performs copy-assignment from source.