risingwave_frontend/
meta_client.rs1use std::collections::HashMap;
16
17use anyhow::Context;
18use risingwave_common::session_config::SessionConfig;
19use risingwave_common::system_param::reader::SystemParamsReader;
20use risingwave_common::util::cluster_limit::ClusterLimit;
21use risingwave_hummock_sdk::HummockVersionId;
22use risingwave_hummock_sdk::version::{HummockVersion, HummockVersionDelta};
23use risingwave_pb::backup_service::MetaSnapshotMetadata;
24use risingwave_pb::catalog::Table;
25use risingwave_pb::common::WorkerNode;
26use risingwave_pb::ddl_service::DdlProgress;
27use risingwave_pb::hummock::write_limits::WriteLimit;
28use risingwave_pb::hummock::{
29 BranchedObject, CompactTaskAssignment, CompactTaskProgress, CompactionGroupInfo,
30};
31use risingwave_pb::meta::cancel_creating_jobs_request::PbJobs;
32use risingwave_pb::meta::list_actor_splits_response::ActorSplit;
33use risingwave_pb::meta::list_actor_states_response::ActorState;
34use risingwave_pb::meta::list_fragment_distribution_response::FragmentDistribution;
35use risingwave_pb::meta::list_object_dependencies_response::PbObjectDependencies;
36use risingwave_pb::meta::list_rate_limits_response::RateLimitInfo;
37use risingwave_pb::meta::list_streaming_job_states_response::StreamingJobState;
38use risingwave_pb::meta::list_table_fragments_response::TableFragmentInfo;
39use risingwave_pb::meta::{EventLog, PbThrottleTarget, RecoveryStatus};
40use risingwave_rpc_client::error::Result;
41use risingwave_rpc_client::{HummockMetaClient, MetaClient};
42
43use crate::catalog::DatabaseId;
44
45#[async_trait::async_trait]
51pub trait FrontendMetaClient: Send + Sync {
52 async fn try_unregister(&self);
53
54 async fn flush(&self, database_id: DatabaseId) -> Result<HummockVersionId>;
55
56 async fn wait(&self) -> Result<()>;
57
58 async fn recover(&self) -> Result<()>;
59
60 async fn cancel_creating_jobs(&self, jobs: PbJobs) -> Result<Vec<u32>>;
61
62 async fn list_table_fragments(
63 &self,
64 table_ids: &[u32],
65 ) -> Result<HashMap<u32, TableFragmentInfo>>;
66
67 async fn list_streaming_job_states(&self) -> Result<Vec<StreamingJobState>>;
68
69 async fn list_fragment_distribution(&self) -> Result<Vec<FragmentDistribution>>;
70
71 async fn list_actor_states(&self) -> Result<Vec<ActorState>>;
72
73 async fn list_actor_splits(&self) -> Result<Vec<ActorSplit>>;
74
75 async fn list_object_dependencies(&self) -> Result<Vec<PbObjectDependencies>>;
76
77 async fn list_meta_snapshots(&self) -> Result<Vec<MetaSnapshotMetadata>>;
78
79 async fn get_system_params(&self) -> Result<SystemParamsReader>;
80
81 async fn set_system_param(
82 &self,
83 param: String,
84 value: Option<String>,
85 ) -> Result<Option<SystemParamsReader>>;
86
87 async fn get_session_params(&self) -> Result<SessionConfig>;
88
89 async fn set_session_param(&self, param: String, value: Option<String>) -> Result<String>;
90
91 async fn get_ddl_progress(&self) -> Result<Vec<DdlProgress>>;
92
93 async fn get_tables(
94 &self,
95 table_ids: &[u32],
96 include_dropped_table: bool,
97 ) -> Result<HashMap<u32, Table>>;
98
99 async fn list_hummock_pinned_versions(&self) -> Result<Vec<(u32, u64)>>;
101
102 async fn get_hummock_current_version(&self) -> Result<HummockVersion>;
103
104 async fn get_hummock_checkpoint_version(&self) -> Result<HummockVersion>;
105
106 async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>>;
107
108 async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>>;
109
110 async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>>;
111
112 async fn list_hummock_active_write_limits(&self) -> Result<HashMap<u64, WriteLimit>>;
113
114 async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>>;
115
116 async fn list_event_log(&self) -> Result<Vec<EventLog>>;
117 async fn list_compact_task_assignment(&self) -> Result<Vec<CompactTaskAssignment>>;
118
119 async fn list_all_nodes(&self) -> Result<Vec<WorkerNode>>;
120
121 async fn list_compact_task_progress(&self) -> Result<Vec<CompactTaskProgress>>;
122
123 async fn apply_throttle(
124 &self,
125 kind: PbThrottleTarget,
126 id: u32,
127 rate_limit: Option<u32>,
128 ) -> Result<()>;
129
130 async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus>;
131
132 async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>>;
133
134 async fn list_rate_limits(&self) -> Result<Vec<RateLimitInfo>>;
135
136 async fn get_meta_store_endpoint(&self) -> Result<String>;
137}
138
139pub struct FrontendMetaClientImpl(pub MetaClient);
140
141#[async_trait::async_trait]
142impl FrontendMetaClient for FrontendMetaClientImpl {
143 async fn try_unregister(&self) {
144 self.0.try_unregister().await;
145 }
146
147 async fn flush(&self, database_id: DatabaseId) -> Result<HummockVersionId> {
148 self.0.flush(database_id).await
149 }
150
151 async fn wait(&self) -> Result<()> {
152 self.0.wait().await
153 }
154
155 async fn recover(&self) -> Result<()> {
156 self.0.recover().await
157 }
158
159 async fn cancel_creating_jobs(&self, infos: PbJobs) -> Result<Vec<u32>> {
160 self.0.cancel_creating_jobs(infos).await
161 }
162
163 async fn list_table_fragments(
164 &self,
165 table_ids: &[u32],
166 ) -> Result<HashMap<u32, TableFragmentInfo>> {
167 self.0.list_table_fragments(table_ids).await
168 }
169
170 async fn list_streaming_job_states(&self) -> Result<Vec<StreamingJobState>> {
171 self.0.list_streaming_job_states().await
172 }
173
174 async fn list_fragment_distribution(&self) -> Result<Vec<FragmentDistribution>> {
175 self.0.list_fragment_distributions().await
176 }
177
178 async fn list_actor_states(&self) -> Result<Vec<ActorState>> {
179 self.0.list_actor_states().await
180 }
181
182 async fn list_actor_splits(&self) -> Result<Vec<ActorSplit>> {
183 self.0.list_actor_splits().await
184 }
185
186 async fn list_object_dependencies(&self) -> Result<Vec<PbObjectDependencies>> {
187 self.0.list_object_dependencies().await
188 }
189
190 async fn list_meta_snapshots(&self) -> Result<Vec<MetaSnapshotMetadata>> {
191 let manifest = self.0.get_meta_snapshot_manifest().await?;
192 Ok(manifest.snapshot_metadata)
193 }
194
195 async fn get_system_params(&self) -> Result<SystemParamsReader> {
196 self.0.get_system_params().await
197 }
198
199 async fn set_system_param(
200 &self,
201 param: String,
202 value: Option<String>,
203 ) -> Result<Option<SystemParamsReader>> {
204 self.0.set_system_param(param, value).await
205 }
206
207 async fn get_session_params(&self) -> Result<SessionConfig> {
208 let session_config: SessionConfig =
209 serde_json::from_str(&self.0.get_session_params().await?)
210 .context("failed to parse session config")?;
211 Ok(session_config)
212 }
213
214 async fn set_session_param(&self, param: String, value: Option<String>) -> Result<String> {
215 self.0.set_session_param(param, value).await
216 }
217
218 async fn get_ddl_progress(&self) -> Result<Vec<DdlProgress>> {
219 let ddl_progress = self.0.get_ddl_progress().await?;
220 Ok(ddl_progress)
221 }
222
223 async fn get_tables(
224 &self,
225 table_ids: &[u32],
226 include_dropped_tables: bool,
227 ) -> Result<HashMap<u32, Table>> {
228 let tables = self.0.get_tables(table_ids, include_dropped_tables).await?;
229 Ok(tables)
230 }
231
232 async fn list_hummock_pinned_versions(&self) -> Result<Vec<(u32, u64)>> {
233 let pinned_versions = self
234 .0
235 .risectl_get_pinned_versions_summary()
236 .await?
237 .summary
238 .unwrap()
239 .pinned_versions;
240 let ret = pinned_versions
241 .into_iter()
242 .map(|v| (v.context_id, v.min_pinned_id))
243 .collect();
244 Ok(ret)
245 }
246
247 async fn get_hummock_current_version(&self) -> Result<HummockVersion> {
248 self.0.get_current_version().await
249 }
250
251 async fn get_hummock_checkpoint_version(&self) -> Result<HummockVersion> {
252 self.0
253 .risectl_get_checkpoint_hummock_version()
254 .await
255 .map(|v| HummockVersion::from_rpc_protobuf(&v.checkpoint_version.unwrap()))
256 }
257
258 async fn list_version_deltas(&self) -> Result<Vec<HummockVersionDelta>> {
259 self.0
261 .list_version_deltas(HummockVersionId::new(0), u32::MAX, u64::MAX)
262 .await
263 }
264
265 async fn list_branched_objects(&self) -> Result<Vec<BranchedObject>> {
266 self.0.list_branched_object().await
267 }
268
269 async fn list_hummock_compaction_group_configs(&self) -> Result<Vec<CompactionGroupInfo>> {
270 self.0.risectl_list_compaction_group().await
271 }
272
273 async fn list_hummock_active_write_limits(&self) -> Result<HashMap<u64, WriteLimit>> {
274 self.0.list_active_write_limit().await
275 }
276
277 async fn list_hummock_meta_configs(&self) -> Result<HashMap<String, String>> {
278 self.0.list_hummock_meta_config().await
279 }
280
281 async fn list_event_log(&self) -> Result<Vec<EventLog>> {
282 self.0.list_event_log().await
283 }
284
285 async fn list_compact_task_assignment(&self) -> Result<Vec<CompactTaskAssignment>> {
286 self.0.list_compact_task_assignment().await
287 }
288
289 async fn list_all_nodes(&self) -> Result<Vec<WorkerNode>> {
290 self.0.list_worker_nodes(None).await
291 }
292
293 async fn list_compact_task_progress(&self) -> Result<Vec<CompactTaskProgress>> {
294 self.0.list_compact_task_progress().await
295 }
296
297 async fn apply_throttle(
298 &self,
299 kind: PbThrottleTarget,
300 id: u32,
301 rate_limit: Option<u32>,
302 ) -> Result<()> {
303 self.0
304 .apply_throttle(kind, id, rate_limit)
305 .await
306 .map(|_| ())
307 }
308
309 async fn get_cluster_recovery_status(&self) -> Result<RecoveryStatus> {
310 self.0.get_cluster_recovery_status().await
311 }
312
313 async fn get_cluster_limits(&self) -> Result<Vec<ClusterLimit>> {
314 self.0.get_cluster_limits().await
315 }
316
317 async fn list_rate_limits(&self) -> Result<Vec<RateLimitInfo>> {
318 self.0.list_rate_limits().await
319 }
320
321 async fn get_meta_store_endpoint(&self) -> Result<String> {
322 self.0.get_meta_store_endpoint().await
323 }
324}