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

risingwave_frontend/
meta_client.rs
