From 64c0ede0261eeb7ccd415221d6f102aa70829b6a Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Wed, 10 Jun 2026 15:58:22 +0800 Subject: [PATCH] feat(table-catalog): add product API surface (#3320) * feat(table-catalog): add product API surface * fix(table-catalog): harden product API commits * fix(table-catalog): close product API review gaps * fix(table-catalog): validate rollback metadata before commit --------- Co-authored-by: Henry Guo --- rustfs/src/admin/handlers/table_catalog.rs | 1073 ++++++++++++++++++- rustfs/src/admin/route_policy.rs | 167 ++- rustfs/src/admin/route_registration_test.rs | 188 ++++ rustfs/src/table_catalog.rs | 278 ++++- 4 files changed, 1700 insertions(+), 6 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 462ab6cda..e546c2349 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -18,7 +18,7 @@ use crate::admin::{ }; use crate::auth::{check_key_valid, get_session_token}; use crate::server::{RemoteAddr, TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX}; -use crate::table_catalog::DEFAULT_WAREHOUSE_ID; +use crate::table_catalog::{DEFAULT_WAREHOUSE_ID, TableCatalogStore}; use http::{HeaderMap, HeaderValue, StatusCode}; use hyper::Method; use matchit::Params; @@ -40,6 +40,8 @@ const CREDENTIAL_VENDING_UNSUPPORTED: &str = "unsupported"; const TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT: &str = "namespaces"; const TABLE_CATALOG_TABLE_RESOURCE_ROOT: &str = "tables"; const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ + "PUT /buckets/{warehouse}", + "GET /buckets/{warehouse}", "GET /{warehouse}/namespaces", "POST /{warehouse}/namespaces", "GET /{warehouse}/namespaces/{namespace}", @@ -50,9 +52,21 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "GET /{warehouse}/namespaces/{namespace}/tables/{table}", "POST /{warehouse}/namespaces/{namespace}/tables/{table}", "DELETE /{warehouse}/namespaces/{namespace}/tables/{table}", + "POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", + "GET /{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + "PUT /{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + "GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + "PUT /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + "GET /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", + "GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export", + "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", + "GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", + "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", ]; static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {}; +static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {}; +static GET_TABLE_BUCKET_HANDLER: GetTableBucketHandler = GetTableBucketHandler {}; static LIST_NAMESPACES_HANDLER: RestListNamespacesHandler = RestListNamespacesHandler {}; static CREATE_NAMESPACE_HANDLER: RestCreateNamespaceHandler = RestCreateNamespaceHandler {}; static GET_NAMESPACE_HANDLER: RestGetNamespaceHandler = RestGetNamespaceHandler {}; @@ -63,7 +77,16 @@ static REGISTER_TABLE_HANDLER: RestRegisterTableHandler = RestRegisterTableHandl static LOAD_TABLE_HANDLER: RestLoadTableHandler = RestLoadTableHandler {}; static COMMIT_TABLE_HANDLER: RestCommitTableHandler = RestCommitTableHandler {}; static DROP_TABLE_HANDLER: RestDropTableHandler = RestDropTableHandler {}; +static GET_TABLE_METADATA_LOCATION_HANDLER: GetTableMetadataLocationHandler = GetTableMetadataLocationHandler {}; +static UPDATE_TABLE_METADATA_LOCATION_HANDLER: UpdateTableMetadataLocationHandler = UpdateTableMetadataLocationHandler {}; static TABLE_METADATA_MAINTENANCE_HANDLER: RestTableMetadataMaintenanceHandler = RestTableMetadataMaintenanceHandler {}; +static GET_TABLE_MAINTENANCE_CONFIG_HANDLER: GetTableMaintenanceConfigHandler = GetTableMaintenanceConfigHandler {}; +static PUT_TABLE_MAINTENANCE_CONFIG_HANDLER: PutTableMaintenanceConfigHandler = PutTableMaintenanceConfigHandler {}; +static GET_TABLE_MAINTENANCE_JOB_HANDLER: GetTableMaintenanceJobHandler = GetTableMaintenanceJobHandler {}; +static EXPORT_TABLE_CATALOG_HANDLER: ExportTableCatalogHandler = ExportTableCatalogHandler {}; +static IMPORT_TABLE_CATALOG_HANDLER: ImportTableCatalogHandler = ImportTableCatalogHandler {}; +static GET_TABLE_CATALOG_DIAGNOSTICS_HANDLER: GetTableCatalogDiagnosticsHandler = GetTableCatalogDiagnosticsHandler {}; +static ROLLBACK_TABLE_CATALOG_HANDLER: RollbackTableCatalogHandler = RollbackTableCatalogHandler {}; #[derive(Debug, Serialize)] struct CatalogConfigResponse { @@ -139,6 +162,62 @@ struct TableMetadataMaintenanceRequest { delete: bool, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct UpdateTableMetadataLocationRequest { + #[serde(rename = "metadata-location", alias = "metadataLocation")] + metadata_location: String, + #[serde(rename = "version-token", alias = "versionToken")] + version_token: String, + #[serde(default, rename = "commit-id", alias = "commitId")] + commit_id: Option, + #[serde(default, rename = "idempotency-key", alias = "idempotencyKey")] + idempotency_key: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct CatalogImportRequest { + #[serde(rename = "metadata-location", alias = "metadataLocation")] + metadata_location: String, + #[serde(default)] + properties: BTreeMap, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct RollbackTableRequest { + #[serde(rename = "metadata-location", alias = "metadataLocation")] + metadata_location: String, + #[serde(rename = "version-token", alias = "versionToken")] + version_token: String, + #[serde(default, rename = "commit-id", alias = "commitId")] + commit_id: Option, + #[serde(default, rename = "idempotency-key", alias = "idempotencyKey")] + idempotency_key: Option, +} + +#[derive(Debug, Serialize)] +struct TableBucketResponse { + #[serde(rename = "table-bucket")] + table_bucket: String, + enabled: bool, + #[serde(rename = "catalog-type")] + catalog_type: String, + warehouse: String, + #[serde(rename = "warehouse-location")] + warehouse_location: String, + #[serde(rename = "catalog-uri")] + catalog_uri: String, + #[serde(rename = "compat-catalog-uri")] + compat_catalog_uri: String, + #[serde(rename = "credential-vending")] + credential_vending: &'static str, + #[serde(rename = "catalog-entry-present")] + catalog_entry_present: bool, + properties: BTreeMap, +} + #[derive(Debug, Serialize)] struct RestNamespaceResponse { namespace: Vec, @@ -189,6 +268,17 @@ struct RestCommitTableResponse { commit_id: String, } +#[derive(Debug, Serialize)] +struct TableMetadataLocationResponse { + #[serde(rename = "metadata-location")] + metadata_location: String, + #[serde(rename = "version-token")] + version_token: String, + generation: u64, + #[serde(rename = "warehouse-location")] + warehouse_location: String, +} + pub fn register_table_catalog_route(r: &mut S3Router) -> std::io::Result<()> { for prefix in [TABLE_CATALOG_PREFIX, TABLE_CATALOG_COMPAT_PREFIX] { register_table_catalog_prefix_routes(r, prefix)?; @@ -199,6 +289,16 @@ pub fn register_table_catalog_route(r: &mut S3Router) -> std::io fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix: &str) -> std::io::Result<()> { r.insert(Method::GET, format!("{prefix}/config").as_str(), AdminOperation(&GET_CONFIG_HANDLER))?; + r.insert( + Method::PUT, + format!("{prefix}/buckets/{{warehouse}}").as_str(), + AdminOperation(&ENABLE_TABLE_BUCKET_HANDLER), + )?; + r.insert( + Method::GET, + format!("{prefix}/buckets/{{warehouse}}").as_str(), + AdminOperation(&GET_TABLE_BUCKET_HANDLER), + )?; r.insert( Method::GET, format!("{prefix}/{{warehouse}}/namespaces").as_str(), @@ -249,11 +349,56 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(), AdminOperation(&DROP_TABLE_HANDLER), )?; + r.insert( + Method::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/metadata-location").as_str(), + AdminOperation(&GET_TABLE_METADATA_LOCATION_HANDLER), + )?; + r.insert( + Method::PUT, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/metadata-location").as_str(), + AdminOperation(&UPDATE_TABLE_METADATA_LOCATION_HANDLER), + )?; r.insert( Method::POST, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/metadata").as_str(), AdminOperation(&TABLE_METADATA_MAINTENANCE_HANDLER), )?; + r.insert( + Method::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/config").as_str(), + AdminOperation(&GET_TABLE_MAINTENANCE_CONFIG_HANDLER), + )?; + r.insert( + Method::PUT, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/config").as_str(), + AdminOperation(&PUT_TABLE_MAINTENANCE_CONFIG_HANDLER), + )?; + r.insert( + Method::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/jobs/{{job}}").as_str(), + AdminOperation(&GET_TABLE_MAINTENANCE_JOB_HANDLER), + )?; + r.insert( + Method::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/export").as_str(), + AdminOperation(&EXPORT_TABLE_CATALOG_HANDLER), + )?; + r.insert( + Method::POST, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/import").as_str(), + AdminOperation(&IMPORT_TABLE_CATALOG_HANDLER), + )?; + r.insert( + Method::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/diagnostics").as_str(), + AdminOperation(&GET_TABLE_CATALOG_DIAGNOSTICS_HANDLER), + )?; + r.insert( + Method::POST, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/rollback").as_str(), + AdminOperation(&ROLLBACK_TABLE_CATALOG_HANDLER), + )?; Ok(()) } @@ -395,6 +540,14 @@ fn table_name_from_params(params: &Params<'_, '_>) -> S3Result { Ok(table.to_string()) } +fn job_id_from_params(params: &Params<'_, '_>) -> S3Result { + let job = params.get("job").unwrap_or(""); + if job.is_empty() { + return Err(s3_error!(InvalidRequest, "maintenance job id is required")); + } + Ok(job.to_string()) +} + fn table_catalog_backend() -> S3Result> { let store = new_object_layer_fn().ok_or_else(|| s3_error!(InternalError, "object store not initialized"))?; Ok(crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store)) @@ -425,6 +578,15 @@ fn table_bucket_entry_from_metadata_marker(bucket: &str) -> crate::table_catalog } } +async fn enable_table_bucket_marker(bucket: &str) -> S3Result<()> { + let marker = crate::table_catalog::table_bucket_marker_json() + .map_err(|err| s3_error!(InternalError, "failed to serialize table bucket marker: {}", err))?; + metadata_sys::update(bucket, crate::table_catalog::TABLE_BUCKET_MARKER_CONFIG, marker) + .await + .map(|_| ()) + .map_err(|err| s3_error!(InvalidRequest, "failed to enable table bucket {bucket}: {}", err)) +} + async fn ensure_table_bucket_entry(store: &S, bucket: &str, table_bucket_enabled: bool) -> S3Result<()> where S: crate::table_catalog::TableCatalogStore + ?Sized, @@ -441,6 +603,44 @@ where .map_err(catalog_store_error) } +async fn table_bucket_response(store: &S, bucket: &str, enabled: bool) -> S3Result +where + S: crate::table_catalog::TableCatalogStore + ?Sized, +{ + let entry = store.get_table_bucket(bucket).await.map_err(catalog_store_error)?; + let (catalog_type, warehouse_location, properties, catalog_entry_present) = match entry { + Some(entry) => (entry.catalog_type, entry.warehouse_root, entry.properties, true), + None => ( + crate::table_catalog::TABLE_BUCKET_CATALOG_TYPE.to_string(), + format!("s3://{bucket}/"), + BTreeMap::new(), + false, + ), + }; + + Ok(TableBucketResponse { + table_bucket: bucket.to_string(), + enabled, + catalog_type, + warehouse: bucket.to_string(), + warehouse_location, + catalog_uri: format!("{TABLE_CATALOG_PREFIX}/{bucket}"), + compat_catalog_uri: format!("{TABLE_CATALOG_COMPAT_PREFIX}/{bucket}"), + credential_vending: CREDENTIAL_VENDING_UNSUPPORTED, + catalog_entry_present, + properties, + }) +} + +async fn enable_table_bucket_response(store: &S, bucket: &str) -> S3Result +where + S: crate::table_catalog::TableCatalogStore + ?Sized, +{ + ensure_table_bucket_entry(store, bucket, true).await?; + enable_table_bucket_marker(bucket).await?; + table_bucket_response(store, bucket, true).await +} + fn namespace_segments(namespace: &crate::table_catalog::Namespace) -> Vec { namespace .segments() @@ -522,6 +722,15 @@ fn commit_table_response_from_result( } } +fn table_metadata_location_response_from_entry(entry: crate::table_catalog::TableEntry) -> TableMetadataLocationResponse { + TableMetadataLocationResponse { + metadata_location: entry.metadata_location, + version_token: entry.version_token, + generation: entry.generation, + warehouse_location: entry.warehouse_location, + } +} + fn table_commit_request_from_rest_request( bucket: &str, namespace: &crate::table_catalog::Namespace, @@ -599,6 +808,39 @@ fn table_entry_from_register_request( }) } +fn table_entry_from_import_request( + bucket: &str, + namespace: &crate::table_catalog::Namespace, + table: &str, + request: CatalogImportRequest, +) -> S3Result { + let table = crate::table_catalog::IdentifierSegment::parse(table.to_string()) + .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?; + if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table, &request.metadata_location) { + return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory")); + } + + let table_id = Uuid::new_v4().to_string(); + Ok(crate::table_catalog::TableEntry { + version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.as_str().to_string(), + table_id: table_id.clone(), + table_uuid: Uuid::new_v4().to_string(), + format: "ICEBERG".to_string(), + format_version: 2, + warehouse_location: format!("s3://{bucket}/tables/{table_id}"), + metadata_location: request.metadata_location, + version_token: format!("token-{}", Uuid::new_v4()), + generation: 1, + state: crate::table_catalog::TableCatalogEntryState::Active, + properties: request.properties, + created_at: None, + updated_at: None, + }) +} + fn table_entry_from_create_table_request( bucket: &str, namespace: &crate::table_catalog::Namespace, @@ -1457,6 +1699,67 @@ where Ok(load_table_response_from_entry(entry, metadata)) } +async fn get_table_metadata_location_response( + store: &S, + bucket: &str, + namespace: &crate::table_catalog::Namespace, + table: &str, +) -> S3Result +where + S: crate::table_catalog::TableCatalogStore + ?Sized, +{ + let Some(entry) = store + .load_table(bucket, &namespace.public_name(), table) + .await + .map_err(catalog_store_error)? + else { + return Err(s3_error!(InvalidRequest, "table not found")); + }; + Ok(table_metadata_location_response_from_entry(entry)) +} + +async fn update_table_metadata_location_response( + store: &S, + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, + bucket: &str, + namespace: &crate::table_catalog::Namespace, + table: &str, + request: UpdateTableMetadataLocationRequest, +) -> S3Result +where + S: crate::table_catalog::TableCatalogStore + ?Sized, +{ + let Some(current) = store + .load_table(bucket, &namespace.public_name(), table) + .await + .map_err(catalog_store_error)? + else { + return Err(s3_error!(InvalidRequest, "table not found")); + }; + let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string()) + .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?; + if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &request.metadata_location) { + return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory")); + } + let target_metadata = read_table_metadata_json(metadata_backend, bucket, &request.metadata_location).await?; + validate_metadata_table_location_in_bucket(bucket, &target_metadata)?; + let commit_request = crate::table_catalog::TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.to_string(), + commit_id: request.commit_id.unwrap_or_else(|| Uuid::new_v4().to_string()), + idempotency_key: request.idempotency_key, + operation: "update-metadata-location".to_string(), + expected_version_token: request.version_token, + expected_metadata_location: current.metadata_location, + new_metadata_location: request.metadata_location, + requirements: Vec::new(), + writer: Some("rustfs-metadata-location-api".to_string()), + }; + let result = store.commit_table(commit_request).await.map_err(catalog_store_error)?; + Ok(table_metadata_location_response_from_entry(result.table)) +} + async fn commit_table_response( store: &S, metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, @@ -1554,7 +1857,7 @@ async fn table_metadata_maintenance_response( where B: crate::table_catalog::TableCatalogObjectBackend, { - if request.delete { + let report = if request.delete { store .delete_table_metadata_maintenance_candidates( bucket, @@ -1569,7 +1872,74 @@ where .plan_table_metadata_maintenance(bucket, &namespace.public_name(), table, request.retain_recent_metadata_files) .await .map_err(catalog_store_error) + }?; + store + .put_table_metadata_maintenance_report(&report) + .await + .map_err(catalog_store_error)?; + Ok(report) +} + +async fn catalog_import_response( + store: &crate::table_catalog::ObjectTableCatalogStore, + metadata_backend: &B, + bucket: &str, + namespace: &crate::table_catalog::Namespace, + table: &str, + request: CatalogImportRequest, + table_bucket_enabled: bool, +) -> S3Result +where + B: crate::table_catalog::TableCatalogObjectBackend, +{ + ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?; + let entry = table_entry_from_import_request(bucket, namespace, table, request)?; + let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; + validate_metadata_table_location_in_bucket(bucket, &metadata)?; + store.register_table(entry.clone()).await.map_err(catalog_store_error)?; + Ok(load_table_response_from_entry(entry, metadata)) +} + +async fn rollback_table_response( + store: &S, + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, + bucket: &str, + namespace: &crate::table_catalog::Namespace, + table: &str, + request: RollbackTableRequest, +) -> S3Result +where + S: crate::table_catalog::TableCatalogStore + ?Sized, +{ + let Some(current) = store + .load_table(bucket, &namespace.public_name(), table) + .await + .map_err(catalog_store_error)? + else { + return Err(s3_error!(InvalidRequest, "table not found")); + }; + let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string()) + .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?; + if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table_name, &request.metadata_location) { + return Err(s3_error!(InvalidRequest, "metadata location must be inside the table metadata directory")); } + let target_metadata = read_table_metadata_json(metadata_backend, bucket, &request.metadata_location).await?; + validate_metadata_table_location_in_bucket(bucket, &target_metadata)?; + let commit_request = crate::table_catalog::TableCommitRequest { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.to_string(), + commit_id: request.commit_id.unwrap_or_else(|| Uuid::new_v4().to_string()), + idempotency_key: request.idempotency_key, + operation: "rollback".to_string(), + expected_version_token: request.version_token, + expected_metadata_location: current.metadata_location, + new_metadata_location: request.metadata_location, + requirements: Vec::new(), + writer: Some("rustfs-catalog-rollback-api".to_string()), + }; + let result = store.commit_table(commit_request).await.map_err(catalog_store_error)?; + Ok(commit_table_response_from_result(result, target_metadata)) } pub struct GetCatalogConfigHandler {} @@ -1582,6 +1952,35 @@ impl Operation for GetCatalogConfigHandler { } } +pub struct EnableTableBucketHandler {} + +#[async_trait::async_trait] +impl Operation for EnableTableBucketHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let resource = TableCatalogResource::warehouse(&warehouse); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableBucketAction).await?; + let store = table_catalog_store()?; + let response = enable_table_bucket_response(&store, &warehouse).await?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct GetTableBucketHandler {} + +#[async_trait::async_trait] +impl Operation for GetTableBucketHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let resource = TableCatalogResource::warehouse(&warehouse); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableBucketAction).await?; + let store = table_catalog_store()?; + let enabled = table_bucket_enabled_from_metadata(&warehouse).await?; + let response = table_bucket_response(&store, &warehouse, enabled).await?; + build_json_response(StatusCode::OK, &response) + } +} + pub struct RestListNamespacesHandler {} #[async_trait::async_trait] @@ -1746,6 +2145,41 @@ impl Operation for RestDropTableHandler { } } +pub struct GetTableMetadataLocationHandler {} + +#[async_trait::async_trait] +impl Operation for GetTableMetadataLocationHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataLocationAction).await?; + let store = table_catalog_store()?; + let response = get_table_metadata_location_response(&store, &warehouse, &namespace, &table).await?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct UpdateTableMetadataLocationHandler {} + +#[async_trait::async_trait] +impl Operation for UpdateTableMetadataLocationHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableMetadataLocationAction).await?; + let request = read_json_body::(req.input).await?; + let metadata_backend = table_catalog_backend()?; + let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let response = + update_table_metadata_location_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?; + build_json_response(StatusCode::OK, &response) + } +} + pub struct RestTableMetadataMaintenanceHandler {} #[async_trait::async_trait] @@ -1764,6 +2198,149 @@ impl Operation for RestTableMetadataMaintenanceHandler { } } +pub struct GetTableMaintenanceConfigHandler {} + +#[async_trait::async_trait] +impl Operation for GetTableMaintenanceConfigHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableLifecycleAction).await?; + let store = table_catalog_store()?; + let response = store + .get_table_maintenance_config(&warehouse, &namespace.public_name(), &table) + .await + .map_err(catalog_store_error)?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct PutTableMaintenanceConfigHandler {} + +#[async_trait::async_trait] +impl Operation for PutTableMaintenanceConfigHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableLifecycleAction).await?; + let request = read_json_body::(req.input).await?; + let store = table_catalog_store()?; + let response = store + .put_table_maintenance_config(&warehouse, &namespace.public_name(), &table, request) + .await + .map_err(catalog_store_error)?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct GetTableMaintenanceJobHandler {} + +#[async_trait::async_trait] +impl Operation for GetTableMaintenanceJobHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let job = job_id_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableLifecycleAction).await?; + let store = table_catalog_store()?; + let Some(response) = store + .get_table_metadata_maintenance_report(&warehouse, &namespace.public_name(), &table, &job) + .await + .map_err(catalog_store_error)? + else { + return Err(s3_error!(InvalidRequest, "maintenance job not found")); + }; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct ExportTableCatalogHandler {} + +#[async_trait::async_trait] +impl Operation for ExportTableCatalogHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; + let store = table_catalog_store()?; + let response = store + .export_table_catalog_entry(&warehouse, &namespace.public_name(), &table) + .await + .map_err(catalog_store_error)?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct ImportTableCatalogHandler {} + +#[async_trait::async_trait] +impl Operation for ImportTableCatalogHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::RegisterTableAction).await?; + let request = read_json_body::(req.input).await?; + let metadata_backend = table_catalog_backend()?; + let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; + let response = + catalog_import_response(&store, &metadata_backend, &warehouse, &namespace, &table, request, table_bucket_enabled) + .await?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct GetTableCatalogDiagnosticsHandler {} + +#[async_trait::async_trait] +impl Operation for GetTableCatalogDiagnosticsHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; + let store = table_catalog_store()?; + let config = store + .get_table_maintenance_config(&warehouse, &namespace.public_name(), &table) + .await + .map_err(catalog_store_error)?; + let response = store + .diagnose_table_catalog(&warehouse, &namespace.public_name(), &table, config.retain_recent_metadata_files) + .await + .map_err(catalog_store_error)?; + build_json_response(StatusCode::OK, &response) + } +} + +pub struct RollbackTableCatalogHandler {} + +#[async_trait::async_trait] +impl Operation for RollbackTableCatalogHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let table = table_name_from_params(¶ms)?; + let resource = TableCatalogResource::table(&warehouse, &namespace, &table); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?; + let request = read_json_body::(req.input).await?; + let metadata_backend = table_catalog_backend()?; + let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let response = rollback_table_response(&store, &metadata_backend, &warehouse, &namespace, &table, request).await?; + build_json_response(StatusCode::OK, &response) + } +} + #[cfg(test)] mod tests { use super::*; @@ -1819,6 +2396,8 @@ mod tests { ); for (handler, action) in [ + ("EnableTableBucketHandler", "AdminAction::SetTableBucketAction"), + ("GetTableBucketHandler", "AdminAction::GetTableBucketAction"), ("RestListNamespacesHandler", "AdminAction::GetTableNamespaceAction"), ("RestCreateNamespaceHandler", "AdminAction::SetTableNamespaceAction"), ("RestGetNamespaceHandler", "AdminAction::GetTableNamespaceAction"), @@ -1829,7 +2408,16 @@ mod tests { ("RestLoadTableHandler", "AdminAction::GetTableMetadataAction"), ("RestCommitTableHandler", "AdminAction::CommitTableAction"), ("RestDropTableHandler", "AdminAction::DeleteTableAction"), + ("GetTableMetadataLocationHandler", "AdminAction::GetTableMetadataLocationAction"), + ("UpdateTableMetadataLocationHandler", "AdminAction::SetTableMetadataLocationAction"), ("RestTableMetadataMaintenanceHandler", "AdminAction::RunTableMaintenanceAction"), + ("GetTableMaintenanceConfigHandler", "AdminAction::GetTableLifecycleAction"), + ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"), + ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), + ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), + ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), + ("GetTableCatalogDiagnosticsHandler", "AdminAction::GetTableMetadataAction"), + ("RollbackTableCatalogHandler", "AdminAction::CommitTableAction"), ] { let block = operation_block(src, handler); assert!( @@ -1850,7 +2438,16 @@ mod tests { ("RestLoadTableHandler", "AdminAction::GetTableMetadataAction"), ("RestCommitTableHandler", "AdminAction::CommitTableAction"), ("RestDropTableHandler", "AdminAction::DeleteTableAction"), + ("GetTableMetadataLocationHandler", "AdminAction::GetTableMetadataLocationAction"), + ("UpdateTableMetadataLocationHandler", "AdminAction::SetTableMetadataLocationAction"), ("RestTableMetadataMaintenanceHandler", "AdminAction::RunTableMaintenanceAction"), + ("GetTableMaintenanceConfigHandler", "AdminAction::GetTableLifecycleAction"), + ("PutTableMaintenanceConfigHandler", "AdminAction::SetTableLifecycleAction"), + ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), + ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), + ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), + ("GetTableCatalogDiagnosticsHandler", "AdminAction::GetTableMetadataAction"), + ("RollbackTableCatalogHandler", "AdminAction::CommitTableAction"), ] { let block = operation_block(src, handler); assert!( @@ -1898,6 +2495,8 @@ mod tests { fn rest_catalog_mvp_routes_use_implemented_handlers() { fn assert_operation() {} + let _: &EnableTableBucketHandler = &ENABLE_TABLE_BUCKET_HANDLER; + let _: &GetTableBucketHandler = &GET_TABLE_BUCKET_HANDLER; let _: &RestListNamespacesHandler = &LIST_NAMESPACES_HANDLER; let _: &RestCreateNamespaceHandler = &CREATE_NAMESPACE_HANDLER; let _: &RestGetNamespaceHandler = &GET_NAMESPACE_HANDLER; @@ -1908,8 +2507,19 @@ mod tests { let _: &RestLoadTableHandler = &LOAD_TABLE_HANDLER; let _: &RestCommitTableHandler = &COMMIT_TABLE_HANDLER; let _: &RestDropTableHandler = &DROP_TABLE_HANDLER; + let _: &GetTableMetadataLocationHandler = &GET_TABLE_METADATA_LOCATION_HANDLER; + let _: &UpdateTableMetadataLocationHandler = &UPDATE_TABLE_METADATA_LOCATION_HANDLER; let _: &RestTableMetadataMaintenanceHandler = &TABLE_METADATA_MAINTENANCE_HANDLER; + let _: &GetTableMaintenanceConfigHandler = &GET_TABLE_MAINTENANCE_CONFIG_HANDLER; + let _: &PutTableMaintenanceConfigHandler = &PUT_TABLE_MAINTENANCE_CONFIG_HANDLER; + let _: &GetTableMaintenanceJobHandler = &GET_TABLE_MAINTENANCE_JOB_HANDLER; + let _: &ExportTableCatalogHandler = &EXPORT_TABLE_CATALOG_HANDLER; + let _: &ImportTableCatalogHandler = &IMPORT_TABLE_CATALOG_HANDLER; + let _: &GetTableCatalogDiagnosticsHandler = &GET_TABLE_CATALOG_DIAGNOSTICS_HANDLER; + let _: &RollbackTableCatalogHandler = &ROLLBACK_TABLE_CATALOG_HANDLER; + assert_operation::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); @@ -1920,7 +2530,16 @@ mod tests { assert_operation::(); assert_operation::(); assert_operation::(); + assert_operation::(); + assert_operation::(); assert_operation::(); + assert_operation::(); + assert_operation::(); + assert_operation::(); + assert_operation::(); + assert_operation::(); + assert_operation::(); + assert_operation::(); } #[test] @@ -1944,6 +2563,27 @@ mod tests { assert!(request.delete); } + #[tokio::test] + async fn table_bucket_response_reports_catalog_discovery_without_credentials() { + let store = TestTableCatalogStore::default(); + ensure_table_bucket_entry(&store, "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + + let response = table_bucket_response(&store, "warehouse", true) + .await + .expect("bucket response should build"); + + assert_eq!(response.table_bucket, "warehouse"); + assert!(response.enabled); + assert_eq!(response.catalog_type, crate::table_catalog::TABLE_BUCKET_CATALOG_TYPE); + assert_eq!(response.warehouse_location, "s3://warehouse/"); + assert_eq!(response.catalog_uri, "/iceberg/v1/warehouse"); + assert_eq!(response.compat_catalog_uri, "/_iceberg/v1/warehouse"); + assert_eq!(response.credential_vending, CREDENTIAL_VENDING_UNSUPPORTED); + assert!(response.catalog_entry_present); + } + #[test] fn table_catalog_ingress_requests_reject_unknown_fields() { assert_rejects_unknown_field::( @@ -1982,6 +2622,36 @@ mod tests { "unexpected": true }), ); + assert_rejects_unknown_field::( + "UpdateTableMetadataLocationRequest", + serde_json::json!({ + "metadata-location": ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00002.metadata.json", + "version-token": "token-v1", + "unexpected": true + }), + ); + assert_rejects_unknown_field::( + "CatalogImportRequest", + serde_json::json!({ + "metadata-location": ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json", + "unexpected": true + }), + ); + assert_rejects_unknown_field::( + "RollbackTableRequest", + serde_json::json!({ + "metadata-location": ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json", + "version-token": "token-v2", + "unexpected": true + }), + ); + assert_rejects_unknown_field::( + "TableMaintenanceConfig", + serde_json::json!({ + "version": 1, + "unexpected": true + }), + ); } fn assert_rejects_unknown_field(target: &str, value: serde_json::Value) @@ -2345,6 +3015,44 @@ mod tests { ) .await; + let default_config = store + .get_table_maintenance_config(bucket, "analytics", "events") + .await + .expect("default maintenance config should load"); + assert_eq!(default_config, crate::table_catalog::TableMaintenanceConfig::default()); + let config = store + .put_table_maintenance_config( + bucket, + "analytics", + "events", + crate::table_catalog::TableMaintenanceConfig { + version: crate::table_catalog::TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 2, + delete_enabled: true, + background_enabled: false, + }, + ) + .await + .expect("maintenance config should persist"); + assert_eq!(config.retain_recent_metadata_files, 2); + assert!(config.delete_enabled); + assert!( + store + .put_table_maintenance_config( + bucket, + "analytics", + "events", + crate::table_catalog::TableMaintenanceConfig { + version: crate::table_catalog::TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 2, + delete_enabled: true, + background_enabled: true, + }, + ) + .await + .is_err() + ); + let dry_run = table_metadata_maintenance_response( &store, bucket, @@ -2359,6 +3067,12 @@ mod tests { .expect("metadata maintenance dry-run should succeed"); assert_eq!(dry_run.cleanup_candidate_locations, vec![old.clone()]); assert_eq!(dry_run.deletable_metadata_locations, vec![old.clone()]); + let stored_dry_run = store + .get_table_metadata_maintenance_report(bucket, "analytics", "events", &dry_run.job.job_id) + .await + .expect("maintenance job lookup should succeed") + .expect("maintenance job should be stored"); + assert_eq!(stored_dry_run, dry_run); assert!( backend .object_exists(bucket, &old) @@ -2505,6 +3219,7 @@ mod tests { namespaces: tokio::sync::Mutex>, tables: tokio::sync::Mutex>, commits: tokio::sync::Mutex>, + fail_put_table_bucket: tokio::sync::Mutex, } #[derive(Clone, Default)] @@ -2681,6 +3396,15 @@ mod tests { &self, entry: crate::table_catalog::TableBucketEntry, ) -> crate::table_catalog::TableCatalogStoreResult<()> { + let mut fail_put_table_bucket = self.fail_put_table_bucket.lock().await; + if *fail_put_table_bucket { + *fail_put_table_bucket = false; + return Err(crate::table_catalog::TableCatalogStoreError::Internal( + "injected table bucket write failure".to_string(), + )); + } + drop(fail_put_table_bucket); + let mut table_buckets = self.table_buckets.lock().await; table_buckets.retain(|existing| existing.table_bucket != entry.table_bucket); table_buckets.push(entry); @@ -2925,6 +3649,22 @@ mod tests { ); } + #[tokio::test] + async fn enable_table_bucket_response_fails_before_marker_when_catalog_entry_fails() { + let store = TestTableCatalogStore::default(); + *store.fail_put_table_bucket.lock().await = true; + + assert!(enable_table_bucket_response(&store, "warehouse").await.is_err()); + assert!( + store + .get_table_bucket("warehouse") + .await + .expect("table bucket lookup should succeed") + .is_none() + ); + assert!(!*store.fail_put_table_bucket.lock().await); + } + #[tokio::test] async fn namespace_helpers_call_catalog_store() { let store = TestTableCatalogStore::default(); @@ -3082,4 +3822,333 @@ mod tests { .is_err() ); } + + #[tokio::test] + async fn metadata_location_api_loads_and_updates_current_pointer() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + ensure_table_bucket_entry(&store, "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let current_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json"; + store + .register_table( + table_entry_from_register_request( + "warehouse", + &namespace, + RegisterTableRequest { + name: "events".to_string(), + metadata_location: current_location.to_string(), + overwrite: false, + }, + ) + .expect("table entry should build"), + ) + .await + .expect("table should register"); + let current = get_table_metadata_location_response(&store, "warehouse", &namespace, "events") + .await + .expect("metadata location should load"); + let next_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00002.metadata.json"; + metadata_backend + .put_json( + "warehouse", + next_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://warehouse/tables/table-id" + }), + ) + .await; + + let updated = update_table_metadata_location_response( + &store, + &metadata_backend, + "warehouse", + &namespace, + "events", + UpdateTableMetadataLocationRequest { + metadata_location: next_location.to_string(), + version_token: current.version_token.clone(), + commit_id: Some("commit-1".to_string()), + idempotency_key: Some("retry-1".to_string()), + }, + ) + .await + .expect("metadata location should update"); + + assert_eq!(updated.metadata_location, next_location); + assert_eq!(updated.generation, current.generation + 1); + assert_ne!(updated.version_token, current.version_token); + } + + #[tokio::test] + async fn metadata_location_api_rejects_invalid_target_metadata_before_commit() { + let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + ensure_table_bucket_entry(&store, "warehouse", true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let current_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json"; + store + .register_table( + table_entry_from_register_request( + "warehouse", + &namespace, + RegisterTableRequest { + name: "events".to_string(), + metadata_location: current_location.to_string(), + overwrite: false, + }, + ) + .expect("table entry should build"), + ) + .await + .expect("table should register"); + let current = get_table_metadata_location_response(&store, "warehouse", &namespace, "events") + .await + .expect("metadata location should load"); + let invalid_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00002.metadata.json"; + metadata_backend + .put_json( + "warehouse", + invalid_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://other-warehouse/tables/table-id" + }), + ) + .await; + + assert!( + update_table_metadata_location_response( + &store, + &metadata_backend, + "warehouse", + &namespace, + "events", + UpdateTableMetadataLocationRequest { + metadata_location: invalid_location.to_string(), + version_token: current.version_token, + commit_id: Some("commit-1".to_string()), + idempotency_key: None, + }, + ) + .await + .is_err() + ); + let unchanged = get_table_metadata_location_response(&store, "warehouse", &namespace, "events") + .await + .expect("metadata location should still load"); + assert_eq!(unchanged.metadata_location, current_location); + assert_eq!(unchanged.generation, current.generation); + } + + #[tokio::test] + async fn catalog_import_and_rollback_use_register_and_commit_paths() { + let backend = TestTableCatalogObjectBackend::default(); + let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone()); + let bucket = "warehouse"; + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse"); + ensure_table_bucket_entry(&store, bucket, true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + bucket, + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let imported_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + backend + .put_json( + bucket, + &imported_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://warehouse/tables/table-id" + }), + ) + .await; + + let imported = catalog_import_response( + &store, + &backend, + bucket, + &namespace, + "events", + CatalogImportRequest { + metadata_location: imported_location.clone(), + properties: BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]), + }, + true, + ) + .await + .expect("catalog import should register table"); + assert_eq!(imported.metadata_location, imported_location); + let current = store + .load_table(bucket, "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should exist"); + assert_eq!(current.properties.get("owner").map(String::as_str), Some("lakehouse")); + + let rollback_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + backend + .put_json( + bucket, + &rollback_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://warehouse/tables/table-id", + "last-sequence-number": 2 + }), + ) + .await; + let rollback = rollback_table_response( + &store, + &backend, + bucket, + &namespace, + "events", + RollbackTableRequest { + metadata_location: rollback_location.clone(), + version_token: current.version_token, + commit_id: Some("rollback-1".to_string()), + idempotency_key: None, + }, + ) + .await + .expect("rollback should commit selected metadata"); + + assert_eq!(rollback.metadata_location, rollback_location); + assert_eq!(rollback.commit_id, "rollback-1"); + } + + #[tokio::test] + async fn rollback_rejects_invalid_target_metadata_before_commit() { + let backend = TestTableCatalogObjectBackend::default(); + let store = crate::table_catalog::ObjectTableCatalogStore::new(backend.clone()); + let bucket = "warehouse"; + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse"); + ensure_table_bucket_entry(&store, bucket, true) + .await + .expect("table bucket entry should be seeded"); + create_namespace_response( + &store, + bucket, + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("namespace should be created"); + let current_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + backend + .put_json( + bucket, + ¤t_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://warehouse/tables/table-id" + }), + ) + .await; + catalog_import_response( + &store, + &backend, + bucket, + &namespace, + "events", + CatalogImportRequest { + metadata_location: current_location.clone(), + properties: BTreeMap::new(), + }, + true, + ) + .await + .expect("catalog import should register table"); + let current = store + .load_table(bucket, "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should exist"); + + let invalid_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002.metadata.json"); + backend + .put_json( + bucket, + &invalid_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://other-warehouse/tables/table-id" + }), + ) + .await; + + assert!( + rollback_table_response( + &store, + &backend, + bucket, + &namespace, + "events", + RollbackTableRequest { + metadata_location: invalid_location, + version_token: current.version_token, + commit_id: Some("rollback-1".to_string()), + idempotency_key: None, + }, + ) + .await + .is_err() + ); + let unchanged = store + .load_table(bucket, "analytics", "events") + .await + .expect("table lookup should succeed") + .expect("table should still exist"); + + assert_eq!(unchanged.metadata_location, current_location); + assert_eq!(unchanged.generation, current.generation); + } } diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 6b63b01e7..004c87fb4 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -39,8 +39,11 @@ const GET_GROUP: AdminActionRef = AdminActionRef::new("GetGroupAdminAction"); const GET_POLICY: AdminActionRef = AdminActionRef::new("GetPolicyAdminAction"); const GET_REPLICATION_METRICS: AdminActionRef = AdminActionRef::new("GetReplicationMetricsAction"); const GET_TABLE: AdminActionRef = AdminActionRef::new("GetTableAction"); +const GET_TABLE_BUCKET: AdminActionRef = AdminActionRef::new("GetTableBucketAction"); const GET_TABLE_CATALOG: AdminActionRef = AdminActionRef::new("GetTableCatalogAction"); +const GET_TABLE_LIFECYCLE: AdminActionRef = AdminActionRef::new("GetTableLifecycleAction"); const GET_TABLE_METADATA: AdminActionRef = AdminActionRef::new("GetTableMetadataAction"); +const GET_TABLE_METADATA_LOCATION: AdminActionRef = AdminActionRef::new("GetTableMetadataLocationAction"); const GET_TABLE_NAMESPACE: AdminActionRef = AdminActionRef::new("GetTableNamespaceAction"); const HEAL: AdminActionRef = AdminActionRef::new("HealAdminAction"); const IMPORT_BUCKET_METADATA: AdminActionRef = AdminActionRef::new("ImportBucketMetadataAction"); @@ -66,6 +69,9 @@ const SERVER_INFO: AdminActionRef = AdminActionRef::new("ServerInfoAdminAction") const SET_BUCKET_QUOTA: AdminActionRef = AdminActionRef::new("SetBucketQuotaAdminAction"); const SET_BUCKET_TARGET: AdminActionRef = AdminActionRef::new("SetBucketTargetAction"); const SET_TABLE: AdminActionRef = AdminActionRef::new("SetTableAction"); +const SET_TABLE_BUCKET: AdminActionRef = AdminActionRef::new("SetTableBucketAction"); +const SET_TABLE_LIFECYCLE: AdminActionRef = AdminActionRef::new("SetTableLifecycleAction"); +const SET_TABLE_METADATA_LOCATION: AdminActionRef = AdminActionRef::new("SetTableMetadataLocationAction"); const SET_TABLE_NAMESPACE: AdminActionRef = AdminActionRef::new("SetTableNamespaceAction"); const SET_TIER: AdminActionRef = AdminActionRef::new("SetTierAction"); const SITE_REPLICATION_ADD: AdminActionRef = AdminActionRef::new("SiteReplicationAddAction"); @@ -590,6 +596,13 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ ), admin(HttpMethod::Post, "/rustfs/admin/v3/oidc/validate", SERVER_INFO, RouteRiskLevel::High), admin(HttpMethod::Get, "/iceberg/v1/config", GET_TABLE_CATALOG, RouteRiskLevel::Sensitive), + admin(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET, RouteRiskLevel::High), + admin( + HttpMethod::Get, + "/iceberg/v1/buckets/{warehouse}", + GET_TABLE_BUCKET, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", @@ -650,13 +663,79 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ DELETE_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + GET_TABLE_METADATA_LOCATION, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Put, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + SET_TABLE_METADATA_LOCATION, + RouteRiskLevel::High, + ), admin( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", RUN_TABLE_MAINTENANCE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + GET_TABLE_LIFECYCLE, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Put, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + SET_TABLE_LIFECYCLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", + GET_TABLE_LIFECYCLE, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", + REGISTER_TABLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", + COMMIT_TABLE, + RouteRiskLevel::High, + ), admin(HttpMethod::Get, "/_iceberg/v1/config", GET_TABLE_CATALOG, RouteRiskLevel::Sensitive), + admin( + HttpMethod::Put, + "/_iceberg/v1/buckets/{warehouse}", + SET_TABLE_BUCKET, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/_iceberg/v1/buckets/{warehouse}", + GET_TABLE_BUCKET, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces", @@ -717,12 +796,66 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ DELETE_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + GET_TABLE_METADATA_LOCATION, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Put, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + SET_TABLE_METADATA_LOCATION, + RouteRiskLevel::High, + ), admin( HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", RUN_TABLE_MAINTENANCE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + GET_TABLE_LIFECYCLE, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Put, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + SET_TABLE_LIFECYCLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", + GET_TABLE_LIFECYCLE, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", + REGISTER_TABLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", + COMMIT_TABLE, + RouteRiskLevel::High, + ), ]; pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[ @@ -896,7 +1029,9 @@ mod tests { let table_specs = ADMIN_ROUTE_POLICY_SPECS .iter() .filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1")); - assert_eq!(table_specs.count(), 24); + assert_eq!(table_specs.count(), 46); + assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET); + assert_action(HttpMethod::Get, "/_iceberg/v1/buckets/{warehouse}", GET_TABLE_BUCKET); assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE); assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE); assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables", CREATE_TABLE); @@ -921,6 +1056,16 @@ mod tests { "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}", COMMIT_TABLE, ); + assert_action( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + GET_TABLE_METADATA_LOCATION, + ); + assert_action( + HttpMethod::Put, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + SET_TABLE_METADATA_LOCATION, + ); assert_action( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", @@ -931,6 +1076,26 @@ mod tests { "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", RUN_TABLE_MAINTENANCE, ); + assert_action( + HttpMethod::Put, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + SET_TABLE_LIFECYCLE, + ); + assert_action( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", + GET_TABLE_LIFECYCLE, + ); + assert_action( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", + REGISTER_TABLE, + ); + assert_action( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", + COMMIT_TABLE, + ); } #[test] diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index cda1257cf..3f3889372 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -281,6 +281,8 @@ fn expected_admin_route_matrix() -> Vec { admin_route_sample(Method::DELETE, "/v3/oidc/config/{provider_id}", "/v3/oidc/config/default"), admin_route(Method::POST, "/v3/oidc/validate"), table_route(Method::GET, "/config"), + table_route_sample(Method::PUT, "/buckets/{warehouse}", "/buckets/analytics"), + table_route_sample(Method::GET, "/buckets/{warehouse}", "/buckets/analytics"), table_route_sample(Method::GET, "/{warehouse}/namespaces", "/analytics/namespaces"), table_route_sample(Method::POST, "/{warehouse}/namespaces", "/analytics/namespaces"), table_route_sample(Method::GET, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), @@ -320,7 +322,54 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", "/analytics/namespaces/sales/tables/orders/maintenance/metadata", ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + "/analytics/namespaces/sales/tables/orders/metadata-location", + ), + table_route_sample( + Method::PUT, + "/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + "/analytics/namespaces/sales/tables/orders/metadata-location", + ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + "/analytics/namespaces/sales/tables/orders/maintenance/config", + ), + table_route_sample( + Method::PUT, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + "/analytics/namespaces/sales/tables/orders/maintenance/config", + ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", + "/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1", + ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export", + "/analytics/namespaces/sales/tables/orders/catalog/export", + ), + table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", + "/analytics/namespaces/sales/tables/orders/catalog/import", + ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", + "/analytics/namespaces/sales/tables/orders/catalog/diagnostics", + ), + table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", + "/analytics/namespaces/sales/tables/orders/catalog/rollback", + ), compat_table_route(Method::GET, "/config"), + compat_table_route_sample(Method::PUT, "/buckets/{warehouse}", "/buckets/analytics"), + compat_table_route_sample(Method::GET, "/buckets/{warehouse}", "/buckets/analytics"), compat_table_route_sample(Method::GET, "/{warehouse}/namespaces", "/analytics/namespaces"), compat_table_route_sample(Method::POST, "/{warehouse}/namespaces", "/analytics/namespaces"), compat_table_route_sample(Method::GET, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), @@ -360,6 +409,51 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", "/analytics/namespaces/sales/tables/orders/maintenance/metadata", ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + "/analytics/namespaces/sales/tables/orders/metadata-location", + ), + compat_table_route_sample( + Method::PUT, + "/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", + "/analytics/namespaces/sales/tables/orders/metadata-location", + ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + "/analytics/namespaces/sales/tables/orders/maintenance/config", + ), + compat_table_route_sample( + Method::PUT, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/config", + "/analytics/namespaces/sales/tables/orders/maintenance/config", + ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/jobs/{job}", + "/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1", + ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/export", + "/analytics/namespaces/sales/tables/orders/catalog/export", + ), + compat_table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", + "/analytics/namespaces/sales/tables/orders/catalog/import", + ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", + "/analytics/namespaces/sales/tables/orders/catalog/diagnostics", + ), + compat_table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", + "/analytics/namespaces/sales/tables/orders/catalog/rollback", + ), ] } @@ -467,6 +561,8 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::GET, &admin_path("/v3/scanner/status")); assert_route(&router, Method::GET, &table_catalog_path("/config")); + assert_route(&router, Method::PUT, &table_catalog_path("/buckets/analytics")); + assert_route(&router, Method::GET, &table_catalog_path("/buckets/analytics")); assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces")); assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces")); assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales")); @@ -482,7 +578,54 @@ fn test_register_routes_cover_representative_admin_paths() { Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/metadata"), ); + assert_route( + &router, + Method::GET, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/metadata-location"), + ); + assert_route( + &router, + Method::PUT, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/metadata-location"), + ); + assert_route( + &router, + Method::PUT, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/config"), + ); + assert_route( + &router, + Method::GET, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/config"), + ); + assert_route( + &router, + Method::GET, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1"), + ); + assert_route( + &router, + Method::GET, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/export"), + ); + assert_route( + &router, + Method::POST, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/import"), + ); + assert_route( + &router, + Method::GET, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/diagnostics"), + ); + assert_route( + &router, + Method::POST, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/rollback"), + ); assert_route(&router, Method::GET, &compat_table_catalog_path("/config")); + assert_route(&router, Method::PUT, &compat_table_catalog_path("/buckets/analytics")); + assert_route(&router, Method::GET, &compat_table_catalog_path("/buckets/analytics")); assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces")); assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces")); assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales")); @@ -510,6 +653,51 @@ fn test_register_routes_cover_representative_admin_paths() { Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/metadata"), ); + assert_route( + &router, + Method::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/metadata-location"), + ); + assert_route( + &router, + Method::PUT, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/metadata-location"), + ); + assert_route( + &router, + Method::PUT, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/config"), + ); + assert_route( + &router, + Method::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/config"), + ); + assert_route( + &router, + Method::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/jobs/job-1"), + ); + assert_route( + &router, + Method::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/export"), + ); + assert_route( + &router, + Method::POST, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/import"), + ); + assert_route( + &router, + Method::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/diagnostics"), + ); + assert_route( + &router, + Method::POST, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/catalog/rollback"), + ); assert_route(&router, Method::POST, &admin_path("/v3/service")); assert_route(&router, Method::GET, &admin_path("/v3/info")); diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index 2b7dcefd0..600cb6e5c 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -54,6 +54,7 @@ pub(crate) const TABLE_NAMESPACE_MARKER_VERSION: u16 = 1; pub(crate) const TABLE_RESOURCE_MARKER_VERSION: u16 = 1; pub(crate) const TABLE_METADATA_POINTER_VERSION: u16 = 1; pub(crate) const TABLE_CATALOG_ENTRY_VERSION: u16 = 1; +pub(crate) const TABLE_MAINTENANCE_CONFIG_VERSION: u16 = 1; pub(crate) const TABLE_METADATA_FILE_NAME_MAX_LEN: usize = 128; pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX; const WAREHOUSE_ROOT: &str = "warehouses"; @@ -71,6 +72,9 @@ const INTERNAL_CATALOG_ROOT: &str = BUCKET_TABLE_CATALOG_META_PREFIX; const TABLE_BUCKET_ROOT: &str = BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX; const COMMIT_LOG_ROOT: &str = "commits"; const COMMIT_IDEMPOTENCY_ROOT: &str = "commit-idempotency"; +const MAINTENANCE_ROOT: &str = "maintenance"; +const MAINTENANCE_CONFIG_FILE: &str = "config.json"; +const MAINTENANCE_JOB_ROOT: &str = "jobs"; const TABLE_CATALOG_LIST_MAX_KEYS: i32 = 1000; const TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS: i64 = 15 * 60; @@ -245,12 +249,36 @@ pub(crate) struct TableCommitResult { pub commit_log: CommitLogEntry, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct TableMaintenanceConfig { + pub version: u16, + #[serde(rename = "retain-recent-metadata-files")] + pub retain_recent_metadata_files: usize, + #[serde(rename = "delete-enabled")] + pub delete_enabled: bool, + #[serde(rename = "background-enabled")] + pub background_enabled: bool, +} + +impl Default for TableMaintenanceConfig { + fn default() -> Self { + Self { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 0, + delete_enabled: false, + background_enabled: false, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct TableMetadataMaintenanceJob { pub job_id: String, pub table_bucket: String, pub namespace: String, pub table: String, + pub table_id: String, pub current_metadata_location: String, pub current_generation: u64, pub retain_recent_metadata_files: usize, @@ -258,7 +286,7 @@ pub(crate) struct TableMetadataMaintenanceJob { pub cleanup_watermark_unix_seconds: i64, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub(crate) struct TableMetadataMaintenanceReport { pub job: TableMetadataMaintenanceJob, pub current_metadata_location: String, @@ -438,6 +466,38 @@ impl TableCatalogObjectPaths { ) } + pub fn table_maintenance_config_path( + &self, + table_bucket: &str, + namespace: &Namespace, + table: &IdentifierSegment, + table_id: &str, + ) -> String { + format!( + "{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_CONFIG_FILE}", + self.table_entries_prefix(table_bucket, namespace), + table.as_str(), + table_catalog_path_hash(table_id) + ) + } + + pub fn table_maintenance_job_path( + &self, + table_bucket: &str, + namespace: &Namespace, + table: &IdentifierSegment, + table_id: &str, + job_id: &str, + ) -> String { + format!( + "{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_JOB_ROOT}/{}.json", + self.table_entries_prefix(table_bucket, namespace), + table.as_str(), + table_catalog_path_hash(table_id), + table_catalog_path_hash(job_id) + ) + } + pub fn commit_log_entry_path(&self, table_bucket: &str, table_id: &str, commit_id: &str) -> String { format!( "{}{}/{}/{}.json", @@ -576,6 +636,108 @@ where Ok(()) } + pub(crate) async fn get_table_maintenance_config( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + ) -> TableCatalogStoreResult { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &table_path).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + + let config_path = self + .paths + .table_maintenance_config_path(table_bucket, &namespace, &table, &entry.table_id); + self.read_entry::(self.catalog_bucket(), &config_path) + .await + .map(|entry| entry.map(|(config, _)| config).unwrap_or_default()) + } + + pub(crate) async fn put_table_maintenance_config( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + config: TableMaintenanceConfig, + ) -> TableCatalogStoreResult { + validate_table_maintenance_config_version(config.version)?; + if config.background_enabled { + return Err(TableCatalogStoreError::Invalid( + "background table maintenance is not supported".to_string(), + )); + } + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &table_path).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + + let config_path = self + .paths + .table_maintenance_config_path(table_bucket, &namespace, &table, &entry.table_id); + self.write_entry(self.catalog_bucket(), &config_path, &config, TableCatalogPutPrecondition::Any) + .await?; + Ok(config) + } + + pub(crate) async fn put_table_metadata_maintenance_report( + &self, + report: &TableMetadataMaintenanceReport, + ) -> TableCatalogStoreResult<()> { + let namespace = parse_namespace_for_store(&report.job.namespace)?; + let table = parse_table_for_store(&report.job.table)?; + let job_path = self.paths.table_maintenance_job_path( + &report.job.table_bucket, + &namespace, + &table, + &report.job.table_id, + &report.job.job_id, + ); + self.write_entry(self.catalog_bucket(), &job_path, report, TableCatalogPutPrecondition::Any) + .await + } + + pub(crate) async fn get_table_metadata_maintenance_report( + &self, + table_bucket: &str, + namespace: &str, + table: &str, + job_id: &str, + ) -> TableCatalogStoreResult> { + let namespace = parse_namespace_for_store(namespace)?; + let table = parse_table_for_store(table)?; + let table_path = self.paths.table_entry_path(table_bucket, &namespace, &table); + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &table_path).await? else { + return Err(TableCatalogStoreError::NotFound(format!( + "table {}/{}/{}", + table_bucket, + namespace.public_name(), + table.as_str() + ))); + }; + let job_path = self + .paths + .table_maintenance_job_path(table_bucket, &namespace, &table, &entry.table_id, job_id); + self.read_entry::(self.catalog_bucket(), &job_path) + .await + .map(|entry| entry.map(|(report, _)| report)) + } + pub(crate) async fn export_table_catalog_entry( &self, table_bucket: &str, @@ -759,6 +921,7 @@ where table_bucket: table_bucket.to_string(), namespace: namespace.public_name(), table: table.as_str().to_string(), + table_id: entry.table_id, current_metadata_location: current_metadata_location.clone(), current_generation: entry.generation, retain_recent_metadata_files, @@ -1365,6 +1528,15 @@ fn validate_catalog_entry_version(kind: &str, version: u16) -> TableCatalogStore Ok(()) } +fn validate_table_maintenance_config_version(version: u16) -> TableCatalogStoreResult<()> { + if version != TABLE_MAINTENANCE_CONFIG_VERSION { + return Err(TableCatalogStoreError::Invalid( + "unsupported table maintenance config entry version".to_string(), + )); + } + Ok(()) +} + fn commit_log_matches_request(commit_log: &CommitLogEntry, request: &TableCommitRequest, table_id: &str) -> bool { commit_log.version == TABLE_CATALOG_ENTRY_VERSION && commit_log.commit_id == request.commit_id @@ -2108,8 +2280,11 @@ mod tests { let commit_path = paths.commit_log_entry_path("table/../bucket", "table/../id", "commit/%2f\nid"); let idempotency_path = paths.commit_idempotency_entry_path("table/../bucket", "table/../id", "client/%2f\nrequest"); + let maintenance_config_path = paths.table_maintenance_config_path("table/../bucket", &namespace, &table, "table/../id"); + let maintenance_job_path = + paths.table_maintenance_job_path("table/../bucket", &namespace, &table, "table/../id", "job/%2f\nid"); - for path in [commit_path, idempotency_path] { + for path in [commit_path, idempotency_path, maintenance_config_path, maintenance_job_path] { assert!(path.starts_with("s3tables/catalog/table-buckets/")); assert!(path.ends_with(".json")); assert!(!path.contains("..")); @@ -2400,6 +2575,7 @@ mod tests { assert_eq!(report.job.table_bucket, bucket); assert_eq!(report.job.namespace, "sales"); assert_eq!(report.job.table, "orders"); + assert_eq!(report.job.table_id, "table-id"); assert_eq!(report.job.current_generation, 1); assert_eq!(report.job.safety_window_seconds, TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS); assert!(!report.job.job_id.is_empty()); @@ -2408,6 +2584,102 @@ mod tests { assert_eq!(report.deletable_metadata_locations, vec![old]); } + #[tokio::test] + async fn maintenance_state_is_scoped_to_current_table_identity() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + + store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap(); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .unwrap(); + + let mut first_table = test_table_entry(bucket, &namespace, &table, current.clone()); + first_table.table_id = "table-id-1".to_string(); + store.create_table(first_table).await.unwrap(); + store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 7, + delete_enabled: true, + background_enabled: false, + }, + ) + .await + .unwrap(); + backend + .seed_object(bucket, ¤t, br#"{"metadata-log":[]}"#.to_vec()) + .await; + let report = store + .plan_table_metadata_maintenance(bucket, "sales", "orders", 0) + .await + .unwrap(); + store.put_table_metadata_maintenance_report(&report).await.unwrap(); + assert!( + store + .get_table_metadata_maintenance_report(bucket, "sales", "orders", &report.job.job_id) + .await + .unwrap() + .is_some() + ); + + store.drop_table(bucket, "sales", "orders").await.unwrap(); + + let mut second_table = test_table_entry(bucket, &namespace, &table, current); + second_table.table_id = "table-id-2".to_string(); + store.create_table(second_table).await.unwrap(); + + assert_eq!( + store.get_table_maintenance_config(bucket, "sales", "orders").await.unwrap(), + TableMaintenanceConfig::default() + ); + assert!( + store + .get_table_metadata_maintenance_report(bucket, "sales", "orders", &report.job.job_id) + .await + .unwrap() + .is_none() + ); + } + + #[tokio::test] + async fn maintenance_config_rejects_unsupported_config_version() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").unwrap(); + let table = IdentifierSegment::parse("orders").unwrap(); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; + + let err = store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION.saturating_add(1), + retain_recent_metadata_files: 1, + delete_enabled: false, + background_enabled: false, + }, + ) + .await + .unwrap_err(); + + assert_matches!(err, TableCatalogStoreError::Invalid(_)); + } + #[tokio::test] async fn maintenance_dry_run_keeps_metadata_log_references() { let backend = TestCatalogObjectBackend::default();