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 clear_file_cache(&self, clear_meta_cache: bool, clear_data_cache: bool) -> Result<()>;
105
106    async fn get_session_params(&self) -> Result<SessionConfig>;
107
108    async fn set_session_param(&self, param: String, value: Option<String>) -> Result<String>;
109
110    async fn get_ddl_progress(&self) -> Result<Vec<DdlProgress>>;
111
112    async fn get_tables(
113        &self,
114        table_ids: Vec<TableId>,
115        include_dropped_table: bool,
116    ) -> Result<HashMap<TableId, Table>>;
117
118    /// Returns vector of (`worker_id`, `min_pinned_version_id`)
119    async fn list_hummock_pinned_versions(&self) -> Result<Vec<(WorkerId, HummockVersionId)>>;
120
121    async fn get_hummock_current_version(&self) -> Result<HummockVersion>;
122
123    async fn get_hummock_table_change_log(
124        &self,
125        start_epoch_inclusive: Option<u64>,
126        end_epoch_inclusive: Option<u64>,
127        table_ids: Option<HashSet<TableId>>,
128        exclude_empty: bool,
129        limit: Option<u32>,
130    ) -> Result<TableChangeLogs>;
131
132    async fn get_hummock_checkpoint_version(&self) -> Result<HummockVersion>;
133
134    async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>>;
135
136    async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>>;
137
138    async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>>;
139
140    async fn list_hummock_active_write_limits(
141        &self,
142    ) -> Result<HashMap<CompactionGroupId, WriteLimit>>;
143
144    async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>>;
145
146    async fn list_event_log(&self) -> Result<Vec<EventLog>>;
147    async fn list_compact_task_assignment(&self) -> Result<Vec<CompactTaskAssignment>>;
148
149    async fn list_all_nodes(&self) -> Result<Vec<WorkerNode>>;
150
151    async fn list_compact_task_progress(&self) -> Result<Vec<CompactTaskProgress>>;
152
153    async fn apply_throttle(
154        &self,
155        throttle_target: PbThrottleTarget,
156        throttle_type: risingwave_pb::common::PbThrottleType,
157        id: u32,
158        rate_limit: Option<u32>,
159    ) -> Result<()>;
160
161    async fn alter_fragment_parallelism(
162        &self,
163        fragment_ids: Vec<FragmentId>,
164        parallelism: Option<PbTableParallelism>,
165    ) -> Result<()>;
166
167    async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus>;
168
169    async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>>;
170
171    async fn list_rate_limits(&self) -> Result<Vec<RateLimitInfo>>;
172
173    async fn list_cdc_progress(&self) -> Result<HashMap<JobId, PbCdcProgress>>;
174
175    async fn list_refresh_table_states(&self) -> Result<Vec<RefreshTableState>>;
176
177    async fn list_iceberg_compaction_status(&self) -> Result<Vec<IcebergCompactionStatus>>;
178
179    async fn get_meta_store_endpoint(&self) -> Result<String>;
180
181    async fn alter_sink_props(
182        &self,
183        sink_id: SinkId,
184        changed_props: BTreeMap<String, String>,
185        changed_secret_refs: BTreeMap<String, PbSecretRef>,
186        connector_conn_ref: Option<ConnectionId>,
187    ) -> Result<()>;
188
189    async fn alter_iceberg_table_props(
190        &self,
191        table_id: TableId,
192        sink_id: SinkId,
193        source_id: SourceId,
194        changed_props: BTreeMap<String, String>,
195        changed_secret_refs: BTreeMap<String, PbSecretRef>,
196        connector_conn_ref: Option<ConnectionId>,
197    ) -> Result<()>;
198
199    async fn alter_source_connector_props(
200        &self,
201        source_id: SourceId,
202        changed_props: BTreeMap<String, String>,
203        changed_secret_refs: BTreeMap<String, PbSecretRef>,
204        connector_conn_ref: Option<ConnectionId>,
205    ) -> Result<()>;
206
207    async fn alter_connection_connector_props(
208        &self,
209        connection_id: u32,
210        changed_props: BTreeMap<String, String>,
211        changed_secret_refs: BTreeMap<String, PbSecretRef>,
212    ) -> Result<()>;
213
214    async fn list_hosted_iceberg_tables(&self) -> Result<Vec<IcebergTable>>;
215
216    async fn get_fragment_by_id(
217        &self,
218        fragment_id: FragmentId,
219    ) -> Result<Option<FragmentDistribution>>;
220
221    async fn get_fragment_vnodes(
222        &self,
223        fragment_id: FragmentId,
224    ) -> Result<Vec<(ActorId, Vec<u32>)>>;
225
226    async fn get_actor_vnodes(&self, actor_id: ActorId) -> Result<Vec<u32>>;
227
228    fn worker_id(&self) -> WorkerId;
229
230    async fn set_sync_log_store_aligned(&self, job_id: JobId, aligned: bool) -> Result<()>;
231
232    async fn compact_iceberg_table(&self, sink_id: SinkId) -> Result<IcebergCompactionTaskId>;
233
234    async fn rewrite_iceberg_table_manifests(&self, sink_id: SinkId) -> Result<()>;
235
236    async fn expire_iceberg_table_snapshots(&self, sink_id: SinkId) -> Result<()>;
237
238    async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse>;
239
240    fn cluster_id(&self) -> &str;
241
242    async fn list_unmigrated_tables(&self) -> Result<HashMap<TableId, String>>;
243
244    async fn update_compaction_config(
245        &self,
246        compaction_group_ids: Vec<CompactionGroupId>,
247        configs: Vec<PbMutableConfig>,
248    ) -> Result<()>;
249}
250
251pub struct FrontendMetaClientImpl(pub MetaClient);
252
253#[async_trait::async_trait]
254impl FrontendMetaClient for FrontendMetaClientImpl {
255    async fn try_unregister(&self) {
256        self.0.try_unregister().await;
257    }
258
259    async fn flush(&self, database_id: DatabaseId) -> Result<HummockVersionId> {
260        self.0.flush(database_id).await
261    }
262
263    async fn backup_meta(&self, remarks: Option<String>) -> Result<u64> {
264        self.0.backup_meta(remarks).await
265    }
266
267    async fn get_backup_job_status(&self, job_id: u64) -> Result<(BackupJobStatus, String)> {
268        self.0.get_backup_job_status(job_id).await
269    }
270
271    async fn delete_meta_snapshot(&self, snapshot_ids: &[u64]) -> Result<()> {
272        self.0.delete_meta_snapshot(snapshot_ids).await
273    }
274
275    async fn recover(&self) -> Result<()> {
276        self.0.recover().await
277    }
278
279    async fn cancel_creating_jobs(&self, infos: PbJobs) -> Result<Vec<u32>> {
280        self.0.cancel_creating_jobs(infos).await
281    }
282
283    async fn list_table_fragments(
284        &self,
285        job_ids: &[JobId],
286    ) -> Result<HashMap<JobId, TableFragmentInfo>> {
287        self.0.list_table_fragments(job_ids).await
288    }
289
290    async fn list_streaming_job_states(&self) -> Result<Vec<StreamingJobState>> {
291        self.0.list_streaming_job_states().await
292    }
293
294    async fn list_fragment_distribution(
295        &self,
296        include_node: bool,
297    ) -> Result<Vec<FragmentDistribution>> {
298        self.0.list_fragment_distributions(include_node).await
299    }
300
301    async fn list_creating_fragment_distribution(&self) -> Result<Vec<FragmentDistribution>> {
302        self.0.list_creating_fragment_distribution().await
303    }
304
305    async fn list_actor_states(&self) -> Result<Vec<ActorState>> {
306        self.0.list_actor_states().await
307    }
308
309    async fn list_actor_splits(&self) -> Result<Vec<ActorSplit>> {
310        self.0.list_actor_splits().await
311    }
312
313    async fn list_meta_snapshots(&self) -> Result<Vec<MetaSnapshotMetadata>> {
314        let manifest = self.0.get_meta_snapshot_manifest().await?;
315        Ok(manifest.snapshot_metadata)
316    }
317
318    async fn list_sink_log_store_tables(
319        &self,
320    ) -> Result<Vec<list_sink_log_store_tables_response::SinkLogStoreTable>> {
321        self.0.list_sink_log_store_tables().await
322    }
323
324    async fn set_system_param(
325        &self,
326        param: String,
327        value: Option<String>,
328    ) -> Result<Option<SystemParamsReader>> {
329        self.0.set_system_param(param, value).await
330    }
331
332    async fn clear_file_cache(&self, clear_meta_cache: bool, clear_data_cache: bool) -> Result<()> {
333        self.0
334            .clear_file_cache(clear_meta_cache, clear_data_cache)
335            .await
336    }
337
338    async fn get_session_params(&self) -> Result<SessionConfig> {
339        let session_config: SessionConfig =
340            serde_json::from_str(&self.0.get_session_params().await?)
341                .context("failed to parse session config")?;
342        Ok(session_config)
343    }
344
345    async fn set_session_param(&self, param: String, value: Option<String>) -> Result<String> {
346        self.0.set_session_param(param, value).await
347    }
348
349    async fn get_ddl_progress(&self) -> Result<Vec<DdlProgress>> {
350        let ddl_progress = self.0.get_ddl_progress().await?;
351        Ok(ddl_progress)
352    }
353
354    async fn get_tables(
355        &self,
356        table_ids: Vec<TableId>,
357        include_dropped_tables: bool,
358    ) -> Result<HashMap<TableId, Table>> {
359        let tables = self.0.get_tables(table_ids, include_dropped_tables).await?;
360        Ok(tables)
361    }
362
363    async fn list_hummock_pinned_versions(&self) -> Result<Vec<(WorkerId, HummockVersionId)>> {
364        let pinned_versions = self
365            .0
366            .risectl_get_pinned_versions_summary()
367            .await?
368            .summary
369            .unwrap()
370            .pinned_versions;
371        let ret = pinned_versions
372            .into_iter()
373            .map(|v| (v.context_id, v.min_pinned_id))
374            .collect();
375        Ok(ret)
376    }
377
378    async fn get_hummock_current_version(&self) -> Result<HummockVersion> {
379        self.0.get_current_version().await
380    }
381
382    async fn get_hummock_table_change_log(
383        &self,
384        start_epoch_inclusive: Option<u64>,
385        end_epoch_inclusive: Option<u64>,
386        table_ids: Option<HashSet<TableId>>,
387        exclude_empty: bool,
388        limit: Option<u32>,
389    ) -> Result<TableChangeLogs> {
390        self.0
391            .get_table_change_logs(
392                true,
393                start_epoch_inclusive,
394                end_epoch_inclusive,
395                table_ids,
396                exclude_empty,
397                limit,
398            )
399            .await
400    }
401
402    async fn get_hummock_checkpoint_version(&self) -> Result<HummockVersion> {
403        self.0
404            .risectl_get_checkpoint_hummock_version()
405            .await
406            .map(|v| HummockVersion::from_rpc_protobuf(&v.checkpoint_version.unwrap()))
407    }
408
409    async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>> {
410        // FIXME #8612: there can be lots of version deltas, so better to fetch them by pages and refactor `SysRowSeqScanExecutor` to yield multiple chunks.
411        self.0
412            .list_version_deltas(HummockVersionId::new(0), u32::MAX, u64::MAX)
413            .await
414    }
415
416    async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>> {
417        self.0.list_branched_object().await
418    }
419
420    async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>> {
421        self.0.risectl_list_compaction_group().await
422    }
423
424    async fn list_hummock_active_write_limits(
425        &self,
426    ) -> Result<HashMap<CompactionGroupId, WriteLimit>> {
427        self.0.list_active_write_limit().await
428    }
429
430    async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>> {
431        self.0.list_hummock_meta_config().await
432    }
433
434    async fn list_event_log(&self) -> Result<Vec<EventLog>> {
435        self.0.list_event_log().await
436    }
437
438    async fn list_compact_task_assignment(&self) -> Result<Vec<CompactTaskAssignment>> {
439        self.0.list_compact_task_assignment().await
440    }
441
442    async fn list_all_nodes(&self) -> Result<Vec<WorkerNode>> {
443        self.0.list_worker_nodes(None).await
444    }
445
446    async fn list_compact_task_progress(&self) -> Result<Vec<CompactTaskProgress>> {
447        self.0.list_compact_task_progress().await
448    }
449
450    async fn apply_throttle(
451        &self,
452        throttle_target: PbThrottleTarget,
453        throttle_type: risingwave_pb::common::PbThrottleType,
454        id: u32,
455        rate_limit: Option<u32>,
456    ) -> Result<()> {
457        self.0
458            .apply_throttle(throttle_target, throttle_type, id, rate_limit)
459            .await
460            .map(|_| ())
461    }
462
463    async fn alter_fragment_parallelism(
464        &self,
465        fragment_ids: Vec<FragmentId>,
466        parallelism: Option<PbTableParallelism>,
467    ) -> Result<()> {
468        self.0
469            .alter_fragment_parallelism(fragment_ids, parallelism)
470            .await
471    }
472
473    async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus> {
474        self.0.get_cluster_recovery_status().await
475    }
476
477    async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>> {
478        self.0.get_cluster_limits().await
479    }
480
481    async fn list_rate_limits(&self) -> Result<Vec<RateLimitInfo>> {
482        self.0.list_rate_limits().await
483    }
484
485    async fn list_cdc_progress(&self) -> Result<HashMap<JobId, PbCdcProgress>> {
486        self.0.list_cdc_progress().await
487    }
488
489    async fn get_meta_store_endpoint(&self) -> Result<String> {
490        self.0.get_meta_store_endpoint().await
491    }
492
493    async fn alter_sink_props(
494        &self,
495        sink_id: SinkId,
496        changed_props: BTreeMap<String, String>,
497        changed_secret_refs: BTreeMap<String, PbSecretRef>,
498        connector_conn_ref: Option<ConnectionId>,
499    ) -> Result<()> {
500        self.0
501            .alter_sink_props(
502                sink_id,
503                changed_props,
504                changed_secret_refs,
505                connector_conn_ref,
506            )
507            .await
508    }
509
510    async fn alter_iceberg_table_props(
511        &self,
512        table_id: TableId,
513        sink_id: SinkId,
514        source_id: SourceId,
515        changed_props: BTreeMap<String, String>,
516        changed_secret_refs: BTreeMap<String, PbSecretRef>,
517        connector_conn_ref: Option<ConnectionId>,
518    ) -> Result<()> {
519        self.0
520            .alter_iceberg_table_props(
521                table_id,
522                sink_id,
523                source_id,
524                changed_props,
525                changed_secret_refs,
526                connector_conn_ref,
527            )
528            .await
529    }
530
531    async fn alter_source_connector_props(
532        &self,
533        source_id: SourceId,
534        changed_props: BTreeMap<String, String>,
535        changed_secret_refs: BTreeMap<String, PbSecretRef>,
536        connector_conn_ref: Option<ConnectionId>,
537    ) -> Result<()> {
538        self.0
539            .alter_source_connector_props(
540                source_id,
541                changed_props,
542                changed_secret_refs,
543                connector_conn_ref,
544            )
545            .await
546    }
547
548    async fn alter_connection_connector_props(
549        &self,
550        connection_id: u32,
551        changed_props: BTreeMap<String, String>,
552        changed_secret_refs: BTreeMap<String, PbSecretRef>,
553    ) -> Result<()> {
554        self.0
555            .alter_connection_connector_props(connection_id, changed_props, changed_secret_refs)
556            .await
557    }
558
559    async fn list_hosted_iceberg_tables(&self) -> Result<Vec<IcebergTable>> {
560        self.0.list_hosted_iceberg_tables().await
561    }
562
563    async fn get_fragment_by_id(
564        &self,
565        fragment_id: FragmentId,
566    ) -> Result<Option<FragmentDistribution>> {
567        self.0.get_fragment_by_id(fragment_id).await
568    }
569
570    async fn get_fragment_vnodes(
571        &self,
572        fragment_id: FragmentId,
573    ) -> Result<Vec<(ActorId, Vec<u32>)>> {
574        self.0.get_fragment_vnodes(fragment_id).await
575    }
576
577    async fn get_actor_vnodes(&self, actor_id: ActorId) -> Result<Vec<u32>> {
578        self.0.get_actor_vnodes(actor_id).await
579    }
580
581    fn worker_id(&self) -> WorkerId {
582        self.0.worker_id()
583    }
584
585    async fn set_sync_log_store_aligned(&self, job_id: JobId, aligned: bool) -> Result<()> {
586        self.0.set_sync_log_store_aligned(job_id, aligned).await
587    }
588
589    async fn compact_iceberg_table(&self, sink_id: SinkId) -> Result<IcebergCompactionTaskId> {
590        self.0.compact_iceberg_table(sink_id).await
591    }
592
593    async fn rewrite_iceberg_table_manifests(&self, sink_id: SinkId) -> Result<()> {
594        self.0.rewrite_iceberg_table_manifests(sink_id).await
595    }
596
597    async fn expire_iceberg_table_snapshots(&self, sink_id: SinkId) -> Result<()> {
598        self.0.expire_iceberg_table_snapshots(sink_id).await
599    }
600
601    async fn refresh(&self, request: RefreshRequest) -> Result<RefreshResponse> {
602        self.0.refresh(request).await
603    }
604
605    fn cluster_id(&self) -> &str {
606        self.0.cluster_id()
607    }
608
609    async fn list_unmigrated_tables(&self) -> Result<HashMap<TableId, String>> {
610        self.0.list_unmigrated_tables().await
611    }
612
613    async fn list_refresh_table_states(&self) -> Result<Vec<RefreshTableState>> {
614        self.0.list_refresh_table_states().await
615    }
616
617    async fn list_iceberg_compaction_status(&self) -> Result<Vec<IcebergCompactionStatus>> {
618        self.0.list_iceberg_compaction_status().await
619    }
620
621    async fn update_compaction_config(
622        &self,
623        compaction_group_ids: Vec<CompactionGroupId>,
624        configs: Vec<PbMutableConfig>,
625    ) -> Result<()> {
626        self.0
627            .risectl_update_compaction_config(&compaction_group_ids, &configs)
628            .await
629    }
630}
risingwave_frontend/meta_client.rs

risingwave_frontend/
meta_client.rs
