You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@fluss.apache.org by ab...@apache.org on 2026/06/19 22:50:31 UTC
(fluss-rust) 01/02: [rust] Add RPC message wrappers for core DB/table operations
This is an automated email from the ASF dual-hosted git repository.
fresh-borzoni pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss-rust.git
commit 6c1c534fb4a70aeed3e6c37c532864e2001449db
Author: warmbupt <wa...@gmail.com>
AuthorDate: Thu Jun 18 10:11:46 2026 +0800
[rust] Add RPC message wrappers for core DB/table operations
Add 9 RPC message wrapper types:
- alter_database, alter_table (DDL operations)
- get_table_stats (table statistics)
- list_database_summaries (database listing with summaries)
- create_acls, list_acls, drop_acls (ACL management)
- describe_cluster_configs, alter_cluster_configs (cluster configuration)
Each wrapper follows the standard pattern: a request struct wrapping the
proto-generated type, implementing RequestBody (tying to ApiKey and ResponseBody),
WriteType, and ReadType.
Co-Authored-By: Claude Opus 4.8 <no...@anthropic.com>
---
crates/fluss/src/metadata/acl.rs | 271 +++++++++++++++++++++
crates/fluss/src/metadata/config.rs | 124 ++++++++++
crates/fluss/src/metadata/mod.rs | 8 +
crates/fluss/src/metadata/table_change.rs | 226 +++++++++++++++++
crates/fluss/src/metadata/table_stats.rs | 65 +++++
.../fluss/src/rpc/message/alter_cluster_configs.rs | 47 ++++
crates/fluss/src/rpc/message/alter_database.rs | 54 ++++
crates/fluss/src/rpc/message/alter_table.rs | 62 +++++
crates/fluss/src/rpc/message/create_acls.rs | 47 ++++
.../src/rpc/message/describe_cluster_configs.rs | 44 ++++
crates/fluss/src/rpc/message/drop_acls.rs | 47 ++++
crates/fluss/src/rpc/message/get_table_stats.rs | 53 ++++
crates/fluss/src/rpc/message/list_acls.rs | 47 ++++
.../message/list_database_summaries.rs} | 41 ++--
crates/fluss/src/rpc/message/mod.rs | 18 ++
15 files changed, 1140 insertions(+), 14 deletions(-)
diff --git a/crates/fluss/src/metadata/acl.rs b/crates/fluss/src/metadata/acl.rs
new file mode 100644
index 00000000..84ec97af
--- /dev/null
+++ b/crates/fluss/src/metadata/acl.rs
@@ -0,0 +1,271 @@
+// 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::error::{Error, Result};
+use crate::proto::{PbAclFilter, PbAclInfo};
+
+/// Mirrors Java `org.apache.fluss.security.acl.ResourceType`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ResourceType {
+ Any,
+ Cluster,
+ Database,
+ Table,
+}
+
+impl ResourceType {
+ pub fn to_i32(self) -> i32 {
+ match self {
+ Self::Any => 1,
+ Self::Cluster => 2,
+ Self::Database => 3,
+ Self::Table => 4,
+ }
+ }
+
+ pub fn try_from_i32(value: i32) -> Result<Self> {
+ match value {
+ 1 => Ok(Self::Any),
+ 2 => Ok(Self::Cluster),
+ 3 => Ok(Self::Database),
+ 4 => Ok(Self::Table),
+ _ => Err(Error::IllegalArgument {
+ message: format!("Unknown resource type code: {value}"),
+ }),
+ }
+ }
+}
+
+/// Mirrors Java `org.apache.fluss.security.acl.OperationType`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OperationType {
+ Any,
+ All,
+ Read,
+ Write,
+ Create,
+ Drop,
+ Alter,
+ Describe,
+}
+
+impl OperationType {
+ pub fn to_i32(self) -> i32 {
+ match self {
+ Self::Any => 1,
+ Self::All => 2,
+ Self::Read => 3,
+ Self::Write => 4,
+ Self::Create => 5,
+ Self::Drop => 6,
+ Self::Alter => 7,
+ Self::Describe => 8,
+ }
+ }
+
+ pub fn try_from_i32(value: i32) -> Result<Self> {
+ match value {
+ 1 => Ok(Self::Any),
+ 2 => Ok(Self::All),
+ 3 => Ok(Self::Read),
+ 4 => Ok(Self::Write),
+ 5 => Ok(Self::Create),
+ 6 => Ok(Self::Drop),
+ 7 => Ok(Self::Alter),
+ 8 => Ok(Self::Describe),
+ _ => Err(Error::IllegalArgument {
+ message: format!("Unknown operation type code: {value}"),
+ }),
+ }
+ }
+}
+
+/// Mirrors Java `org.apache.fluss.security.acl.PermissionType`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PermissionType {
+ Any,
+ Allow,
+}
+
+impl PermissionType {
+ pub fn to_i32(self) -> i32 {
+ match self {
+ Self::Any => 1,
+ Self::Allow => 2,
+ }
+ }
+
+ pub fn try_from_i32(value: i32) -> Result<Self> {
+ match value {
+ 1 => Ok(Self::Any),
+ 2 => Ok(Self::Allow),
+ _ => Err(Error::IllegalArgument {
+ message: format!("Unknown permission type: {value}"),
+ }),
+ }
+ }
+}
+
+/// Mirrors Java `org.apache.fluss.security.acl.AclBinding` (concrete ACL entry).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AclInfo {
+ pub resource_name: String,
+ pub resource_type: ResourceType,
+ pub principal_name: String,
+ pub principal_type: String,
+ pub host: String,
+ pub operation_type: OperationType,
+ pub permission_type: PermissionType,
+}
+
+impl AclInfo {
+ pub fn to_pb(&self) -> PbAclInfo {
+ PbAclInfo {
+ resource_name: self.resource_name.clone(),
+ resource_type: self.resource_type.to_i32(),
+ principal_name: self.principal_name.clone(),
+ principal_type: self.principal_type.clone(),
+ host: self.host.clone(),
+ operation_type: self.operation_type.to_i32(),
+ permission_type: self.permission_type.to_i32(),
+ }
+ }
+
+ pub fn from_pb(pb: &PbAclInfo) -> Result<Self> {
+ Ok(Self {
+ resource_name: pb.resource_name.clone(),
+ resource_type: ResourceType::try_from_i32(pb.resource_type)?,
+ principal_name: pb.principal_name.clone(),
+ principal_type: pb.principal_type.clone(),
+ host: pb.host.clone(),
+ operation_type: OperationType::try_from_i32(pb.operation_type)?,
+ permission_type: PermissionType::try_from_i32(pb.permission_type)?,
+ })
+ }
+}
+
+/// Mirrors Java `org.apache.fluss.security.acl.AclBindingFilter` — like `AclInfo` but with
+/// optional name/principal/host fields for partial matching.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AclFilter {
+ pub resource_name: Option<String>,
+ pub resource_type: ResourceType,
+ pub principal_name: Option<String>,
+ pub principal_type: Option<String>,
+ pub host: Option<String>,
+ pub operation_type: OperationType,
+ pub permission_type: PermissionType,
+}
+
+impl AclFilter {
+ pub fn to_pb(&self) -> PbAclFilter {
+ PbAclFilter {
+ resource_name: self.resource_name.clone(),
+ resource_type: self.resource_type.to_i32(),
+ principal_name: self.principal_name.clone(),
+ principal_type: self.principal_type.clone(),
+ host: self.host.clone(),
+ operation_type: self.operation_type.to_i32(),
+ permission_type: self.permission_type.to_i32(),
+ }
+ }
+
+ pub fn from_pb(pb: &PbAclFilter) -> Result<Self> {
+ Ok(Self {
+ resource_name: pb.resource_name.clone(),
+ resource_type: ResourceType::try_from_i32(pb.resource_type)?,
+ principal_name: pb.principal_name.clone(),
+ principal_type: pb.principal_type.clone(),
+ host: pb.host.clone(),
+ operation_type: OperationType::try_from_i32(pb.operation_type)?,
+ permission_type: PermissionType::try_from_i32(pb.permission_type)?,
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_resource_type_roundtrip() {
+ for v in [
+ ResourceType::Any,
+ ResourceType::Cluster,
+ ResourceType::Database,
+ ResourceType::Table,
+ ] {
+ assert_eq!(ResourceType::try_from_i32(v.to_i32()).unwrap(), v);
+ }
+ assert!(ResourceType::try_from_i32(0).is_err());
+ }
+
+ #[test]
+ fn test_operation_type_roundtrip() {
+ for v in [
+ OperationType::Any,
+ OperationType::All,
+ OperationType::Read,
+ OperationType::Write,
+ OperationType::Create,
+ OperationType::Drop,
+ OperationType::Alter,
+ OperationType::Describe,
+ ] {
+ assert_eq!(OperationType::try_from_i32(v.to_i32()).unwrap(), v);
+ }
+ assert!(OperationType::try_from_i32(0).is_err());
+ }
+
+ #[test]
+ fn test_permission_type_roundtrip() {
+ for v in [PermissionType::Any, PermissionType::Allow] {
+ assert_eq!(PermissionType::try_from_i32(v.to_i32()).unwrap(), v);
+ }
+ assert!(PermissionType::try_from_i32(0).is_err());
+ }
+
+ #[test]
+ fn test_acl_info_pb_roundtrip() {
+ let original = AclInfo {
+ resource_name: "topic-a".to_string(),
+ resource_type: ResourceType::Table,
+ principal_name: "alice".to_string(),
+ principal_type: "User".to_string(),
+ host: "*".to_string(),
+ operation_type: OperationType::Read,
+ permission_type: PermissionType::Allow,
+ };
+ let pb = original.to_pb();
+ assert_eq!(AclInfo::from_pb(&pb).unwrap(), original);
+ }
+
+ #[test]
+ fn test_acl_filter_pb_roundtrip() {
+ let original = AclFilter {
+ resource_name: None,
+ resource_type: ResourceType::Any,
+ principal_name: Some("alice".to_string()),
+ principal_type: None,
+ host: Some("*".to_string()),
+ operation_type: OperationType::Any,
+ permission_type: PermissionType::Any,
+ };
+ let pb = original.to_pb();
+ assert_eq!(AclFilter::from_pb(&pb).unwrap(), original);
+ }
+}
diff --git a/crates/fluss/src/metadata/config.rs b/crates/fluss/src/metadata/config.rs
new file mode 100644
index 00000000..4e045c45
--- /dev/null
+++ b/crates/fluss/src/metadata/config.rs
@@ -0,0 +1,124 @@
+// 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::error::{Error, Result};
+use crate::proto::PbAlterConfig;
+
+/// Mirrors Java `org.apache.fluss.config.cluster.AlterConfigOpType`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AlterConfigOpType {
+ Set,
+ Delete,
+ Append,
+ Subtract,
+}
+
+impl AlterConfigOpType {
+ pub fn to_i32(self) -> i32 {
+ match self {
+ Self::Set => 0,
+ Self::Delete => 1,
+ Self::Append => 2,
+ Self::Subtract => 3,
+ }
+ }
+
+ pub fn try_from_i32(value: i32) -> Result<Self> {
+ match value {
+ 0 => Ok(Self::Set),
+ 1 => Ok(Self::Delete),
+ 2 => Ok(Self::Append),
+ 3 => Ok(Self::Subtract),
+ _ => Err(Error::IllegalArgument {
+ message: format!("Unsupported AlterConfigOpType: {value}"),
+ }),
+ }
+ }
+}
+
+/// Mirrors Java `org.apache.fluss.config.cluster.AlterConfig`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AlterConfig {
+ pub config_key: String,
+ pub config_value: Option<String>,
+ pub op_type: AlterConfigOpType,
+}
+
+impl AlterConfig {
+ pub fn new<K: Into<String>>(
+ config_key: K,
+ config_value: Option<String>,
+ op_type: AlterConfigOpType,
+ ) -> Self {
+ Self {
+ config_key: config_key.into(),
+ config_value,
+ op_type,
+ }
+ }
+
+ pub fn to_pb(&self) -> PbAlterConfig {
+ PbAlterConfig {
+ config_key: self.config_key.clone(),
+ config_value: self.config_value.clone(),
+ op_type: self.op_type.to_i32(),
+ }
+ }
+
+ pub fn from_pb(pb: &PbAlterConfig) -> Result<Self> {
+ Ok(Self {
+ config_key: pb.config_key.clone(),
+ config_value: pb.config_value.clone(),
+ op_type: AlterConfigOpType::try_from_i32(pb.op_type)?,
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_alter_config_op_type_roundtrip() {
+ for op in [
+ AlterConfigOpType::Set,
+ AlterConfigOpType::Delete,
+ AlterConfigOpType::Append,
+ AlterConfigOpType::Subtract,
+ ] {
+ assert_eq!(AlterConfigOpType::try_from_i32(op.to_i32()).unwrap(), op);
+ }
+ }
+
+ #[test]
+ fn test_alter_config_op_type_unknown() {
+ assert!(AlterConfigOpType::try_from_i32(99).is_err());
+ }
+
+ #[test]
+ fn test_alter_config_pb_roundtrip() {
+ let cases = [
+ AlterConfig::new("foo", Some("bar".to_string()), AlterConfigOpType::Set),
+ AlterConfig::new("baz", None, AlterConfigOpType::Delete),
+ ];
+ for original in cases {
+ let pb = original.to_pb();
+ let restored = AlterConfig::from_pb(&pb).unwrap();
+ assert_eq!(original, restored);
+ }
+ }
+}
diff --git a/crates/fluss/src/metadata/mod.rs b/crates/fluss/src/metadata/mod.rs
index c1d1b72c..28cda723 100644
--- a/crates/fluss/src/metadata/mod.rs
+++ b/crates/fluss/src/metadata/mod.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+mod acl;
+mod config;
mod data_lake_format;
mod database;
mod datatype;
@@ -22,7 +24,11 @@ mod json_serde;
mod partition;
mod schema_util;
mod table;
+mod table_change;
+mod table_stats;
+pub use acl::*;
+pub use config::*;
pub use data_lake_format::*;
pub use database::*;
pub use datatype::*;
@@ -30,3 +36,5 @@ pub use json_serde::*;
pub use partition::*;
pub(crate) use schema_util::{UNEXIST_MAPPING, index_mapping};
pub use table::*;
+pub use table_change::*;
+pub use table_stats::*;
diff --git a/crates/fluss/src/metadata/table_change.rs b/crates/fluss/src/metadata/table_change.rs
new file mode 100644
index 00000000..2ecf6cfa
--- /dev/null
+++ b/crates/fluss/src/metadata/table_change.rs
@@ -0,0 +1,226 @@
+// 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::error::{Error, Result};
+use crate::proto::{PbAddColumn, PbDropColumn, PbModifyColumn, PbRenameColumn};
+
+/// Mirrors Java `org.apache.fluss.config.cluster.ColumnPositionType`.
+///
+/// The Java server today only handles `Last`; `First` and `After` are reserved by
+/// the proto schema (`column_position_type = 0=LAST, 1=FIRST, 3=AFTER`) for future
+/// use. Sending other variants will be rejected by the server.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ColumnPositionType {
+ Last,
+}
+
+impl ColumnPositionType {
+ pub fn to_i32(self) -> i32 {
+ match self {
+ Self::Last => 0,
+ }
+ }
+
+ pub fn try_from_i32(value: i32) -> Result<Self> {
+ match value {
+ 0 => Ok(Self::Last),
+ _ => Err(Error::IllegalArgument {
+ message: format!("Unsupported ColumnPositionType: {value}"),
+ }),
+ }
+ }
+}
+
+/// Add a column. Mirrors the `AddColumn` variant of Java `TableChange`.
+///
+/// `data_type_json` carries the column's `DataType` already serialized to its JSON
+/// representation (matching the Java client, which transports `DataType` as JSON over
+/// the wire). Callers can produce it via `DataType::serialize_json` + `serde_json::to_vec`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AddColumn {
+ pub column_name: String,
+ pub data_type_json: Vec<u8>,
+ pub comment: Option<String>,
+ pub position: ColumnPositionType,
+}
+
+impl AddColumn {
+ pub fn to_pb(&self) -> PbAddColumn {
+ PbAddColumn {
+ column_name: self.column_name.clone(),
+ data_type_json: self.data_type_json.clone(),
+ comment: self.comment.clone(),
+ column_position_type: self.position.to_i32(),
+ }
+ }
+
+ pub fn from_pb(pb: &PbAddColumn) -> Result<Self> {
+ Ok(Self {
+ column_name: pb.column_name.clone(),
+ data_type_json: pb.data_type_json.clone(),
+ comment: pb.comment.clone(),
+ position: ColumnPositionType::try_from_i32(pb.column_position_type)?,
+ })
+ }
+}
+
+/// Drop a column. Mirrors the `DropColumn` variant of Java `TableChange`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct DropColumn {
+ pub column_name: String,
+}
+
+impl DropColumn {
+ pub fn to_pb(&self) -> PbDropColumn {
+ PbDropColumn {
+ column_name: self.column_name.clone(),
+ }
+ }
+
+ pub fn from_pb(pb: &PbDropColumn) -> Self {
+ Self {
+ column_name: pb.column_name.clone(),
+ }
+ }
+}
+
+/// Rename a column. Mirrors the `RenameColumn` variant of Java `TableChange`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RenameColumn {
+ pub old_column_name: String,
+ pub new_column_name: String,
+}
+
+impl RenameColumn {
+ pub fn to_pb(&self) -> PbRenameColumn {
+ PbRenameColumn {
+ old_column_name: self.old_column_name.clone(),
+ new_column_name: self.new_column_name.clone(),
+ }
+ }
+
+ pub fn from_pb(pb: &PbRenameColumn) -> Self {
+ Self {
+ old_column_name: pb.old_column_name.clone(),
+ new_column_name: pb.new_column_name.clone(),
+ }
+ }
+}
+
+/// Modify a column's type/comment/position. Mirrors the `ModifyColumn` variant of
+/// Java `TableChange`. All fields except `column_name` are optional — only the
+/// non-`None` ones are applied.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ModifyColumn {
+ pub column_name: String,
+ pub data_type_json: Option<Vec<u8>>,
+ pub comment: Option<String>,
+ pub position: Option<ColumnPositionType>,
+}
+
+impl ModifyColumn {
+ pub fn to_pb(&self) -> PbModifyColumn {
+ PbModifyColumn {
+ column_name: self.column_name.clone(),
+ data_type_json: self.data_type_json.clone(),
+ comment: self.comment.clone(),
+ column_position_type: self.position.map(|p| p.to_i32()),
+ }
+ }
+
+ pub fn from_pb(pb: &PbModifyColumn) -> Result<Self> {
+ Ok(Self {
+ column_name: pb.column_name.clone(),
+ data_type_json: pb.data_type_json.clone(),
+ comment: pb.comment.clone(),
+ position: pb
+ .column_position_type
+ .map(ColumnPositionType::try_from_i32)
+ .transpose()?,
+ })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_column_position_type_roundtrip() {
+ assert_eq!(
+ ColumnPositionType::try_from_i32(ColumnPositionType::Last.to_i32()).unwrap(),
+ ColumnPositionType::Last
+ );
+ assert!(ColumnPositionType::try_from_i32(1).is_err());
+ assert!(ColumnPositionType::try_from_i32(3).is_err());
+ }
+
+ #[test]
+ fn test_add_column_pb_roundtrip() {
+ let original = AddColumn {
+ column_name: "amount".to_string(),
+ data_type_json: b"{\"type\":\"BIGINT\"}".to_vec(),
+ comment: Some("the amount".to_string()),
+ position: ColumnPositionType::Last,
+ };
+ let pb = original.to_pb();
+ assert_eq!(AddColumn::from_pb(&pb).unwrap(), original);
+ }
+
+ #[test]
+ fn test_drop_column_pb_roundtrip() {
+ let original = DropColumn {
+ column_name: "obsolete".to_string(),
+ };
+ let pb = original.to_pb();
+ assert_eq!(DropColumn::from_pb(&pb), original);
+ }
+
+ #[test]
+ fn test_rename_column_pb_roundtrip() {
+ let original = RenameColumn {
+ old_column_name: "old_name".to_string(),
+ new_column_name: "new_name".to_string(),
+ };
+ let pb = original.to_pb();
+ assert_eq!(RenameColumn::from_pb(&pb), original);
+ }
+
+ #[test]
+ fn test_modify_column_pb_roundtrip_full() {
+ let original = ModifyColumn {
+ column_name: "x".to_string(),
+ data_type_json: Some(b"{\"type\":\"DOUBLE\"}".to_vec()),
+ comment: Some("changed".to_string()),
+ position: Some(ColumnPositionType::Last),
+ };
+ let pb = original.to_pb();
+ assert_eq!(ModifyColumn::from_pb(&pb).unwrap(), original);
+ }
+
+ #[test]
+ fn test_modify_column_pb_roundtrip_empty() {
+ let original = ModifyColumn {
+ column_name: "x".to_string(),
+ data_type_json: None,
+ comment: None,
+ position: None,
+ };
+ let pb = original.to_pb();
+ assert_eq!(ModifyColumn::from_pb(&pb).unwrap(), original);
+ }
+}
diff --git a/crates/fluss/src/metadata/table_stats.rs b/crates/fluss/src/metadata/table_stats.rs
new file mode 100644
index 00000000..ad2d1fc2
--- /dev/null
+++ b/crates/fluss/src/metadata/table_stats.rs
@@ -0,0 +1,65 @@
+// 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::PbTableStatsReqForBucket;
+
+/// Per-bucket request item for `GetTableStats`.
+/// Mirrors the bucket-stats request shape used by the Java client.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct BucketStatsRequest {
+ pub partition_id: Option<i64>,
+ pub bucket_id: i32,
+}
+
+impl BucketStatsRequest {
+ pub fn new(partition_id: Option<i64>, bucket_id: i32) -> Self {
+ Self {
+ partition_id,
+ bucket_id,
+ }
+ }
+
+ pub fn to_pb(&self) -> PbTableStatsReqForBucket {
+ PbTableStatsReqForBucket {
+ partition_id: self.partition_id,
+ bucket_id: self.bucket_id,
+ }
+ }
+
+ pub fn from_pb(pb: &PbTableStatsReqForBucket) -> Self {
+ Self {
+ partition_id: pb.partition_id,
+ bucket_id: pb.bucket_id,
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_bucket_stats_request_pb_roundtrip() {
+ for req in [
+ BucketStatsRequest::new(None, 0),
+ BucketStatsRequest::new(Some(42), 7),
+ ] {
+ let pb = req.to_pb();
+ assert_eq!(BucketStatsRequest::from_pb(&pb), req);
+ }
+ }
+}
diff --git a/crates/fluss/src/rpc/message/alter_cluster_configs.rs b/crates/fluss/src/rpc/message/alter_cluster_configs.rs
new file mode 100644
index 00000000..fb8ced2d
--- /dev/null
+++ b/crates/fluss/src/rpc/message/alter_cluster_configs.rs
@@ -0,0 +1,47 @@
+// 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::metadata::AlterConfig;
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug)]
+pub struct AlterClusterConfigsRequest {
+ pub inner_request: proto::AlterClusterConfigsRequest,
+}
+
+impl AlterClusterConfigsRequest {
+ pub fn new(alter_configs: Vec<AlterConfig>) -> Self {
+ AlterClusterConfigsRequest {
+ inner_request: proto::AlterClusterConfigsRequest {
+ alter_configs: alter_configs.iter().map(AlterConfig::to_pb).collect(),
+ },
+ }
+ }
+}
+
+impl RequestBody for AlterClusterConfigsRequest {
+ type ResponseBody = proto::AlterClusterConfigsResponse;
+ const API_KEY: ApiKey = ApiKey::AlterClusterConfigs;
+}
+
+impl_write_type!(AlterClusterConfigsRequest);
+impl_read_type!(proto::AlterClusterConfigsResponse);
diff --git a/crates/fluss/src/rpc/message/alter_database.rs b/crates/fluss/src/rpc/message/alter_database.rs
new file mode 100644
index 00000000..dc05f6ed
--- /dev/null
+++ b/crates/fluss/src/rpc/message/alter_database.rs
@@ -0,0 +1,54 @@
+// 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::metadata::AlterConfig;
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug)]
+pub struct AlterDatabaseRequest {
+ pub inner_request: proto::AlterDatabaseRequest,
+}
+
+impl AlterDatabaseRequest {
+ pub fn new(
+ database_name: &str,
+ ignore_if_not_exists: bool,
+ config_changes: Vec<AlterConfig>,
+ ) -> Self {
+ AlterDatabaseRequest {
+ inner_request: proto::AlterDatabaseRequest {
+ database_name: database_name.to_string(),
+ ignore_if_not_exists,
+ config_changes: config_changes.iter().map(AlterConfig::to_pb).collect(),
+ comment: None,
+ },
+ }
+ }
+}
+
+impl RequestBody for AlterDatabaseRequest {
+ type ResponseBody = proto::AlterDatabaseResponse;
+ const API_KEY: ApiKey = ApiKey::AlterDatabase;
+}
+
+impl_write_type!(AlterDatabaseRequest);
+impl_read_type!(proto::AlterDatabaseResponse);
diff --git a/crates/fluss/src/rpc/message/alter_table.rs b/crates/fluss/src/rpc/message/alter_table.rs
new file mode 100644
index 00000000..7cd59ae0
--- /dev/null
+++ b/crates/fluss/src/rpc/message/alter_table.rs
@@ -0,0 +1,62 @@
+// 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::metadata::{AddColumn, AlterConfig, DropColumn, ModifyColumn, RenameColumn, TablePath};
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::convert::to_table_path;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug)]
+pub struct AlterTableRequest {
+ pub inner_request: proto::AlterTableRequest,
+}
+
+impl AlterTableRequest {
+ pub fn new(
+ table_path: &TablePath,
+ ignore_if_not_exists: bool,
+ config_changes: Vec<AlterConfig>,
+ add_columns: Vec<AddColumn>,
+ drop_columns: Vec<DropColumn>,
+ rename_columns: Vec<RenameColumn>,
+ modify_columns: Vec<ModifyColumn>,
+ ) -> Self {
+ AlterTableRequest {
+ inner_request: proto::AlterTableRequest {
+ table_path: to_table_path(table_path),
+ ignore_if_not_exists,
+ config_changes: config_changes.iter().map(AlterConfig::to_pb).collect(),
+ add_columns: add_columns.iter().map(AddColumn::to_pb).collect(),
+ drop_columns: drop_columns.iter().map(DropColumn::to_pb).collect(),
+ rename_columns: rename_columns.iter().map(RenameColumn::to_pb).collect(),
+ modify_columns: modify_columns.iter().map(ModifyColumn::to_pb).collect(),
+ },
+ }
+ }
+}
+
+impl RequestBody for AlterTableRequest {
+ type ResponseBody = proto::AlterTableResponse;
+ const API_KEY: ApiKey = ApiKey::AlterTable;
+}
+
+impl_write_type!(AlterTableRequest);
+impl_read_type!(proto::AlterTableResponse);
diff --git a/crates/fluss/src/rpc/message/create_acls.rs b/crates/fluss/src/rpc/message/create_acls.rs
new file mode 100644
index 00000000..658aa44d
--- /dev/null
+++ b/crates/fluss/src/rpc/message/create_acls.rs
@@ -0,0 +1,47 @@
+// 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::metadata::AclInfo;
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug)]
+pub struct CreateAclsRequest {
+ pub inner_request: proto::CreateAclsRequest,
+}
+
+impl CreateAclsRequest {
+ pub fn new(acl: Vec<AclInfo>) -> Self {
+ CreateAclsRequest {
+ inner_request: proto::CreateAclsRequest {
+ acl: acl.iter().map(AclInfo::to_pb).collect(),
+ },
+ }
+ }
+}
+
+impl RequestBody for CreateAclsRequest {
+ type ResponseBody = proto::CreateAclsResponse;
+ const API_KEY: ApiKey = ApiKey::CreateAcls;
+}
+
+impl_write_type!(CreateAclsRequest);
+impl_read_type!(proto::CreateAclsResponse);
diff --git a/crates/fluss/src/rpc/message/describe_cluster_configs.rs b/crates/fluss/src/rpc/message/describe_cluster_configs.rs
new file mode 100644
index 00000000..d085b3e6
--- /dev/null
+++ b/crates/fluss/src/rpc/message/describe_cluster_configs.rs
@@ -0,0 +1,44 @@
+// 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::rpc::api_key::ApiKey;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug, Default)]
+pub struct DescribeClusterConfigsRequest {
+ pub inner_request: proto::DescribeClusterConfigsRequest,
+}
+
+impl DescribeClusterConfigsRequest {
+ pub fn new() -> Self {
+ DescribeClusterConfigsRequest {
+ inner_request: proto::DescribeClusterConfigsRequest {},
+ }
+ }
+}
+
+impl RequestBody for DescribeClusterConfigsRequest {
+ type ResponseBody = proto::DescribeClusterConfigsResponse;
+ const API_KEY: ApiKey = ApiKey::DescribeClusterConfigs;
+}
+
+impl_write_type!(DescribeClusterConfigsRequest);
+impl_read_type!(proto::DescribeClusterConfigsResponse);
diff --git a/crates/fluss/src/rpc/message/drop_acls.rs b/crates/fluss/src/rpc/message/drop_acls.rs
new file mode 100644
index 00000000..cad27e7b
--- /dev/null
+++ b/crates/fluss/src/rpc/message/drop_acls.rs
@@ -0,0 +1,47 @@
+// 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::metadata::AclFilter;
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug)]
+pub struct DropAclsRequest {
+ pub inner_request: proto::DropAclsRequest,
+}
+
+impl DropAclsRequest {
+ pub fn new(acl_filter: Vec<AclFilter>) -> Self {
+ DropAclsRequest {
+ inner_request: proto::DropAclsRequest {
+ acl_filter: acl_filter.iter().map(AclFilter::to_pb).collect(),
+ },
+ }
+ }
+}
+
+impl RequestBody for DropAclsRequest {
+ type ResponseBody = proto::DropAclsResponse;
+ const API_KEY: ApiKey = ApiKey::DropAcls;
+}
+
+impl_write_type!(DropAclsRequest);
+impl_read_type!(proto::DropAclsResponse);
diff --git a/crates/fluss/src/rpc/message/get_table_stats.rs b/crates/fluss/src/rpc/message/get_table_stats.rs
new file mode 100644
index 00000000..e8ea1165
--- /dev/null
+++ b/crates/fluss/src/rpc/message/get_table_stats.rs
@@ -0,0 +1,53 @@
+// 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::metadata::BucketStatsRequest;
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug)]
+pub struct GetTableStatsRequest {
+ pub inner_request: proto::GetTableStatsRequest,
+}
+
+impl GetTableStatsRequest {
+ pub fn new(
+ table_id: i64,
+ buckets_req: Vec<BucketStatsRequest>,
+ target_columns: Vec<i32>,
+ ) -> Self {
+ GetTableStatsRequest {
+ inner_request: proto::GetTableStatsRequest {
+ table_id,
+ buckets_req: buckets_req.iter().map(BucketStatsRequest::to_pb).collect(),
+ target_columns,
+ },
+ }
+ }
+}
+
+impl RequestBody for GetTableStatsRequest {
+ type ResponseBody = proto::GetTableStatsResponse;
+ const API_KEY: ApiKey = ApiKey::GetTableStats;
+}
+
+impl_write_type!(GetTableStatsRequest);
+impl_read_type!(proto::GetTableStatsResponse);
diff --git a/crates/fluss/src/rpc/message/list_acls.rs b/crates/fluss/src/rpc/message/list_acls.rs
new file mode 100644
index 00000000..68c1177c
--- /dev/null
+++ b/crates/fluss/src/rpc/message/list_acls.rs
@@ -0,0 +1,47 @@
+// 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::metadata::AclFilter;
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::frame::{ReadError, WriteError};
+use crate::rpc::message::{ReadType, RequestBody, WriteType};
+use crate::{impl_read_type, impl_write_type, proto};
+use bytes::{Buf, BufMut};
+use prost::Message;
+
+#[derive(Debug)]
+pub struct ListAclsRequest {
+ pub inner_request: proto::ListAclsRequest,
+}
+
+impl ListAclsRequest {
+ pub fn new(acl_filter: AclFilter) -> Self {
+ ListAclsRequest {
+ inner_request: proto::ListAclsRequest {
+ acl_filter: acl_filter.to_pb(),
+ },
+ }
+ }
+}
+
+impl RequestBody for ListAclsRequest {
+ type ResponseBody = proto::ListAclsResponse;
+ const API_KEY: ApiKey = ApiKey::ListAcls;
+}
+
+impl_write_type!(ListAclsRequest);
+impl_read_type!(proto::ListAclsResponse);
diff --git a/crates/fluss/src/metadata/mod.rs b/crates/fluss/src/rpc/message/list_database_summaries.rs
similarity index 51%
copy from crates/fluss/src/metadata/mod.rs
copy to crates/fluss/src/rpc/message/list_database_summaries.rs
index c1d1b72c..406233fc 100644
--- a/crates/fluss/src/metadata/mod.rs
+++ b/crates/fluss/src/rpc/message/list_database_summaries.rs
@@ -15,18 +15,31 @@
// specific language governing permissions and limitations
// under the License.
-mod data_lake_format;
-mod database;
-mod datatype;
-mod json_serde;
-mod partition;
-mod schema_util;
-mod table;
+use crate::rpc::api_key::ApiKey;
+use crate::rpc::frame::WriteError;
+use crate::rpc::message::{RequestBody, WriteType};
+use crate::{impl_write_type, proto};
+use bytes::BufMut;
+use prost::Message;
-pub use data_lake_format::*;
-pub use database::*;
-pub use datatype::*;
-pub use json_serde::*;
-pub use partition::*;
-pub(crate) use schema_util::{UNEXIST_MAPPING, index_mapping};
-pub use table::*;
+#[derive(Debug, Default)]
+pub struct ListDatabaseSummariesRequest {
+ pub inner_request: proto::ListDatabasesRequest,
+}
+
+impl ListDatabaseSummariesRequest {
+ pub fn new() -> Self {
+ ListDatabaseSummariesRequest {
+ inner_request: proto::ListDatabasesRequest {
+ include_summary: Some(true),
+ },
+ }
+ }
+}
+
+impl RequestBody for ListDatabaseSummariesRequest {
+ type ResponseBody = proto::ListDatabasesResponse;
+ const API_KEY: ApiKey = ApiKey::ListDatabases;
+}
+
+impl_write_type!(ListDatabaseSummariesRequest);
diff --git a/crates/fluss/src/rpc/message/mod.rs b/crates/fluss/src/rpc/message/mod.rs
index 096066ed..f126d214 100644
--- a/crates/fluss/src/rpc/message/mod.rs
+++ b/crates/fluss/src/rpc/message/mod.rs
@@ -19,12 +19,18 @@ use crate::rpc::api_key::ApiKey;
use crate::rpc::frame::{ReadError, WriteError};
use bytes::{Buf, BufMut};
+mod alter_cluster_configs;
+mod alter_database;
+mod alter_table;
mod api_versions;
mod authenticate;
+mod create_acls;
mod create_database;
mod create_partition;
mod create_table;
mod database_exists;
+mod describe_cluster_configs;
+mod drop_acls;
mod drop_database;
mod drop_partition;
mod drop_table;
@@ -34,9 +40,12 @@ mod get_latest_lake_snapshot;
mod get_security_token;
mod get_table;
mod get_table_schema;
+mod get_table_stats;
mod header;
mod init_writer;
mod limit_scan;
+mod list_acls;
+mod list_database_summaries;
mod list_databases;
mod list_offsets;
mod list_partition_infos;
@@ -49,12 +58,18 @@ mod table_exists;
mod update_metadata;
pub use crate::rpc::RpcError;
+pub use alter_cluster_configs::*;
+pub use alter_database::*;
+pub use alter_table::*;
pub use api_versions::*;
pub use authenticate::*;
+pub use create_acls::*;
pub use create_database::*;
pub use create_partition::*;
pub use create_table::*;
pub use database_exists::*;
+pub use describe_cluster_configs::*;
+pub use drop_acls::*;
pub use drop_database::*;
pub use drop_partition::*;
pub use drop_table::*;
@@ -64,9 +79,12 @@ pub use get_latest_lake_snapshot::*;
pub use get_security_token::*;
pub use get_table::*;
pub use get_table_schema::*;
+pub use get_table_stats::*;
pub use header::*;
pub use init_writer::*;
pub use limit_scan::*;
+pub use list_acls::*;
+pub use list_database_summaries::*;
pub use list_databases::*;
pub use list_offsets::*;
pub use list_partition_infos::*;