gnuhpc commented on code in PR #631:
URL: https://github.com/apache/fluss-rust/pull/631#discussion_r3449553367
##########
crates/fluss/src/client/admin.rs:
##########
@@ -483,4 +496,371 @@ impl FlussAdmin {
}
Ok(tasks)
}
+
+ /// List database summaries (name, created_time, table_count).
+ pub async fn list_database_summaries(&self) ->
Result<Vec<DatabaseSummary>> {
+ let response = self
+ .admin_gateway()
+ .await?
+ .request(ListDatabaseSummariesRequest::new())
+ .await?;
+ Ok(response
+ .database_summary
+ .iter()
+ .map(DatabaseSummary::from_pb)
+ .collect())
+ }
+
+ /// Alter a database's configuration.
+ pub async fn alter_database(
+ &self,
+ name: &str,
+ config_changes: Vec<AlterConfig>,
+ ignore_if_not_exists: bool,
+ ) -> Result<()> {
+ let _response = self
+ .admin_gateway()
+ .await?
+ .request(AlterDatabaseRequest::new(
+ name,
+ ignore_if_not_exists,
+ config_changes,
+ ))
+ .await?;
+ Ok(())
+ }
+
+ /// Alter a table: config changes plus any combination of
add/drop/rename/modify columns.
+ /// Bundle the column-level edits in [`AlterTableChanges`].
+ pub async fn alter_table(
+ &self,
+ table_path: &TablePath,
+ ignore_if_not_exists: bool,
+ changes: AlterTableChanges,
+ ) -> Result<()> {
+ let _response = self
+ .admin_gateway()
+ .await?
+ .request(AlterTableRequest::new(
+ table_path,
+ ignore_if_not_exists,
+ changes.config_changes,
+ changes.add_columns,
+ changes.drop_columns,
+ changes.rename_columns,
+ changes.modify_columns,
+ ))
+ .await?;
+ Ok(())
+ }
+
+ /// Get table statistics for buckets. Pass empty `target_columns` to
request stats for all columns.
+ pub async fn get_table_stats(
+ &self,
+ table_id: i64,
+ buckets_req: Vec<crate::metadata::BucketStatsRequest>,
+ target_columns: Vec<i32>,
+ ) -> Result<TableStats> {
+ let response = self
+ .admin_gateway()
+ .await?
+ .request(GetTableStatsRequest::new(
+ table_id,
+ buckets_req,
+ target_columns,
+ ))
+ .await?;
+ Ok(TableStats::from_pb(&response))
+ }
+
+ /// Get the latest KV snapshots for a table (optionally scoped to one
partition).
+ pub async fn get_latest_kv_snapshots(
+ &self,
+ table_path: &TablePath,
+ partition_name: Option<&str>,
+ ) -> Result<LatestKvSnapshots> {
+ let response = self
+ .admin_gateway()
+ .await?
+ .request(GetLatestKvSnapshotsRequest::new(table_path,
partition_name))
+ .await?;
+ Ok(LatestKvSnapshots::from_pb(&response))
+ }
+
+ /// Get KV snapshot metadata (manifest file list).
+ pub async fn get_kv_snapshot_metadata(
+ &self,
+ table_id: i64,
+ partition_id: Option<i64>,
+ bucket_id: i32,
+ snapshot_id: i64,
+ ) -> Result<KvSnapshotMetadata> {
+ let response = self
+ .admin_gateway()
+ .await?
+ .request(GetKvSnapshotMetadataRequest::new(
+ table_id,
+ partition_id,
+ bucket_id,
+ snapshot_id,
+ ))
+ .await?;
+ Ok(KvSnapshotMetadata::from_pb(&response))
+ }
+
+ /// Acquire a KV snapshot lease. Returns the snapshots the server could
not lease.
+ pub async fn create_kv_snapshot_lease(
+ &self,
+ lease_id: &str,
+ lease_duration_ms: i64,
+ snapshots_to_lease: Vec<KvSnapshotLeaseForTable>,
+ ) -> Result<AcquireKvSnapshotLeaseResult> {
+ let response = self
+ .admin_gateway()
+ .await?
+ .request(AcquireKvSnapshotLeaseRequest::new(
+ lease_id,
+ lease_duration_ms,
+ snapshots_to_lease,
+ ))
+ .await?;
+ Ok(AcquireKvSnapshotLeaseResult::from_pb(&response))
+ }
+
+ /// Get a specific lake snapshot for a table.
+ pub async fn get_lake_snapshot(
+ &self,
+ table_path: &TablePath,
+ snapshot_id: Option<i64>,
+ ) -> Result<LakeSnapshotInfo> {
+ let response = self
+ .admin_gateway()
+ .await?
+ .request(GetLakeSnapshotRequest::new(table_path, snapshot_id,
None))
Review Comment:
Done. Exposed as a parameter in `get_lake_snapshot(table_path, snapshot_id,
readable)`.
##########
crates/fluss/src/metadata/cluster_health.rs:
##########
@@ -0,0 +1,59 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::proto::GetClusterHealthResponse;
+
+/// Result of `get_cluster_health`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClusterHealth {
+ pub num_replicas: i32,
+ pub in_sync_replicas: i32,
+ pub num_leader_replicas: i32,
+ pub active_leader_replicas: i32,
+ pub status: i32,
Review Comment:
Done. Added `ClusterHealthStatus` enum (Green/Yellow/Red/Unknown). `from_pb`
now returns `Result` since the conversion can fail on unknown values.
##########
crates/fluss/src/metadata/rebalance.rs:
##########
@@ -0,0 +1,125 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::proto::{
+ ListRebalanceProgressResponse, PbRebalancePlanForBucket,
PbRebalanceProgressForBucket,
+ PbRebalanceProgressForTable,
+};
+
+/// Per-bucket plan in a rebalance: who the leader was and who it will be, who
+/// the replicas were and who they will be.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BucketRebalancePlan {
+ pub partition_id: Option<i64>,
+ pub bucket_id: i32,
+ pub original_leader: Option<i32>,
+ pub new_leader: Option<i32>,
+ pub original_replicas: Vec<i32>,
+ pub new_replicas: Vec<i32>,
+}
+
+impl BucketRebalancePlan {
+ pub fn from_pb(pb: &PbRebalancePlanForBucket) -> Self {
+ Self {
+ partition_id: pb.partition_id,
+ bucket_id: pb.bucket_id,
+ original_leader: pb.original_leader,
+ new_leader: pb.new_leader,
+ original_replicas: pb.original_replicas.clone(),
+ new_replicas: pb.new_replicas.clone(),
+ }
+ }
+}
+
+/// Per-bucket rebalance progress: the planned move and its current status
code.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BucketRebalanceProgress {
+ pub rebalance_plan: BucketRebalancePlan,
+ pub rebalance_status: i32,
Review Comment:
Done. Added `RebalanceStatus` enum
(NotStarted/Rebalancing/Failed/Completed/Canceled/Timeout). `from_pb` returns
`Result` with `collect::<Result<Vec<_>>>()?` for the nested bucket progress.
##########
crates/fluss/src/metadata/rebalance.rs:
##########
@@ -0,0 +1,125 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::proto::{
+ ListRebalanceProgressResponse, PbRebalancePlanForBucket,
PbRebalanceProgressForBucket,
+ PbRebalanceProgressForTable,
+};
+
+/// Per-bucket plan in a rebalance: who the leader was and who it will be, who
+/// the replicas were and who they will be.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BucketRebalancePlan {
+ pub partition_id: Option<i64>,
+ pub bucket_id: i32,
+ pub original_leader: Option<i32>,
+ pub new_leader: Option<i32>,
+ pub original_replicas: Vec<i32>,
+ pub new_replicas: Vec<i32>,
+}
+
+impl BucketRebalancePlan {
+ pub fn from_pb(pb: &PbRebalancePlanForBucket) -> Self {
+ Self {
+ partition_id: pb.partition_id,
+ bucket_id: pb.bucket_id,
+ original_leader: pb.original_leader,
+ new_leader: pb.new_leader,
+ original_replicas: pb.original_replicas.clone(),
+ new_replicas: pb.new_replicas.clone(),
+ }
+ }
+}
+
+/// Per-bucket rebalance progress: the planned move and its current status
code.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BucketRebalanceProgress {
+ pub rebalance_plan: BucketRebalancePlan,
+ pub rebalance_status: i32,
+}
+
+impl BucketRebalanceProgress {
+ pub fn from_pb(pb: &PbRebalanceProgressForBucket) -> Self {
+ Self {
+ rebalance_plan: BucketRebalancePlan::from_pb(&pb.rebalance_plan),
+ rebalance_status: pb.rebalance_status,
+ }
+ }
+}
+
+/// All bucket progress for one table.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct TableRebalanceProgress {
+ pub table_id: i64,
+ pub buckets_progress: Vec<BucketRebalanceProgress>,
+}
+
+impl TableRebalanceProgress {
+ pub fn from_pb(pb: &PbRebalanceProgressForTable) -> Self {
+ Self {
+ table_id: pb.table_id,
+ buckets_progress: pb
+ .buckets_progress
+ .iter()
+ .map(BucketRebalanceProgress::from_pb)
+ .collect(),
+ }
+ }
+}
+
+/// Result of `list_rebalance_progress`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RebalanceProgress {
+ pub rebalance_id: Option<String>,
+ pub rebalance_status: Option<i32>,
Review Comment:
Done. Same enum applied here as `Option<RebalanceStatus>`.
##########
crates/fluss/src/metadata/kv_snapshot.rs:
##########
@@ -0,0 +1,170 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use crate::proto::{
+ AcquireKvSnapshotLeaseResponse, GetKvSnapshotMetadataResponse,
GetLatestKvSnapshotsResponse,
+ ListKvSnapshotsResponse, PbKvSnapshot, PbRemotePathAndLocalFile,
+};
+
+use crate::metadata::KvSnapshotLeaseForTable;
+
+/// Per-bucket KV snapshot info.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct KvSnapshot {
+ pub bucket_id: i32,
Review Comment:
Done. Applied `TableId`/`PartitionId`/`BucketId` across all metadata types:
kv_snapshot, lake_snapshot, producer_offsets, rebalance, table_stats, and
kv_snapshot_lease.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
