Struct MetaClient

pub struct MetaClient {
    worker_id: u32,
    worker_type: WorkerType,
    host_addr: HostAddr,
    inner: GrpcMetaClient,
    meta_config: MetaConfig,
    cluster_id: String,
    shutting_down: Arc<AtomicBool>,
}

Client to meta server. Cloning the instance is lightweight.

Fields

worker_id: u32

worker_type: WorkerType

host_addr: HostAddr

inner: GrpcMetaClient

meta_config: MetaConfig

cluster_id: String

shutting_down: Arc<AtomicBool>

Implementations

impl MetaClient

pub fn worker_id(&self) -> u32

pub fn host_addr(&self) -> &HostAddr

pub fn worker_type(&self) -> WorkerType

pub fn cluster_id(&self) -> &str

pub async fn subscribe( &self, subscribe_type: SubscribeType, ) -> Result<Streaming<SubscribeResponse>>

Subscribe to notification from meta.

pub async fn create_connection( &self, connection_name: String, database_id: u32, schema_id: u32, owner_id: u32, req: Payload, ) -> Result<WaitVersion>

pub async fn create_secret( &self, secret_name: String, database_id: u32, schema_id: u32, owner_id: u32, value: Vec<u8>, ) -> Result<WaitVersion>

pub async fn list_connections( &self, _name: Option<&str>, ) -> Result<Vec<Connection>>

pub async fn drop_connection(&self, connection_id: u32) -> Result<WaitVersion>

pub async fn drop_secret(&self, secret_id: SecretId) -> Result<WaitVersion>

pub async fn register_new( addr_strategy: MetaAddressStrategy, worker_type: WorkerType, addr: &HostAddr, property: Property, meta_config: &MetaConfig, ) -> (Self, SystemParamsReader)

Register the current node to the cluster and set the corresponding worker id.

Retry if there’s connection issue with the meta node. Exit the process if the registration fails.

async fn register_new_inner( addr_strategy: MetaAddressStrategy, worker_type: WorkerType, addr: &HostAddr, property: Property, meta_config: &MetaConfig, ) -> Result<(Self, SystemParamsReader)>

pub async fn activate(&self, addr: &HostAddr) -> Result<()>

Activate the current node in cluster to confirm it’s ready to serve.

pub async fn send_heartbeat(&self, node_id: u32) -> Result<()>

Send heartbeat signal to meta service.

pub async fn create_database(&self, db: PbDatabase) -> Result<WaitVersion>

pub async fn create_schema(&self, schema: PbSchema) -> Result<WaitVersion>

pub async fn create_materialized_view( &self, table: PbTable, graph: StreamFragmentGraph, dependencies: HashSet<ObjectId>, specific_resource_group: Option<String>, ) -> Result<WaitVersion>

pub async fn drop_materialized_view( &self, table_id: TableId, cascade: bool, ) -> Result<WaitVersion>

pub async fn create_source( &self, source: PbSource, graph: Option<StreamFragmentGraph>, ) -> Result<WaitVersion>

pub async fn create_sink( &self, sink: PbSink, graph: StreamFragmentGraph, affected_table_change: Option<ReplaceJobPlan>, dependencies: HashSet<ObjectId>, ) -> Result<WaitVersion>

pub async fn create_subscription( &self, subscription: PbSubscription, ) -> Result<WaitVersion>

pub async fn create_function(&self, function: PbFunction) -> Result<WaitVersion>

pub async fn create_table( &self, source: Option<PbSource>, table: PbTable, graph: StreamFragmentGraph, job_type: PbTableJobType, ) -> Result<WaitVersion>

pub async fn comment_on(&self, comment: PbComment) -> Result<WaitVersion>

pub async fn alter_name( &self, object: Object, name: &str, ) -> Result<WaitVersion>

pub async fn alter_owner( &self, object: Object, owner_id: u32, ) -> Result<WaitVersion>

pub async fn alter_set_schema( &self, object: Object, new_schema_id: u32, ) -> Result<WaitVersion>

pub async fn alter_source(&self, source: PbSource) -> Result<WaitVersion>

pub async fn alter_parallelism( &self, table_id: u32, parallelism: PbTableParallelism, deferred: bool, ) -> Result<()>

pub async fn alter_resource_group( &self, table_id: u32, resource_group: Option<String>, deferred: bool, ) -> Result<()>

pub async fn alter_swap_rename(&self, object: Object) -> Result<WaitVersion>

pub async fn alter_secret( &self, secret_id: u32, secret_name: String, database_id: u32, schema_id: u32, owner_id: u32, value: Vec<u8>, ) -> Result<WaitVersion>

pub async fn replace_job( &self, graph: StreamFragmentGraph, table_col_index_mapping: ColIndexMapping, replace_job: ReplaceJob, ) -> Result<WaitVersion>

pub async fn auto_schema_change( &self, schema_change: SchemaChangeEnvelope, ) -> Result<()>

pub async fn create_view(&self, view: PbView) -> Result<WaitVersion>

pub async fn create_index( &self, index: PbIndex, table: PbTable, graph: StreamFragmentGraph, ) -> Result<WaitVersion>

pub async fn drop_table( &self, source_id: Option<u32>, table_id: TableId, cascade: bool, ) -> Result<WaitVersion>

pub async fn drop_view( &self, view_id: u32, cascade: bool, ) -> Result<WaitVersion>

pub async fn drop_source( &self, source_id: u32, cascade: bool, ) -> Result<WaitVersion>

pub async fn drop_sink( &self, sink_id: u32, cascade: bool, affected_table_change: Option<ReplaceJobPlan>, ) -> Result<WaitVersion>

pub async fn drop_subscription( &self, subscription_id: u32, cascade: bool, ) -> Result<WaitVersion>

pub async fn drop_index( &self, index_id: IndexId, cascade: bool, ) -> Result<WaitVersion>

pub async fn drop_function( &self, function_id: FunctionId, ) -> Result<WaitVersion>

pub async fn drop_database(&self, database_id: u32) -> Result<WaitVersion>

pub async fn drop_schema( &self, schema_id: u32, cascade: bool, ) -> Result<WaitVersion>

pub async fn create_user(&self, user: UserInfo) -> Result<u64>

pub async fn drop_user(&self, user_id: u32) -> Result<u64>

pub async fn update_user( &self, user: UserInfo, update_fields: Vec<UpdateField>, ) -> Result<u64>

pub async fn grant_privilege( &self, user_ids: Vec<u32>, privileges: Vec<GrantPrivilege>, with_grant_option: bool, granted_by: u32, ) -> Result<u64>

pub async fn revoke_privilege( &self, user_ids: Vec<u32>, privileges: Vec<GrantPrivilege>, granted_by: u32, revoke_by: u32, revoke_grant_option: bool, cascade: bool, ) -> Result<u64>

pub async fn unregister(&self) -> Result<()>

Unregister the current node from the cluster.

pub async fn try_unregister(&self)

Try to unregister the current worker from the cluster with best effort. Log the result.

pub async fn update_schedulability( &self, worker_ids: &[u32], schedulability: Schedulability, ) -> Result<UpdateWorkerNodeSchedulabilityResponse>

pub async fn list_worker_nodes( &self, worker_type: Option<WorkerType>, ) -> Result<Vec<WorkerNode>>

pub fn start_heartbeat_loop( meta_client: MetaClient, min_interval: Duration, ) -> (JoinHandle<()>, Sender<()>)

Starts a heartbeat worker.

pub async fn risectl_list_state_tables(&self) -> Result<Vec<PbTable>>

pub async fn flush(&self, database_id: u32) -> Result<HummockVersionId>

pub async fn wait(&self) -> Result<()>

pub async fn recover(&self) -> Result<()>

pub async fn cancel_creating_jobs(&self, jobs: PbJobs) -> Result<Vec<u32>>

pub async fn list_table_fragments( &self, table_ids: &[u32], ) -> Result<HashMap<u32, TableFragmentInfo>>

pub async fn list_streaming_job_states(&self) -> Result<Vec<StreamingJobState>>

pub async fn list_fragment_distributions( &self, ) -> Result<Vec<FragmentDistribution>>

pub async fn get_fragment_by_id( &self, fragment_id: u32, ) -> Result<Option<FragmentDistribution>>

pub async fn list_actor_states(&self) -> Result<Vec<ActorState>>

pub async fn list_actor_splits(&self) -> Result<Vec<ActorSplit>>

pub async fn list_object_dependencies( &self, ) -> Result<Vec<PbObjectDependencies>>

pub async fn pause(&self) -> Result<PauseResponse>

pub async fn resume(&self) -> Result<ResumeResponse>

pub async fn apply_throttle( &self, kind: PbThrottleTarget, id: u32, rate: Option<u32>, ) -> Result<ApplyThrottleResponse>

pub async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus>

pub async fn get_cluster_info(&self) -> Result<GetClusterInfoResponse>

pub async fn reschedule( &self, worker_reschedules: HashMap<u32, PbWorkerReschedule>, revision: u64, resolve_no_shuffle_upstream: bool, ) -> Result<(bool, u64)>

pub async fn risectl_get_pinned_versions_summary( &self, ) -> Result<RiseCtlGetPinnedVersionsSummaryResponse>

pub async fn risectl_get_checkpoint_hummock_version( &self, ) -> Result<RiseCtlGetCheckpointVersionResponse>

pub async fn risectl_pause_hummock_version_checkpoint( &self, ) -> Result<RiseCtlPauseVersionCheckpointResponse>

pub async fn risectl_resume_hummock_version_checkpoint( &self, ) -> Result<RiseCtlResumeVersionCheckpointResponse>

pub async fn init_metadata_for_replay( &self, tables: Vec<PbTable>, compaction_groups: Vec<CompactionGroupInfo>, ) -> Result<()>

pub async fn replay_version_delta( &self, version_delta: HummockVersionDelta, ) -> Result<(HummockVersion, Vec<CompactionGroupId>)>

pub async fn list_version_deltas( &self, start_id: HummockVersionId, num_limit: u32, committed_epoch_limit: HummockEpoch, ) -> Result<Vec<HummockVersionDelta>>

pub async fn trigger_compaction_deterministic( &self, version_id: HummockVersionId, compaction_groups: Vec<CompactionGroupId>, ) -> Result<()>

pub async fn disable_commit_epoch(&self) -> Result<HummockVersion>

pub async fn get_assigned_compact_task_num(&self) -> Result<usize>

pub async fn risectl_list_compaction_group( &self, ) -> Result<Vec<CompactionGroupInfo>>

pub async fn risectl_update_compaction_config( &self, compaction_groups: &[CompactionGroupId], configs: &[MutableConfig], ) -> Result<()>

pub async fn backup_meta(&self, remarks: Option<String>) -> Result<u64>

pub async fn get_backup_job_status( &self, job_id: u64, ) -> Result<(BackupJobStatus, String)>

pub async fn delete_meta_snapshot(&self, snapshot_ids: &[u64]) -> Result<()>

pub async fn get_meta_snapshot_manifest(&self) -> Result<MetaSnapshotManifest>

pub async fn get_telemetry_info(&self) -> Result<TelemetryInfoResponse>

pub async fn get_system_params(&self) -> Result<SystemParamsReader>

pub async fn get_meta_store_endpoint(&self) -> Result<String>

pub async fn alter_sink_props( &self, sink_id: u32, changed_props: BTreeMap<String, String>, changed_secret_refs: BTreeMap<String, PbSecretRef>, connector_conn_ref: Option<u32>, ) -> Result<()>

pub async fn set_system_param( &self, param: String, value: Option<String>, ) -> Result<Option<SystemParamsReader>>

pub async fn get_session_params(&self) -> Result<String>

pub async fn set_session_param( &self, param: String, value: Option<String>, ) -> Result<String>

pub async fn get_ddl_progress(&self) -> Result<Vec<DdlProgress>>

pub async fn split_compaction_group( &self, group_id: CompactionGroupId, table_ids_to_new_group: &[StateTableId], partition_vnode_count: u32, ) -> Result<CompactionGroupId>

pub async fn get_tables( &self, table_ids: &[u32], include_dropped_tables: bool, ) -> Result<HashMap<u32, Table>>

pub async fn list_serving_vnode_mappings( &self, ) -> Result<HashMap<u32, (u32, WorkerSlotMapping)>>

pub async fn risectl_list_compaction_status( &self, ) -> Result<(Vec<CompactStatus>, Vec<CompactTaskAssignment>, Vec<CompactTaskProgress>)>

pub async fn get_compaction_score( &self, compaction_group_id: CompactionGroupId, ) -> Result<Vec<PickerInfo>>

pub async fn risectl_rebuild_table_stats(&self) -> Result<()>

pub async fn list_branched_object(&self) -> Result<Vec<BranchedObject>>

pub async fn list_active_write_limit(&self) -> Result<HashMap<u64, WriteLimit>>

pub async fn list_hummock_meta_config(&self) -> Result<HashMap<String, String>>

pub async fn delete_worker_node(&self, worker: HostAddress) -> Result<()>

pub async fn rw_cloud_validate_source( &self, source_type: SourceType, source_config: HashMap<String, String>, ) -> Result<RwCloudValidateSourceResponse>

pub async fn sink_coordinate_client(&self) -> SinkCoordinationRpcClient

pub async fn list_compact_task_assignment( &self, ) -> Result<Vec<CompactTaskAssignment>>

pub async fn list_event_log(&self) -> Result<Vec<EventLog>>

pub async fn list_compact_task_progress( &self, ) -> Result<Vec<CompactTaskProgress>>

pub fn try_add_panic_event_blocking( &self, panic_info: impl Display, timeout_millis: Option<u64>, )

If timeout_millis is None, default is used.

pub async fn add_sink_fail_evet( &self, sink_id: u32, sink_name: String, connector: String, error: String, ) -> Result<()>

pub async fn cancel_compact_task( &self, task_id: u64, task_status: TaskStatus, ) -> Result<bool>

pub async fn get_version_by_epoch( &self, epoch: HummockEpoch, table_id: u32, ) -> Result<PbHummockVersion>

pub async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>>

pub async fn merge_compaction_group( &self, left_group_id: CompactionGroupId, right_group_id: CompactionGroupId, ) -> Result<()>

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

List all rate limits for sources and backfills

pub async fn list_hosted_iceberg_tables(&self) -> Result<Vec<IcebergTable>>

Trait Implementations

impl Clone for MetaClient

fn clone(&self) -> MetaClient

Returns a copy of the value.

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

Performs copy-assignment from source.

impl Debug for MetaClient

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

Formats the value using the given formatter.

impl HummockMetaClient for MetaClient

fn unpin_version_before<'life0, 'async_trait>( &'life0 self, unpin_version_before: HummockVersionId, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get_current_version<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<HummockVersion>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get_new_object_ids<'life0, 'async_trait>( &'life0 self, number: u32, ) -> Pin<Box<dyn Future<Output = Result<ObjectIdRange>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn commit_epoch_with_change_log<'life0, 'async_trait>( &'life0 self, _epoch: HummockEpoch, _sync_result: SyncResult, _change_log_info: Option<HummockMetaClientChangeLogInfo>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn trigger_manual_compaction<'life0, 'async_trait>( &'life0 self, compaction_group_id: u64, table_id: u32, level: u32, sst_ids: Vec<u64>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn trigger_full_gc<'life0, 'async_trait>( &'life0 self, sst_retention_time_sec: u64, prefix: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn subscribe_compaction_event<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(UnboundedSender<SubscribeCompactionEventRequest>, BoxStream<'static, CompactionEventItem>)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn get_version_by_epoch<'life0, 'async_trait>( &'life0 self, epoch: HummockEpoch, table_id: u32, ) -> Pin<Box<dyn Future<Output = Result<PbHummockVersion>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn subscribe_iceberg_compaction_event<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<(UnboundedSender<SubscribeIcebergCompactionEventRequest>, BoxStream<'static, IcebergCompactionEventItem>)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

fn commit_epoch<'life0, 'async_trait>( &'life0 self, epoch: HummockEpoch, sync_result: SyncResult, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

impl TelemetryInfoFetcher for MetaClient

fn fetch_telemetry_info<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Option<String>, String>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Fetches telemetry info from meta. Currently it’s only tracking_id (cluster_id).