risingwave_frontend/
meta_client.rs

1// Copyright 2022 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{BTreeMap, HashMap, HashSet};
16
17use anyhow::Context;
18use risingwave_common::id::{ConnectionId, JobId, SourceId, TableId, WorkerId};
19use risingwave_common::session_config::SessionConfig;
20use risingwave_common::system_param::reader::SystemParamsReader;
21use risingwave_common::util::cluster_limit::ClusterLimit;
22use risingwave_hummock_sdk::change_log::TableChangeLogs;
23use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
24use risingwave_hummock_sdk::{CompactionGroupId, HummockVersionId};
25use risingwave_pb::backup_service::{BackupJobStatus, MetaSnapshotMetadata};
26use risingwave_pb::catalog::Table;
27use risingwave_pb::common::WorkerNode;
28use risingwave_pb::ddl_service::DdlProgress;
29use risingwave_pb::hummock::rise_ctl_update_compaction_config_request::mutable_config::MutableConfig as PbMutableConfig;
30use risingwave_pb::hummock::write_limits::WriteLimit;
31use risingwave_pb::hummock::{
32    BranchedObject, CompactTaskAssignment, CompactTaskProgress, CompactionGroupInfo,
33};
34use risingwave_pb::id::{ActorId, IcebergCompactionTaskId};
35use risingwave_pb::meta::cancel_creating_jobs_request::PbJobs;
36use risingwave_pb::meta::list_actor_splits_response::ActorSplit;
37use risingwave_pb::meta::list_actor_states_response::ActorState;
38use risingwave_pb::meta::list_cdc_progress_response::PbCdcProgress;
39use risingwave_pb::meta::list_iceberg_compaction_status_response::IcebergCompactionStatus;
40use risingwave_pb::meta::list_iceberg_tables_response::IcebergTable;
41use risingwave_pb::meta::list_rate_limits_response::RateLimitInfo;
42use risingwave_pb::meta::list_refresh_table_states_response::RefreshTableState;
43use risingwave_pb::meta::list_streaming_job_states_response::StreamingJobState;
44use risingwave_pb::meta::list_table_fragments_response::TableFragmentInfo;
45use risingwave_pb::meta::{
46    EventLog, FragmentDistribution, PbTableParallelism, PbThrottleTarget, RecoveryStatus,
47    RefreshRequest, RefreshResponse, list_sink_log_store_tables_response,
48};
49use risingwave_pb::secret::PbSecretRef;
50use risingwave_rpc_client::error::Result;
51use risingwave_rpc_client::{HummockMetaClient, MetaClient};
52
53use crate::catalog::{DatabaseId, FragmentId, SinkId};
54
55/// A wrapper around the `MetaClient` that only provides a minor set of meta rpc.
56/// Most of the rpc to meta are delegated by other separate structs like `CatalogWriter`,
57/// `WorkerNodeManager`, etc. So frontend rarely needs to call `MetaClient` directly.
58/// Hence instead of to mock all rpc of `MetaClient` in tests, we aggregate those "direct" rpc
59/// in this trait so that the mocking can be simplified.
60#[async_trait::async_trait]
61pub trait FrontendMetaClient: Send + Sync {
62    async fn try_unregister(&self);
63
64    async fn flush(&self, database_id: DatabaseId) -> Result<HummockVersionId>;
65
66    async fn backup_meta(&self, remarks: Option<String>) -> Result<u64>;
67    async fn get_backup_job_status(&self, job_id: u64) -> Result<(BackupJobStatus, String)>;
68    async fn delete_meta_snapshot(&self, snapshot_ids: &[u64]) -> Result<()>;
69
70    async fn recover(&self) -> Result<()>;
71
72    async fn cancel_creating_jobs(&self, jobs: PbJobs) -> Result<Vec<u32>>;
73
74    async fn list_table_fragments(
75        &self,
76        table_ids: &[JobId],
77    ) -> Result<HashMap<JobId, TableFragmentInfo>>;
78
79    async fn list_streaming_job_states(&self) -> Result<Vec<StreamingJobState>>;
80
81    async fn list_fragment_distribution(
82        &self,
83        include_node: bool,
84    ) -> Result<Vec<FragmentDistribution>>;
85
86    async fn list_creating_fragment_distribution(&self) -> Result<Vec<FragmentDistribution>>;
87
88    async fn list_actor_states(&self) -> Result<Vec<ActorState>>;
89
90    async fn list_actor_splits(&self) -> Result<Vec<ActorSplit>>;
91
92    async fn list_meta_snapshots(&self) -> Result<Vec<MetaSnapshotMetadata>>;
93
94    async fn list_sink_log_store_tables(
95        &self,
96    ) -> Result<Vec<list_sink_log_store_tables_response::SinkLogStoreTable>>;
97
98    async fn set_system_param(
99        &self,
100        param: String,
101        value: Option<String>,
102    ) -> Result<Option<SystemParamsReader>>;
103
104    async fn get_session_params(&self) -> Result<SessionConfig>;
105
106    async fn set_session_param(&self, param: String, value: Option<String>) -> Result<String>;
107
108    async fn get_ddl_progress(&self) -> Result<Vec<DdlProgress>>;
109
110    async fn get_tables(
111        &self,
112        table_ids: Vec<TableId>,
113        include_dropped_table: bool,
114    ) -> Result<HashMap<TableId, Table>>;
115
116    /// Returns vector of (`worker_id`, `min_pinned_version_id`)
117    async fn list_hummock_pinned_versions(&self) -> Result<Vec<(WorkerId, HummockVersionId)>>;
118
119    async fn get_hummock_current_version(&self) -> Result<HummockVersion>;
120
121    async fn get_hummock_table_change_log(
122        &self,
123        start_epoch_inclusive: Option<u64>,
124        end_epoch_inclusive: Option<u64>,
125        table_ids: Option<HashSet<TableId>>,
126        exclude_empty: bool,
127        limit: Option<u32>,
128    ) -> Result<TableChangeLogs>;
129
130    async fn get_hummock_checkpoint_version(&self) -> Result<HummockVersion>;
131
132    async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>>;
133
134    async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>>;
135
136    async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>>;
137
138    async fn list_hummock_active_write_limits(
139        &self,
140    ) -> Result<HashMap<CompactionGroupId, WriteLimit>>;
141
142    async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>>;
143
144    async fn list_event_log(&self) -> Result<Vec<EventLog>>;
145    async fn list_compact_task_assignment(&self) -> Result<Vec<CompactTaskAssignment>>;
146
147    async fn list_all_nodes(&self) -> Result<Vec<WorkerNode>>;
148
149    async fn list_compact_task_progress(&self) -> Result<Vec<CompactTaskProgress>>;
150
151    async fn apply_throttle(
152        &self,
153        throttle_target: PbThrottleTarget,
154        throttle_type: risingwave_pb::common::PbThrottleType,
155        id: u32,
156        rate_limit: Option<u32>,
157    ) -> Result<()>;
158
159    async fn alter_fragment_parallelism(
160        &self,
161        fragment_ids: Vec<FragmentId>,
162        parallelism: Option<PbTableParallelism>,
163    ) -> Result<()>;
164
165    async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus>;
166
167    async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>>;
168
169    async fn list_rate_limits(&self) -> Result<Vec<RateLimitInfo>>;
170
171    async fn list_cdc_progress(&self) -> Result<HashMap<JobId, PbCdcProgress>>;
172
173    async fn list_refresh_table_states(&self) -> Result<Vec<RefreshTableState>>;
174
175    async fn list_iceberg_compaction_status(&self) -> Result<Vec<IcebergCompactionStatus>>;
176
177    async fn get_meta_store_endpoint(&self) -> Result<String>;
178
179    async fn alter_sink_props(
180        &self,
181        sink_id: SinkId,
182        changed_props: BTreeMap<String, String>,
183        changed_secret_refs: BTreeMap<String, PbSecretRef>,
184        connector_conn_ref: Option<ConnectionId>,
185    ) -> Result<()>;
186
187    async fn alter_iceberg_table_props(
188        &self,
189        table_id: TableId,
190        sink_id: SinkId,
191        source_id: SourceId,
192        changed_props: BTreeMap<String, String>,
193        changed_secret_refs: BTreeMap<String, PbSecretRef>,
194        connector_conn_ref: Option<ConnectionId>,
195    ) -> Result<()>;
196
197    async fn alter_source_connector_props(
198        &self,
199        source_id: SourceId,
200        changed_props: BTreeMap<String, String>,
201        changed_secret_refs: BTreeMap<String, PbSecretRef>,
202        connector_conn_ref: Option<ConnectionId>,
203    ) -> Result<()>;
204
205    async fn alter_connection_connector_props(
206        &self,
207        connection_id: u32,
208        changed_props: BTreeMap<String, String>,
209        changed_secret_refs: BTreeMap<String, PbSecretRef>,
210    ) -> Result<()>;
211
212    async fn list_hosted_iceberg_tables(&self) -> Result<Vec<IcebergTable>>;
213
214    async fn get_fragment_by_id(
215        &self,
216        fragment_id: FragmentId,
217    ) -> Result<Option<FragmentDistribution>>;
218
219    async fn get_fragment_vnodes(
220        &self,
221        fragment_id: FragmentId,
222    ) -> Result<Vec<(ActorId, Vec<u32>)>>;
223
224    async fn get_actor_vnodes(&self, actor_id: ActorId) -> Result<Vec<u32>>;
225
226    fn worker_id(&self) -> WorkerId;
227
228    async fn set_sync_log_store_aligned(&self, job_id: JobId, aligned: bool) -> Result<()>;
229
230    async fn compact_iceberg_table(&self, sink_id: SinkId) -> Result<IcebergCompactionTaskId>;
231
232    async fn rewrite_iceberg_table_manifests(&self, sink_id: SinkId) -> Result<()>;
233
234    async fn expire_iceberg_table_snapshots(&self, sink_id: SinkId) -> Result<()>;
235
236    async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse>;
237
238    fn cluster_id(&self) -> &str;
239
240    async fn list_unmigrated_tables(&self) -> Result<HashMap<TableId, String>>;
241
242    async fn update_compaction_config(
243        &self,
244        compaction_group_ids: Vec<CompactionGroupId>,
245        configs: Vec<PbMutableConfig>,
246    ) -> Result<()>;
247}
248
249pub struct FrontendMetaClientImpl(pub MetaClient);
250
251#[async_trait::async_trait]
252impl FrontendMetaClient for FrontendMetaClientImpl {
253    async fn try_unregister(&self) {
254        self.0.try_unregister().await;
255    }
256
257    async fn flush(&self, database_id: DatabaseId) -> Result<HummockVersionId> {
258        self.0.flush(database_id).await
259    }
260
261    async fn backup_meta(&self, remarks: Option<String>) -> Result<u64> {
262        self.0.backup_meta(remarks).await
263    }
264
265    async fn get_backup_job_status(&self, job_id: u64) -> Result<(BackupJobStatus, String)> {
266        self.0.get_backup_job_status(job_id).await
267    }
268
269    async fn delete_meta_snapshot(&self, snapshot_ids: &[u64]) -> Result<()> {
270        self.0.delete_meta_snapshot(snapshot_ids).await
271    }
272
273    async fn recover(&self) -> Result<()> {
274        self.0.recover().await
275    }
276
277    async fn cancel_creating_jobs(&self, infos: PbJobs) -> Result<Vec<u32>> {
278        self.0.cancel_creating_jobs(infos).await
279    }
280
281    async fn list_table_fragments(
282        &self,
283        job_ids: &[JobId],
284    ) -> Result<HashMap<JobId, TableFragmentInfo>> {
285        self.0.list_table_fragments(job_ids).await
286    }
287
288    async fn list_streaming_job_states(&self) -> Result<Vec<StreamingJobState>> {
289        self.0.list_streaming_job_states().await
290    }
291
292    async fn list_fragment_distribution(
293        &self,
294        include_node: bool,
295    ) -> Result<Vec<FragmentDistribution>> {
296        self.0.list_fragment_distributions(include_node).await
297    }
298
299    async fn list_creating_fragment_distribution(&self) -> Result<Vec<FragmentDistribution>> {
300        self.0.list_creating_fragment_distribution().await
301    }
302
303    async fn list_actor_states(&self) -> Result<Vec<ActorState>> {
304        self.0.list_actor_states().await
305    }
306
307    async fn list_actor_splits(&self) -> Result<Vec<ActorSplit>> {
308        self.0.list_actor_splits().await
309    }
310
311    async fn list_meta_snapshots(&self) -> Result<Vec<MetaSnapshotMetadata>> {
312        let manifest = self.0.get_meta_snapshot_manifest().await?;
313        Ok(manifest.snapshot_metadata)
314    }
315
316    async fn list_sink_log_store_tables(
317        &self,
318    ) -> Result<Vec<list_sink_log_store_tables_response::SinkLogStoreTable>> {
319        self.0.list_sink_log_store_tables().await
320    }
321
322    async fn set_system_param(
323        &self,
324        param: String,
325        value: Option<String>,
326    ) -> Result<Option<SystemParamsReader>> {
327        self.0.set_system_param(param, value).await
328    }
329
330    async fn get_session_params(&self) -> Result<SessionConfig> {
331        let session_config: SessionConfig =
332            serde_json::from_str(&self.0.get_session_params().await?)
333                .context("failed to parse session config")?;
334        Ok(session_config)
335    }
336
337    async fn set_session_param(&self, param: String, value: Option<String>) -> Result<String> {
338        self.0.set_session_param(param, value).await
339    }
340
341    async fn get_ddl_progress(&self) -> Result<Vec<DdlProgress>> {
342        let ddl_progress = self.0.get_ddl_progress().await?;
343        Ok(ddl_progress)
344    }
345
346    async fn get_tables(
347        &self,
348        table_ids: Vec<TableId>,
349        include_dropped_tables: bool,
350    ) -> Result<HashMap<TableId, Table>> {
351        let tables = self.0.get_tables(table_ids, include_dropped_tables).await?;
352        Ok(tables)
353    }
354
355    async fn list_hummock_pinned_versions(&self) -> Result<Vec<(WorkerId, HummockVersionId)>> {
356        let pinned_versions = self
357            .0
358            .risectl_get_pinned_versions_summary()
359            .await?
360            .summary
361            .unwrap()
362            .pinned_versions;
363        let ret = pinned_versions
364            .into_iter()
365            .map(|v| (v.context_id, v.min_pinned_id))
366            .collect();
367        Ok(ret)
368    }
369
370    async fn get_hummock_current_version(&self) -> Result<HummockVersion> {
371        self.0.get_current_version().await
372    }
373
374    async fn get_hummock_table_change_log(
375        &self,
376        start_epoch_inclusive: Option<u64>,
377        end_epoch_inclusive: Option<u64>,
378        table_ids: Option<HashSet<TableId>>,
379        exclude_empty: bool,
380        limit: Option<u32>,
381    ) -> Result<TableChangeLogs> {
382        self.0
383            .get_table_change_logs(
384                true,
385                start_epoch_inclusive,
386                end_epoch_inclusive,
387                table_ids,
388                exclude_empty,
389                limit,
390            )
391            .await
392    }
393
394    async fn get_hummock_checkpoint_version(&self) -> Result<HummockVersion> {
395        self.0
396            .risectl_get_checkpoint_hummock_version()
397            .await
398            .map(|v| HummockVersion::from_rpc_protobuf(&v.checkpoint_version.unwrap()))
399    }
400
401    async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>> {
402        // FIXME #8612: there can be lots of version deltas, so better to fetch them by pages and refactor `SysRowSeqScanExecutor` to yield multiple chunks.
403        self.0
404            .list_version_deltas(HummockVersionId::new(0), u32::MAX, u64::MAX)
405            .await
406    }
407
408    async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>> {
409        self.0.list_branched_object().await
410    }
411
412    async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>> {
413        self.0.risectl_list_compaction_group().await
414    }
415
416    async fn list_hummock_active_write_limits(
417        &self,
418    ) -> Result<HashMap<CompactionGroupId, WriteLimit>> {
419        self.0.list_active_write_limit().await
420    }
421
422    async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>> {
423        self.0.list_hummock_meta_config().await
424    }
425
426    async fn list_event_log(&self) -> Result<Vec<EventLog>> {
427        self.0.list_event_log().await
428    }
429
430    async fn list_compact_task_assignment(&self) -> Result<Vec<CompactTaskAssignment>> {
431        self.0.list_compact_task_assignment().await
432    }
433
434    async fn list_all_nodes(&self) -> Result<Vec<WorkerNode>> {
435        self.0.list_worker_nodes(None).await
436    }
437
438    async fn list_compact_task_progress(&self) -> Result<Vec<CompactTaskProgress>> {
439        self.0.list_compact_task_progress().await
440    }
441
442    async fn apply_throttle(
443        &self,
444        throttle_target: PbThrottleTarget,
445        throttle_type: risingwave_pb::common::PbThrottleType,
446        id: u32,
447        rate_limit: Option<u32>,
448    ) -> Result<()> {
449        self.0
450            .apply_throttle(throttle_target, throttle_type, id, rate_limit)
451            .await
452            .map(|_| ())
453    }
454
455    async fn alter_fragment_parallelism(
456        &self,
457        fragment_ids: Vec<FragmentId>,
458        parallelism: Option<PbTableParallelism>,
459    ) -> Result<()> {
460        self.0
461            .alter_fragment_parallelism(fragment_ids, parallelism)
462            .await
463    }
464
465    async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus> {
466        self.0.get_cluster_recovery_status().await
467    }
468
469    async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>> {
470        self.0.get_cluster_limits().await
471    }
472
473    async fn list_rate_limits(&self) -> Result<Vec<RateLimitInfo>> {
474        self.0.list_rate_limits().await
475    }
476
477    async fn list_cdc_progress(&self) -> Result<HashMap<JobId, PbCdcProgress>> {
478        self.0.list_cdc_progress().await
479    }
480
481    async fn get_meta_store_endpoint(&self) -> Result<String> {
482        self.0.get_meta_store_endpoint().await
483    }
484
485    async fn alter_sink_props(
486        &self,
487        sink_id: SinkId,
488        changed_props: BTreeMap<String, String>,
489        changed_secret_refs: BTreeMap<String, PbSecretRef>,
490        connector_conn_ref: Option<ConnectionId>,
491    ) -> Result<()> {
492        self.0
493            .alter_sink_props(
494                sink_id,
495                changed_props,
496                changed_secret_refs,
497                connector_conn_ref,
498            )
499            .await
500    }
501
502    async fn alter_iceberg_table_props(
503        &self,
504        table_id: TableId,
505        sink_id: SinkId,
506        source_id: SourceId,
507        changed_props: BTreeMap<String, String>,
508        changed_secret_refs: BTreeMap<String, PbSecretRef>,
509        connector_conn_ref: Option<ConnectionId>,
510    ) -> Result<()> {
511        self.0
512            .alter_iceberg_table_props(
513                table_id,
514                sink_id,
515                source_id,
516                changed_props,
517                changed_secret_refs,
518                connector_conn_ref,
519            )
520            .await
521    }
522
523    async fn alter_source_connector_props(
524        &self,
525        source_id: SourceId,
526        changed_props: BTreeMap<String, String>,
527        changed_secret_refs: BTreeMap<String, PbSecretRef>,
528        connector_conn_ref: Option<ConnectionId>,
529    ) -> Result<()> {
530        self.0
531            .alter_source_connector_props(
532                source_id,
533                changed_props,
534                changed_secret_refs,
535                connector_conn_ref,
536            )
537            .await
538    }
539
540    async fn alter_connection_connector_props(
541        &self,
542        connection_id: u32,
543        changed_props: BTreeMap<String, String>,
544        changed_secret_refs: BTreeMap<String, PbSecretRef>,
545    ) -> Result<()> {
546        self.0
547            .alter_connection_connector_props(connection_id, changed_props, changed_secret_refs)
548            .await
549    }
550
551    async fn list_hosted_iceberg_tables(&self) -> Result<Vec<IcebergTable>> {
552        self.0.list_hosted_iceberg_tables().await
553    }
554
555    async fn get_fragment_by_id(
556        &self,
557        fragment_id: FragmentId,
558    ) -> Result<Option<FragmentDistribution>> {
559        self.0.get_fragment_by_id(fragment_id).await
560    }
561
562    async fn get_fragment_vnodes(
563        &self,
564        fragment_id: FragmentId,
565    ) -> Result<Vec<(ActorId, Vec<u32>)>> {
566        self.0.get_fragment_vnodes(fragment_id).await
567    }
568
569    async fn get_actor_vnodes(&self, actor_id: ActorId) -> Result<Vec<u32>> {
570        self.0.get_actor_vnodes(actor_id).await
571    }
572
573    fn worker_id(&self) -> WorkerId {
574        self.0.worker_id()
575    }
576
577    async fn set_sync_log_store_aligned(&self, job_id: JobId, aligned: bool) -> Result<()> {
578        self.0.set_sync_log_store_aligned(job_id, aligned).await
579    }
580
581    async fn compact_iceberg_table(&self, sink_id: SinkId) -> Result<IcebergCompactionTaskId> {
582        self.0.compact_iceberg_table(sink_id).await
583    }
584
585    async fn rewrite_iceberg_table_manifests(&self, sink_id: SinkId) -> Result<()> {
586        self.0.rewrite_iceberg_table_manifests(sink_id).await
587    }
588
589    async fn expire_iceberg_table_snapshots(&self, sink_id: SinkId) -> Result<()> {
590        self.0.expire_iceberg_table_snapshots(sink_id).await
591    }
592
593    async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse> {
594        self.0.refresh(request).await
595    }
596
597    fn cluster_id(&self) -> &str {
598        self.0.cluster_id()
599    }
600
601    async fn list_unmigrated_tables(&self) -> Result<HashMap<TableId, String>> {
602        self.0.list_unmigrated_tables().await
603    }
604
605    async fn list_refresh_table_states(&self) -> Result<Vec<RefreshTableState>> {
606        self.0.list_refresh_table_states().await
607    }
608
609    async fn list_iceberg_compaction_status(&self) -> Result<Vec<IcebergCompactionStatus>> {
610        self.0.list_iceberg_compaction_status().await
611    }
612
613    async fn update_compaction_config(
614        &self,
615        compaction_group_ids: Vec<CompactionGroupId>,
616        configs: Vec<PbMutableConfig>,
617    ) -> Result<()> {
618        self.0
619            .risectl_update_compaction_config(&compaction_group_ids, &configs)
620            .await
621    }
622}
risingwave_frontend/meta_client.rs

risingwave_frontend/
meta_client.rs
