fresh-borzoni commented on code in PR #631: URL: https://github.com/apache/fluss-rust/pull/631#discussion_r3448522974
########## 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: ditto ########## 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: enum? ########## 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: these use raw i64/i32 for table_id/partition_id/bucket_id, but the crate has TableId/PartitionId/BucketId aliases (lib.rs) that the rest of the code use. Let's stay consistent ########## 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: why do we hardcode readable to None? ########## 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: enum? -- 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]
