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
The following commit(s) were added to refs/heads/main by this push:
new 8c108ba4 [rust] Sync Fluss 1.x proto definitions and ApiKey registry
(#628)
8c108ba4 is described below
commit 8c108ba408f8be17feff1c3d94a08af052438d57
Author: Pengcheng Huang <[email protected]>
AuthorDate: Fri Jun 19 21:12:18 2026 +0800
[rust] Sync Fluss 1.x proto definitions and ApiKey registry (#628)
- Update fluss_api.proto with all 1.x message types (ACLs, KV snapshots,
producer offsets, cluster config, rebalance, server tags, etc.)
- Add optional fields: rack, remote_data_dir, leader_epoch, agg_mode, etc.
- Register 24 new ApiKey variants (1023-1064) in api_key.rs
- Update build.rs prost bytes config for new proto fields
- Add None defaults in convert.rs and partition.rs
- Update pre-existing message wrappers that reference renamed proto fields /
ApiKey variants so the crate still builds:
* create_partition.rs: ignore_if_exists -> ignore_if_not_exists
* get_latest_lake_snapshot.rs: ApiKey::GetLatestLakeSnapshot ->
GetLakeSnapshot
* list_databases.rs: populate new include_summary field
* lookup.rs: PbLookupReqForBucket.key -> keys; new LookupRequest fields
Co-authored-by: warmbupt <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
crates/fluss/build.rs | 3 +-
crates/fluss/src/client/admin.rs | 1 +
crates/fluss/src/client/table/scanner.rs | 5 +
crates/fluss/src/metadata/partition.rs | 1 +
crates/fluss/src/proto/fluss_api.proto | 464 ++++++++++++++++++++-
crates/fluss/src/rpc/api_key.rs | 137 +++++-
crates/fluss/src/rpc/convert.rs | 2 +
crates/fluss/src/rpc/message/create_partition.rs | 2 +-
.../src/rpc/message/get_latest_lake_snapshot.rs | 2 +-
crates/fluss/src/rpc/message/list_databases.rs | 4 +-
crates/fluss/src/rpc/message/lookup.rs | 5 +-
11 files changed, 612 insertions(+), 14 deletions(-)
diff --git a/crates/fluss/build.rs b/crates/fluss/build.rs
index 040ee521..f9248c36 100644
--- a/crates/fluss/build.rs
+++ b/crates/fluss/build.rs
@@ -22,8 +22,9 @@ fn main() -> Result<()> {
config.bytes([
".proto.PbProduceLogReqForBucket.records",
".proto.PbPutKvReqForBucket.records",
- ".proto.PbLookupReqForBucket.key",
+ ".proto.PbLookupReqForBucket.keys",
".proto.PbPrefixLookupReqForBucket.keys",
+ ".proto.ScanKvResponse.records",
]);
config.compile_protos(&["src/proto/fluss_api.proto"], &["src/proto"])?;
Ok(())
diff --git a/crates/fluss/src/client/admin.rs b/crates/fluss/src/client/admin.rs
index 0828b83b..d3d5a5e3 100644
--- a/crates/fluss/src/client/admin.rs
+++ b/crates/fluss/src/client/admin.rs
@@ -150,6 +150,7 @@ impl FlussAdmin {
table_json,
created_time,
modified_time,
+ ..
} = response;
let v: &[u8] = &table_json[..];
let table_descriptor =
diff --git a/crates/fluss/src/client/table/scanner.rs
b/crates/fluss/src/client/table/scanner.rs
index 8578daa4..b36b0e42 100644
--- a/crates/fluss/src/client/table/scanner.rs
+++ b/crates/fluss/src/client/table/scanner.rs
@@ -2058,6 +2058,8 @@ impl LogFetcher {
projection_pushdown_enabled: projection_enabled,
projected_fields: projected_fields.clone(),
buckets_req: feq_for_buckets,
+ filter_predicate: None,
+ filter_schema_id: None,
};
let fetch_log_request = FetchLogRequest {
@@ -2408,6 +2410,7 @@ mod tests {
log_start_offset: None,
remote_log_fetch_info: None,
records: None,
+ filtered_end_offset: None,
}],
}],
};
@@ -2464,6 +2467,7 @@ mod tests {
log_start_offset: None,
remote_log_fetch_info: None,
records: None,
+ filtered_end_offset: None,
}],
}],
};
@@ -2801,6 +2805,7 @@ mod tests {
log_start_offset: Some(0),
remote_log_fetch_info: None,
records: None,
+ filtered_end_offset: None,
}],
}],
};
diff --git a/crates/fluss/src/metadata/partition.rs
b/crates/fluss/src/metadata/partition.rs
index 18402354..c63fe296 100644
--- a/crates/fluss/src/metadata/partition.rs
+++ b/crates/fluss/src/metadata/partition.rs
@@ -300,6 +300,7 @@ impl PartitionInfo {
PbPartitionInfo {
partition_id: self.partition_id,
partition_spec: self.partition_spec.to_pb(),
+ remote_data_dir: None,
}
}
diff --git a/crates/fluss/src/proto/fluss_api.proto
b/crates/fluss/src/proto/fluss_api.proto
index 2add80d7..7d7fd89e 100644
--- a/crates/fluss/src/proto/fluss_api.proto
+++ b/crates/fluss/src/proto/fluss_api.proto
@@ -94,6 +94,7 @@ message PbServerNode {
required string host = 2;
required int32 port = 3;
optional string listeners = 4;
+ optional string rack = 5;
}
message PbTableMetadata {
@@ -104,6 +105,7 @@ message PbTableMetadata {
repeated PbBucketMetadata bucket_metadata = 5;
required int64 created_time = 6;
required int64 modified_time = 7;
+ optional string remote_data_dir = 8;
}
message PbPartitionMetadata {
@@ -119,7 +121,7 @@ message PbBucketMetadata {
// optional as some time the leader may not elected yet
optional int32 leader_id = 2;
repeated int32 replica_id = 3 [packed = true];
- // TODO: Add isr here.
+ optional int32 leader_epoch = 4;
}
message PbProduceLogReqForBucket {
@@ -145,6 +147,7 @@ message PutKvRequest {
// if empty, means write all columns
repeated int32 target_columns = 4 [packed = true];
repeated PbPutKvReqForBucket buckets_req = 5;
+ optional int32 agg_mode = 6;
}
message PutKvResponse {
@@ -162,6 +165,7 @@ message PbPutKvRespForBucket {
required int32 bucket_id = 2;
optional int32 error_code = 3;
optional string error_message = 4;
+ optional int64 log_end_offset = 5;
}
message CreateTableRequest {
@@ -199,6 +203,7 @@ message GetTableInfoResponse {
required bytes table_json = 3;
required int64 created_time = 4;
required int64 modified_time = 5;
+ optional string remote_data_dir = 6;
}
// get table schema request and response. Mirrors the Java RPC at api key 1011.
@@ -258,10 +263,12 @@ message DatabaseExistsResponse {
}
message ListDatabasesRequest {
+ optional bool include_summary = 1;
}
message ListDatabasesResponse {
repeated string database_name = 1;
+ repeated PbDatabaseSummary database_summary = 2;
}
// list offsets request and response
@@ -291,11 +298,46 @@ message FetchLogResponse {
repeated PbFetchLogRespForTable tables_resp = 1;
}
+message PbPredicate {
+ required int32 type = 1;
+ optional PbLeafPredicate leaf = 2;
+ optional PbCompoundPredicate compound = 3;
+}
+
+message PbLeafPredicate {
+ required int32 function = 1;
+ required int32 field_id = 2;
+ repeated PbLiteralValue literals = 3;
+}
+
+message PbCompoundPredicate {
+ required int32 function = 1;
+ repeated PbPredicate children = 2;
+}
+
+message PbLiteralValue {
+ required int32 literal_type = 1;
+ required bool is_null = 2;
+ optional bool boolean_value = 3;
+ optional int32 int_value = 4;
+ optional int64 bigint_value = 5;
+ optional float float_value = 6;
+ optional double double_value = 7;
+ optional string string_value = 8;
+ optional bytes binary_value = 9;
+ optional int64 decimal_value = 10;
+ optional int64 timestamp_millis_value = 11;
+ optional int32 timestamp_nano_of_millis_value = 12;
+ optional bytes decimal_bytes = 13;
+}
+
message PbFetchLogReqForTable {
required int64 table_id = 1;
required bool projection_pushdown_enabled = 2;
repeated int32 projected_fields = 3 [packed = true];
repeated PbFetchLogReqForBucket buckets_req = 4;
+ optional PbPredicate filter_predicate = 5;
+ optional int32 filter_schema_id = 6;
}
@@ -321,6 +363,7 @@ message PbFetchLogRespForBucket {
optional int64 log_start_offset = 6; // TODO now we don't introduce log
start offset, but remain it in protobuf
optional PbRemoteLogFetchInfo remote_log_fetch_info = 7;
optional bytes records = 8;
+ optional int64 filtered_end_offset = 9;
}
message PbRemoteLogFetchInfo {
@@ -360,6 +403,7 @@ message PbLakeSnapshotForBucket {
optional int64 partition_id = 1;
required int32 bucket_id = 2;
optional int64 log_offset = 3;
+ optional string partition_name = 4;
}
message PbKeyValue {
@@ -381,6 +425,9 @@ message GetFileSystemSecurityTokenResponse {
message LookupRequest {
required int64 table_id = 1;
repeated PbLookupReqForBucket buckets_req = 2;
+ optional bool insert_if_not_exists = 3;
+ optional int32 acks = 4;
+ optional int32 timeout_ms = 5;
}
message LookupResponse {
@@ -390,7 +437,7 @@ message LookupResponse {
message PbLookupReqForBucket {
optional int64 partition_id = 1;
required int32 bucket_id = 2;
- repeated bytes key = 3;
+ repeated bytes keys = 3;
}
message PbLookupRespForBucket {
@@ -440,6 +487,7 @@ message PbPartitionSpec {
message PbPartitionInfo {
required int64 partition_id = 1;
required PbPartitionSpec partition_spec = 2;
+ optional string remote_data_dir = 3;
}
message ListPartitionInfosRequest {
@@ -454,7 +502,7 @@ message ListPartitionInfosResponse {
message CreatePartitionRequest {
required PbTablePath table_path = 1;
required PbPartitionSpec partition_spec = 2;
- required bool ignore_if_exists = 3;
+ required bool ignore_if_not_exists = 3;
}
message CreatePartitionResponse {}
@@ -501,3 +549,413 @@ message InitWriterRequest {
message InitWriterResponse {
required int64 writer_id = 1;
}
+
+message PbDatabaseSummary {
+ required string database_name = 1;
+ required int64 created_time = 2;
+ required int32 table_count = 3;
+}
+
+message AlterDatabaseRequest {
+ required string database_name = 1;
+ required bool ignore_if_not_exists = 2;
+ repeated PbAlterConfig config_changes = 3;
+ optional string comment = 4;
+}
+
+message AlterDatabaseResponse {
+}
+
+message AlterTableRequest {
+ required PbTablePath table_path = 1;
+ required bool ignore_if_not_exists = 2;
+ repeated PbAlterConfig config_changes = 3;
+ repeated PbAddColumn add_columns = 4;
+ repeated PbDropColumn drop_columns = 5;
+ repeated PbRenameColumn rename_columns = 6;
+ repeated PbModifyColumn modify_columns = 7;
+}
+
+message AlterTableResponse {
+}
+
+message PbAlterConfig {
+ required string config_key = 1;
+ optional string config_value = 2;
+ required int32 op_type = 3;
+}
+
+message PbAddColumn {
+ required string column_name = 1;
+ required bytes data_type_json = 2;
+ optional string comment = 3;
+ required int32 column_position_type = 4;
+}
+
+message PbDropColumn {
+ required string column_name = 1;
+}
+
+message PbRenameColumn {
+ required string old_column_name = 1;
+ required string new_column_name = 2;
+}
+
+message PbModifyColumn {
+ required string column_name = 1;
+ optional bytes data_type_json = 2;
+ optional string comment = 3;
+ optional int32 column_position_type = 4;
+}
+
+message GetTableStatsRequest {
+ required int64 table_id = 1;
+ repeated PbTableStatsReqForBucket buckets_req = 2;
+ repeated int32 target_columns = 3 [packed = true];
+}
+
+message GetTableStatsResponse {
+ repeated PbTableStatsRespForBucket buckets_resp = 1;
+}
+
+message PbTableStatsReqForBucket {
+ optional int64 partition_id = 1;
+ required int32 bucket_id = 2;
+}
+
+message PbTableStatsRespForBucket {
+ optional int32 error_code = 1;
+ optional string error_message = 2;
+ optional int64 partition_id = 3;
+ required int32 bucket_id = 4;
+ optional int64 row_count = 5;
+}
+
+message GetLatestKvSnapshotsRequest {
+ required PbTablePath table_path = 1;
+ optional string partition_name = 2;
+}
+
+message GetLatestKvSnapshotsResponse {
+ required int64 table_id = 1;
+ optional int64 partition_id = 2;
+ repeated PbKvSnapshot latest_snapshots = 3;
+}
+
+message PbKvSnapshot {
+ required int32 bucket_id = 1;
+ optional int64 snapshot_id = 2;
+ optional int64 log_offset = 3;
+}
+
+message GetKvSnapshotMetadataRequest {
+ required int64 table_id = 1;
+ optional int64 partition_id = 2;
+ required int32 bucket_id = 3;
+ required int64 snapshot_id = 4;
+}
+
+message GetKvSnapshotMetadataResponse {
+ required int64 log_offset = 1;
+ repeated PbRemotePathAndLocalFile snapshot_files = 2;
+}
+
+message PbRemotePathAndLocalFile {
+ required string remote_path = 1;
+ required string local_file_name = 2;
+}
+
+message AcquireKvSnapshotLeaseRequest {
+ required string lease_id = 1;
+ required int64 lease_duration_ms = 2;
+ repeated PbKvSnapshotLeaseForTable snapshots_to_lease = 3;
+}
+
+message AcquireKvSnapshotLeaseResponse {
+ repeated PbKvSnapshotLeaseForTable unavailable_snapshots = 1;
+}
+
+message PbKvSnapshotLeaseForTable {
+ required int64 table_id = 1;
+ repeated PbKvSnapshotLeaseForBucket bucket_snapshots = 2;
+}
+
+message PbKvSnapshotLeaseForBucket {
+ optional int64 partition_id = 1;
+ required int32 bucket_id = 2;
+ required int64 snapshot_id = 3;
+}
+
+message GetLakeSnapshotRequest {
+ required PbTablePath table_path = 1;
+ optional int64 snapshot_id = 2;
+ optional bool readable = 3;
+}
+
+message GetLakeSnapshotResponse {
+ required int64 table_id = 1;
+ required int64 snapshotId = 2;
+ repeated PbLakeSnapshotForBucket bucket_snapshots = 3;
+}
+
+message PbAclInfo {
+ required string resource_name = 1;
+ required int32 resource_type = 2;
+ required string principal_name = 3;
+ required string principal_type = 4;
+ required string host = 5;
+ required int32 operation_type = 6;
+ required int32 permission_type = 7;
+}
+
+message PbAclFilter {
+ optional string resource_name = 1;
+ required int32 resource_type = 2;
+ optional string principal_name = 3;
+ optional string principal_type = 4;
+ optional string host = 5;
+ required int32 operation_type = 6;
+ required int32 permission_type = 7;
+}
+
+message PbCreateAclRespInfo {
+ required PbAclInfo acl = 1;
+ optional int32 error_code = 2;
+ optional string error_message = 3;
+}
+
+message PbDropAclsFilterResult {
+ repeated PbDropAclsMatchingAcl matching_acls = 1;
+ optional int32 error_code = 2;
+ optional string error_message = 3;
+}
+
+message PbDropAclsMatchingAcl {
+ required PbAclInfo acl = 1;
+ optional int32 error_code = 2;
+ optional string error_message = 3;
+}
+
+message ListAclsRequest {
+ required PbAclFilter acl_filter = 1;
+}
+
+message ListAclsResponse {
+ repeated PbAclInfo acl = 1;
+}
+
+message CreateAclsRequest {
+ repeated PbAclInfo acl = 1;
+}
+
+message CreateAclsResponse {
+ repeated PbCreateAclRespInfo aclRes = 1;
+}
+
+message DropAclsRequest {
+ repeated PbAclFilter acl_filter = 1;
+}
+
+message DropAclsResponse {
+ repeated PbDropAclsFilterResult filter_results = 1;
+}
+
+message PbDescribeConfig {
+ required string config_key = 1;
+ optional string config_value = 2;
+ required string config_source = 3;
+}
+
+message DescribeClusterConfigsRequest {
+}
+
+message DescribeClusterConfigsResponse {
+ repeated PbDescribeConfig configs = 1;
+}
+
+message AlterClusterConfigsRequest {
+ repeated PbAlterConfig alter_configs = 1;
+}
+
+message AlterClusterConfigsResponse {
+}
+
+message AddServerTagRequest {
+ repeated int32 server_ids = 1 [packed = true];
+ required int32 server_tag = 2;
+}
+
+message AddServerTagResponse {
+}
+
+message RemoveServerTagRequest {
+ repeated int32 server_ids = 1 [packed = true];
+ required int32 server_tag = 2;
+}
+
+message RemoveServerTagResponse {
+}
+
+message RebalanceRequest {
+ repeated int32 goals = 1 [packed = true];
+}
+
+message RebalanceResponse {
+ required string rebalance_id = 1;
+}
+
+message ListRebalanceProgressRequest {
+ optional string rebalance_id = 1;
+}
+
+message ListRebalanceProgressResponse {
+ optional string rebalance_id = 1;
+ optional int32 rebalance_status = 2;
+ repeated PbRebalanceProgressForTable table_progress = 3;
+}
+
+message PbRebalanceProgressForTable {
+ required int64 table_id = 1;
+ repeated PbRebalanceProgressForBucket buckets_progress = 2;
+}
+
+message PbRebalanceProgressForBucket {
+ required PbRebalancePlanForBucket rebalance_plan = 1;
+ required int32 rebalance_status = 2;
+}
+
+message PbRebalancePlanForBucket {
+ optional int64 partition_id = 1;
+ required int32 bucket_id = 2;
+ optional int32 original_leader = 3;
+ optional int32 new_leader = 4;
+ repeated int32 original_replicas = 5 [packed = true];
+ repeated int32 new_replicas = 6 [packed = true];
+}
+
+message CancelRebalanceRequest {
+ optional string rebalance_id = 1;
+}
+
+message CancelRebalanceResponse {
+}
+
+message PbBucketOffset {
+ optional int64 partition_id = 1;
+ required int32 bucket_id = 2;
+ optional int64 log_end_offset = 4;
+}
+
+message PbProducerTableOffsets {
+ required int64 table_id = 1;
+ repeated PbBucketOffset bucket_offsets = 2;
+}
+
+message RegisterProducerOffsetsRequest {
+ required string producer_id = 1;
+ repeated PbProducerTableOffsets table_offsets = 2;
+ optional int64 ttl_ms = 3;
+}
+
+message RegisterProducerOffsetsResponse {
+ optional int32 result = 1;
+}
+
+message GetProducerOffsetsRequest {
+ required string producer_id = 1;
+}
+
+message GetProducerOffsetsResponse {
+ optional string producer_id = 1;
+ optional int64 expiration_time = 2;
+ repeated PbProducerTableOffsets table_offsets = 3;
+}
+
+message DeleteProducerOffsetsRequest {
+ required string producer_id = 1;
+}
+
+message DeleteProducerOffsetsResponse {
+}
+
+message PbTableBucket {
+ required int64 table_id = 1;
+ optional int64 partition_id = 2;
+ required int32 bucket_id = 3;
+}
+
+message ReleaseKvSnapshotLeaseRequest {
+ required string lease_id = 1;
+ repeated PbTableBucket buckets_to_release = 2;
+}
+
+message ReleaseKvSnapshotLeaseResponse {
+}
+
+message DropKvSnapshotLeaseRequest {
+ required string lease_id = 1;
+}
+
+message DropKvSnapshotLeaseResponse {
+}
+
+message PbScanReqForBucket {
+ required int64 table_id = 1;
+ optional int64 partition_id = 2;
+ required int32 bucket_id = 3;
+ optional int64 limit = 4;
+}
+
+message ScanKvRequest {
+ optional bytes scanner_id = 1;
+ optional PbScanReqForBucket bucket_scan_req = 2;
+ optional int32 call_seq_id = 3;
+ optional int32 batch_size_bytes = 4;
+ optional bool close_scanner = 5;
+}
+
+message ScanKvResponse {
+ optional int32 error_code = 1;
+ optional string error_message = 2;
+ optional bytes scanner_id = 3;
+ optional bool has_more_results = 4;
+ optional bytes records = 5;
+ optional int64 log_offset = 6;
+}
+
+message GetClusterHealthRequest {
+}
+
+message GetClusterHealthResponse {
+ required int32 num_replicas = 1;
+ required int32 in_sync_replicas = 2;
+ required int32 num_leader_replicas = 3;
+ required int32 active_leader_replicas = 4;
+ required int32 status = 5;
+}
+
+message PbRemoteLogManifestEntry {
+ required PbTableBucket table_bucket = 1;
+ required string remote_log_manifest_path = 2;
+ required int64 remote_log_end_offset = 3;
+}
+
+message ListRemoteLogManifestsRequest {
+ required int64 table_id = 1;
+ optional int64 partition_id = 2;
+}
+
+message ListRemoteLogManifestsResponse {
+ repeated PbRemoteLogManifestEntry manifests = 1;
+}
+
+message ListKvSnapshotsRequest {
+ required int64 table_id = 1;
+ optional int64 partition_id = 2;
+}
+
+message ListKvSnapshotsResponse {
+ required int64 table_id = 1;
+ optional int64 partition_id = 2;
+ repeated PbKvSnapshot active_snapshots = 3;
+}
diff --git a/crates/fluss/src/rpc/api_key.rs b/crates/fluss/src/rpc/api_key.rs
index aaeca70d..cab034d2 100644
--- a/crates/fluss/src/rpc/api_key.rs
+++ b/crates/fluss/src/rpc/api_key.rs
@@ -38,15 +38,40 @@ pub enum ApiKey {
PutKv, // 1016
Lookup, // 1017
ListOffsets, // 1021
+ GetLatestKvSnapshots, // 1023
+ GetKvSnapshotMetadata, // 1024
GetFileSystemSecurityToken, // 1025
InitWriter, // 1026
- GetLatestLakeSnapshot, // 1032
+ GetLakeSnapshot, // 1032
LimitScan, // 1033
PrefixLookup, // 1034
GetDatabaseInfo, // 1035
CreatePartition, // 1036
DropPartition, // 1037
Authenticate, // 1038
+ CreateAcls, // 1039
+ ListAcls, // 1040
+ DropAcls, // 1041
+ AlterTable, // 1044
+ DescribeClusterConfigs, // 1045
+ AlterClusterConfigs, // 1046
+ AddServerTag, // 1047
+ RemoveServerTag, // 1048
+ Rebalance, // 1049
+ ListRebalanceProgress, // 1050
+ CancelRebalance, // 1051
+ RegisterProducerOffsets, // 1053
+ GetProducerOffsets, // 1054
+ DeleteProducerOffsets, // 1055
+ AcquireKvSnapshotLease, // 1056
+ ReleaseKvSnapshotLease, // 1057
+ DropKvSnapshotLease, // 1058
+ GetTableStats, // 1059
+ AlterDatabase, // 1060
+ ScanKv, // 1061
+ GetClusterHealth, // 1062
+ ListRemoteLogManifests, // 1063
+ ListKvSnapshots, // 1064
Unknown(i16),
}
@@ -71,14 +96,39 @@ impl ApiKey {
| ApiKey::ProduceLog
| ApiKey::FetchLog
| ApiKey::ListOffsets
+ | ApiKey::GetLatestKvSnapshots
+ | ApiKey::GetKvSnapshotMetadata
| ApiKey::GetFileSystemSecurityToken
| ApiKey::InitWriter
- | ApiKey::GetLatestLakeSnapshot
+ | ApiKey::GetLakeSnapshot
| ApiKey::LimitScan
| ApiKey::GetDatabaseInfo
| ApiKey::CreatePartition
| ApiKey::DropPartition
- | ApiKey::Authenticate => Some(ApiVersionRange::new(ApiVersion(0),
ApiVersion(0))),
+ | ApiKey::Authenticate
+ | ApiKey::CreateAcls
+ | ApiKey::ListAcls
+ | ApiKey::DropAcls
+ | ApiKey::AlterTable
+ | ApiKey::DescribeClusterConfigs
+ | ApiKey::AlterClusterConfigs
+ | ApiKey::AddServerTag
+ | ApiKey::RemoveServerTag
+ | ApiKey::Rebalance
+ | ApiKey::ListRebalanceProgress
+ | ApiKey::CancelRebalance
+ | ApiKey::RegisterProducerOffsets
+ | ApiKey::GetProducerOffsets
+ | ApiKey::DeleteProducerOffsets
+ | ApiKey::AcquireKvSnapshotLease
+ | ApiKey::ReleaseKvSnapshotLease
+ | ApiKey::DropKvSnapshotLease
+ | ApiKey::GetTableStats
+ | ApiKey::AlterDatabase
+ | ApiKey::ScanKv
+ | ApiKey::GetClusterHealth
+ | ApiKey::ListRemoteLogManifests
+ | ApiKey::ListKvSnapshots =>
Some(ApiVersionRange::new(ApiVersion(0), ApiVersion(0))),
// PutKv / Lookup / PrefixLookup support v0 (legacy key encoding)
// and v1 (Paimon BinaryRow key encoding for kv_format_version=2
// non-default bucket keys). The Rust client encodes both.
@@ -111,15 +161,40 @@ impl From<i16> for ApiKey {
1016 => ApiKey::PutKv,
1017 => ApiKey::Lookup,
1021 => ApiKey::ListOffsets,
+ 1023 => ApiKey::GetLatestKvSnapshots,
+ 1024 => ApiKey::GetKvSnapshotMetadata,
1025 => ApiKey::GetFileSystemSecurityToken,
1026 => ApiKey::InitWriter,
- 1032 => ApiKey::GetLatestLakeSnapshot,
+ 1032 => ApiKey::GetLakeSnapshot,
1033 => ApiKey::LimitScan,
1034 => ApiKey::PrefixLookup,
1035 => ApiKey::GetDatabaseInfo,
1036 => ApiKey::CreatePartition,
1037 => ApiKey::DropPartition,
1038 => ApiKey::Authenticate,
+ 1039 => ApiKey::CreateAcls,
+ 1040 => ApiKey::ListAcls,
+ 1041 => ApiKey::DropAcls,
+ 1044 => ApiKey::AlterTable,
+ 1045 => ApiKey::DescribeClusterConfigs,
+ 1046 => ApiKey::AlterClusterConfigs,
+ 1047 => ApiKey::AddServerTag,
+ 1048 => ApiKey::RemoveServerTag,
+ 1049 => ApiKey::Rebalance,
+ 1050 => ApiKey::ListRebalanceProgress,
+ 1051 => ApiKey::CancelRebalance,
+ 1053 => ApiKey::RegisterProducerOffsets,
+ 1054 => ApiKey::GetProducerOffsets,
+ 1055 => ApiKey::DeleteProducerOffsets,
+ 1056 => ApiKey::AcquireKvSnapshotLease,
+ 1057 => ApiKey::ReleaseKvSnapshotLease,
+ 1058 => ApiKey::DropKvSnapshotLease,
+ 1059 => ApiKey::GetTableStats,
+ 1060 => ApiKey::AlterDatabase,
+ 1061 => ApiKey::ScanKv,
+ 1062 => ApiKey::GetClusterHealth,
+ 1063 => ApiKey::ListRemoteLogManifests,
+ 1064 => ApiKey::ListKvSnapshots,
_ => Unknown(key),
}
@@ -147,15 +222,40 @@ impl From<ApiKey> for i16 {
ApiKey::PutKv => 1016,
ApiKey::Lookup => 1017,
ApiKey::ListOffsets => 1021,
+ ApiKey::GetLatestKvSnapshots => 1023,
+ ApiKey::GetKvSnapshotMetadata => 1024,
ApiKey::GetFileSystemSecurityToken => 1025,
ApiKey::InitWriter => 1026,
- ApiKey::GetLatestLakeSnapshot => 1032,
+ ApiKey::GetLakeSnapshot => 1032,
ApiKey::LimitScan => 1033,
ApiKey::PrefixLookup => 1034,
ApiKey::GetDatabaseInfo => 1035,
ApiKey::CreatePartition => 1036,
ApiKey::DropPartition => 1037,
ApiKey::Authenticate => 1038,
+ ApiKey::CreateAcls => 1039,
+ ApiKey::ListAcls => 1040,
+ ApiKey::DropAcls => 1041,
+ ApiKey::AlterTable => 1044,
+ ApiKey::DescribeClusterConfigs => 1045,
+ ApiKey::AlterClusterConfigs => 1046,
+ ApiKey::AddServerTag => 1047,
+ ApiKey::RemoveServerTag => 1048,
+ ApiKey::Rebalance => 1049,
+ ApiKey::ListRebalanceProgress => 1050,
+ ApiKey::CancelRebalance => 1051,
+ ApiKey::RegisterProducerOffsets => 1053,
+ ApiKey::GetProducerOffsets => 1054,
+ ApiKey::DeleteProducerOffsets => 1055,
+ ApiKey::AcquireKvSnapshotLease => 1056,
+ ApiKey::ReleaseKvSnapshotLease => 1057,
+ ApiKey::DropKvSnapshotLease => 1058,
+ ApiKey::GetTableStats => 1059,
+ ApiKey::AlterDatabase => 1060,
+ ApiKey::ScanKv => 1061,
+ ApiKey::GetClusterHealth => 1062,
+ ApiKey::ListRemoteLogManifests => 1063,
+ ApiKey::ListKvSnapshots => 1064,
Unknown(x) => x,
}
}
@@ -186,15 +286,40 @@ mod tests {
(1016, ApiKey::PutKv),
(1017, ApiKey::Lookup),
(1021, ApiKey::ListOffsets),
+ (1023, ApiKey::GetLatestKvSnapshots),
+ (1024, ApiKey::GetKvSnapshotMetadata),
(1025, ApiKey::GetFileSystemSecurityToken),
(1026, ApiKey::InitWriter),
- (1032, ApiKey::GetLatestLakeSnapshot),
+ (1032, ApiKey::GetLakeSnapshot),
(1033, ApiKey::LimitScan),
(1034, ApiKey::PrefixLookup),
(1035, ApiKey::GetDatabaseInfo),
(1036, ApiKey::CreatePartition),
(1037, ApiKey::DropPartition),
(1038, ApiKey::Authenticate),
+ (1039, ApiKey::CreateAcls),
+ (1040, ApiKey::ListAcls),
+ (1041, ApiKey::DropAcls),
+ (1044, ApiKey::AlterTable),
+ (1045, ApiKey::DescribeClusterConfigs),
+ (1046, ApiKey::AlterClusterConfigs),
+ (1047, ApiKey::AddServerTag),
+ (1048, ApiKey::RemoveServerTag),
+ (1049, ApiKey::Rebalance),
+ (1050, ApiKey::ListRebalanceProgress),
+ (1051, ApiKey::CancelRebalance),
+ (1053, ApiKey::RegisterProducerOffsets),
+ (1054, ApiKey::GetProducerOffsets),
+ (1055, ApiKey::DeleteProducerOffsets),
+ (1056, ApiKey::AcquireKvSnapshotLease),
+ (1057, ApiKey::ReleaseKvSnapshotLease),
+ (1058, ApiKey::DropKvSnapshotLease),
+ (1059, ApiKey::GetTableStats),
+ (1060, ApiKey::AlterDatabase),
+ (1061, ApiKey::ScanKv),
+ (1062, ApiKey::GetClusterHealth),
+ (1063, ApiKey::ListRemoteLogManifests),
+ (1064, ApiKey::ListKvSnapshots),
];
for (raw, key) in cases {
diff --git a/crates/fluss/src/rpc/convert.rs b/crates/fluss/src/rpc/convert.rs
index 1862589b..441645c2 100644
--- a/crates/fluss/src/rpc/convert.rs
+++ b/crates/fluss/src/rpc/convert.rs
@@ -73,6 +73,7 @@ mod tests {
host: "127.0.0.1".to_string(),
port: 9092,
listeners: None,
+ rack: None,
};
let node = from_pb_server_node(pb, ServerType::TabletServer);
assert_eq!(node.id(), 7);
@@ -84,6 +85,7 @@ mod tests {
host: "localhost".to_string(),
port: 8123,
listeners: None,
+ rack: None,
};
let node = from_pb_server_node(pb, ServerType::CoordinatorServer);
assert_eq!(node.uid(), "cs-3");
diff --git a/crates/fluss/src/rpc/message/create_partition.rs
b/crates/fluss/src/rpc/message/create_partition.rs
index ad633655..1646f333 100644
--- a/crates/fluss/src/rpc/message/create_partition.rs
+++ b/crates/fluss/src/rpc/message/create_partition.rs
@@ -40,7 +40,7 @@ impl CreatePartitionRequest {
inner_request: proto::CreatePartitionRequest {
table_path: to_table_path(table_path),
partition_spec: partition_spec.to_pb(),
- ignore_if_exists,
+ ignore_if_not_exists: ignore_if_exists,
},
}
}
diff --git a/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
b/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
index 0b59384d..a23a985d 100644
--- a/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
+++ b/crates/fluss/src/rpc/message/get_latest_lake_snapshot.rs
@@ -48,7 +48,7 @@ impl GetLatestLakeSnapshotRequest {
impl RequestBody for GetLatestLakeSnapshotRequest {
type ResponseBody = proto::GetLatestLakeSnapshotResponse;
- const API_KEY: ApiKey = ApiKey::GetLatestLakeSnapshot;
+ const API_KEY: ApiKey = ApiKey::GetLakeSnapshot;
}
impl_write_type!(GetLatestLakeSnapshotRequest);
diff --git a/crates/fluss/src/rpc/message/list_databases.rs
b/crates/fluss/src/rpc/message/list_databases.rs
index 21e16400..74ca4944 100644
--- a/crates/fluss/src/rpc/message/list_databases.rs
+++ b/crates/fluss/src/rpc/message/list_databases.rs
@@ -32,7 +32,9 @@ pub struct ListDatabasesRequest {
impl ListDatabasesRequest {
pub fn new() -> Self {
ListDatabasesRequest {
- inner_request: proto::ListDatabasesRequest {},
+ inner_request: proto::ListDatabasesRequest {
+ include_summary: None,
+ },
}
}
}
diff --git a/crates/fluss/src/rpc/message/lookup.rs
b/crates/fluss/src/rpc/message/lookup.rs
index 200d4bc8..e205fa6b 100644
--- a/crates/fluss/src/rpc/message/lookup.rs
+++ b/crates/fluss/src/rpc/message/lookup.rs
@@ -42,7 +42,7 @@ impl LookupRequest {
|(bucket_id, partition_id, keys)| proto::PbLookupReqForBucket {
partition_id,
bucket_id,
- key: keys,
+ keys,
},
)
.collect();
@@ -50,6 +50,9 @@ impl LookupRequest {
let request = proto::LookupRequest {
table_id,
buckets_req,
+ insert_if_not_exists: None,
+ acks: None,
+ timeout_ms: None,
};
Self {
