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 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 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 {